From 4639d355eed12f8a979fa26b33248262fd3e3866 Mon Sep 17 00:00:00 2001 From: ai-ag2026 Date: Sat, 30 May 2026 16:31:24 +0200 Subject: [PATCH] test: cover hidden snapshot pin quota route --- tests/test_issue2508_session_pin_cap.py | 67 ++++++++++++++++++++++++- 1 file changed, 66 insertions(+), 1 deletion(-) diff --git a/tests/test_issue2508_session_pin_cap.py b/tests/test_issue2508_session_pin_cap.py index 01d533bc..b49855c0 100644 --- a/tests/test_issue2508_session_pin_cap.py +++ b/tests/test_issue2508_session_pin_cap.py @@ -2,11 +2,12 @@ import json import pathlib +import time from types import SimpleNamespace import urllib.error import urllib.request -from tests._pytest_port import BASE +from tests._pytest_port import BASE, TEST_STATE_DIR ROOT = pathlib.Path(__file__).resolve().parent.parent @@ -42,6 +43,41 @@ def make_session(created): return sid + +def inject_hidden_pinned_snapshot(sid="hidden-pinned-snapshot"): + """Add a persisted legacy hidden snapshot without touching server memory.""" + sessions_dir = TEST_STATE_DIR / "sessions" + sessions_dir.mkdir(parents=True, exist_ok=True) + now = time.time() + row = { + "session_id": sid, + "title": "Hidden pinned snapshot", + "workspace": str(TEST_STATE_DIR / "test-workspace"), + "model": "test/pin-cap", + "created_at": now, + "updated_at": now, + "last_message_at": now, + "message_count": 1, + "messages": [{"role": "user", "content": "legacy hidden snapshot"}], + "tool_calls": [], + "pinned": True, + "archived": False, + "pre_compression_snapshot": True, + "_show_pre_compression_snapshot": False, + } + (sessions_dir / f"{sid}.json").write_text(json.dumps(row), encoding="utf-8") + index_path = sessions_dir / "_index.json" + try: + index = json.loads(index_path.read_text(encoding="utf-8")) + except (FileNotFoundError, json.JSONDecodeError): + index = [] + compact = {k: v for k, v in row.items() if k not in {"messages", "tool_calls"}} + index = [item for item in index if item.get("session_id") != sid] + index.append(compact) + index_path.write_text(json.dumps(index), encoding="utf-8") + return sid + + def test_session_pin_endpoint_caps_pinned_sessions_at_three(): created = [] try: @@ -68,6 +104,35 @@ def test_session_pin_endpoint_caps_pinned_sessions_at_three(): post("/api/session/delete", {"session_id": sid}) +def test_session_pin_endpoint_ignores_hidden_snapshot_when_enforcing_cap(): + created = [] + hidden_sid = "hidden-pinned-snapshot-quota-route" + try: + hidden = inject_hidden_pinned_snapshot(hidden_sid) + pinned = [make_session(created) for _ in range(2)] + for sid in pinned: + d, status = post("/api/session/pin", {"session_id": sid, "pinned": True}) + assert status == 200 + assert d["session"]["pinned"] is True + + third_visible = make_session(created) + d, status = post("/api/session/pin", {"session_id": third_visible, "pinned": True}) + assert status == 200, d + assert d["session"]["pinned"] is True + assert hidden not in {third_visible, *pinned} + finally: + for sid in created: + post("/api/session/delete", {"session_id": sid}) + (TEST_STATE_DIR / "sessions" / f"{hidden_sid}.json").unlink(missing_ok=True) + index_path = TEST_STATE_DIR / "sessions" / "_index.json" + try: + index = json.loads(index_path.read_text(encoding="utf-8")) + index = [item for item in index if item.get("session_id") != hidden_sid] + index_path.write_text(json.dumps(index), encoding="utf-8") + except (FileNotFoundError, json.JSONDecodeError): + pass + + def test_hidden_pre_compression_snapshot_does_not_count_toward_pin_quota(): from api.routes import _session_counts_toward_pin_quota