state-hub/tests/test_c25_message_pagination.py
tegwick 3b9a5c58f5 Paginate C-25 owner-return reply collection (STATE-WP-0094-T03)
GET /messages/ accepts offset with stable created_at,id ordering; C-25 pages
every reply per counterpart, dedups by message id and warns explicitly when
collection is incomplete instead of capping at 200.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 671493@bnt-lap001
Assistant-Session: 60d7b11f-78f9-4731-9367-918511b2762d
2026-10-04 12:54:57 +02:00

94 lines
3.6 KiB
Python

"""C-25 owner-return collection pages past 200 replies (STATE-WP-0094-T03)."""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
from consistency_check import ConsistencyReport, _check_blocked_unblock_sweep # noqa: E402
BODY = (
"```task\nid: DEMO-WP-0001-T01\nstatus: wait\nblocked_on: message-from:provider\n```\n\n"
"2026-10-01: waiting on provider."
)
def _replies(total: int) -> list[dict]:
# Newest first, as the hub orders them.
return [
{"id": f"reply-{i:04d}", "from_agent": "provider",
"created_at": f"2026-10-03T{(i // 60) % 24:02d}:{i % 60:02d}:00Z", "read_at": None}
for i in range(total)
][::-1]
def _run(tmp_path, monkeypatch, fake_get):
monkeypatch.setattr("consistency_check._api_get", fake_get)
report = ConsistencyReport(repo_slug="demo", repo_path=str(tmp_path))
_check_blocked_unblock_sweep(
"http://unused", "demo", tmp_path,
[(tmp_path / "plan.md", {"status": "active", "flavor": "implementation"}, BODY)], report,
)
return report
def test_collects_every_reply_across_pages_with_stable_ids(tmp_path, monkeypatch):
replies = _replies(450)
queries = []
def paged(_api_base, _path, params):
queries.append(dict(params))
start = params["offset"]
return replies[start:start + params["limit"]]
report = _run(tmp_path, monkeypatch, paged)
assert [q["offset"] for q in queries] == [0, 200, 400]
assert len(report.issues) == 1
ids = report.issues[0]._fix_context["message_ids"]
assert len(ids) == len(set(ids)) == 450
assert ids == sorted(ids) # chronological, so the dedup key is stable
assert report.issues[0]._fix_context["task_record_id"] == "DEMO-WP-0001-T01"
# Overlapping pages (new mail shifting offsets) are deduplicated by id.
shifted = [{"id": "reply-new", "from_agent": "provider",
"created_at": "2026-10-03T23:59:00Z", "read_at": None}] + replies
def shifting(_api_base, _path, params):
source = replies if params["offset"] == 0 else shifted
return source[params["offset"]:params["offset"] + params["limit"]]
again = _run(tmp_path, monkeypatch, shifting)
assert len(again.issues[0]._fix_context["message_ids"]) == 450
def test_hub_ignoring_offset_is_reported_not_silently_capped(tmp_path, monkeypatch):
replies = _replies(200)
report = _run(tmp_path, monkeypatch, lambda _a, _p, params: replies)
checks = [(issue.check_id, "incomplete" in issue.message) for issue in report.issues]
assert ("C-25", True) in checks
owner = [i for i in report.issues if i._fix_context.get("task_record_id")]
assert len(owner[0]._fix_context["message_ids"]) == 200
@pytest.mark.asyncio
async def test_messages_api_offset_pages_are_disjoint_and_ordered(client):
for index in range(5):
response = await client.post("/messages/", json={
"from_agent": "provider", "to_agent": "demo",
"subject": f"reply {index}", "body": "done",
})
assert response.status_code == 201, response.text
pages = []
for offset in (0, 2, 4):
response = await client.get("/messages/", params={
"to_agent": "demo", "from_agent": "provider", "limit": 2, "offset": offset})
assert response.status_code == 200, response.text
pages.append([row["id"] for row in response.json()])
flat = [item for page in pages for item in page]
assert [len(page) for page in pages] == [2, 2, 1]
assert len(set(flat)) == 5
bad = await client.get("/messages/", params={"offset": -1})
assert bad.status_code == 422