test: cover hidden snapshot pin quota route

This commit is contained in:
ai-ag2026 2026-05-30 16:31:24 +02:00
parent 09d2f4be2f
commit 4639d355ee

View file

@ -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