Bound assigned-issue reminder snapshots

tests/test_deadline_reminders.py

@@ -270,23 +270,92 @@ def test_existing_deadline_preferences_migrate_to_two_day_horizon(tmp_path):
 @pytest.mark.anyio
 async def test_assigned_deadline_snapshot_is_pagination_complete(monkeypatch):
     pages = {
-        1: {"items": [{"id": 1}], "has_more": True},
-        2: {"items": [{"id": 2}], "has_more": False},
+        1: {"items": [{"id": 1}], "total": 2, "has_more": True},
+        2: {"items": [{"id": 2}], "total": 2, "has_more": False},
     }
 
     async def work_page(stream, page, limit):
         assert stream == "issue"
-        assert limit == 50
+        assert limit == 1
         return pages[page]
 
     monkeypatch.setattr(gitea_proxy, "work_page", work_page)
 
-    assert await gitea_proxy.assigned_issue_snapshot() == {
+    assert await gitea_proxy.assigned_issue_snapshot(limit=1) == {
         "items": [{"id": 1}, {"id": 2}],
         "complete": True,
     }
 
 
+@pytest.mark.anyio
+async def test_assigned_deadline_snapshot_fetches_remaining_pages_concurrently(monkeypatch):
+    active = 0
+    peak = 0
+    remaining_started = asyncio.Event()
+
+    async def work_page(stream, page, limit):
+        nonlocal active, peak
+        assert stream == "issue"
+        assert limit == 1
+        if page == 1:
+            return {"items": [{"id": 1}], "total": 5, "has_more": True}
+        active += 1
+        peak = max(peak, active)
+        if active == 2:
+            remaining_started.set()
+        await asyncio.wait_for(remaining_started.wait(), timeout=0.2)
+        await asyncio.sleep(0)
+        active -= 1
+        return {"items": [{"id": page}], "total": 5, "has_more": page < 5}
+
+    monkeypatch.setattr(gitea_proxy, "work_page", work_page)
+
+    assert await gitea_proxy.assigned_issue_snapshot(limit=1, max_concurrency=2) == {
+        "items": [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}, {"id": 5}],
+        "complete": True,
+    }
+    assert peak == 2
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "remaining",
+    [
+        {"items": [{"id": 2}], "total": 3, "has_more": True},
+        {"items": [{"id": 1}], "total": 2, "has_more": False},
+    ],
+)
+async def test_assigned_deadline_snapshot_rejects_changed_or_duplicate_pages(
+    monkeypatch, remaining
+):
+    async def work_page(_stream, page, _limit):
+        if page == 1:
+            return {"items": [{"id": 1}], "total": 2, "has_more": True}
+        return remaining
+
+    monkeypatch.setattr(gitea_proxy, "work_page", work_page)
+
+    with pytest.raises(ValueError, match="incomplete|changed"):
+        await gitea_proxy.assigned_issue_snapshot(limit=1)
+
+
+@pytest.mark.anyio
+async def test_assigned_deadline_snapshot_enforces_aggregate_deadline(monkeypatch):
+    cancelled = asyncio.Event()
+
+    async def work_page(_stream, _page, _limit):
+        try:
+            await asyncio.sleep(60)
+        finally:
+            cancelled.set()
+
+    monkeypatch.setattr(gitea_proxy, "work_page", work_page)
+
+    with pytest.raises(TimeoutError):
+        await gitea_proxy.assigned_issue_snapshot(deadline_seconds=0.01)
+    assert cancelled.is_set()
+
+
 @pytest.mark.anyio
 async def test_competing_workers_send_one_deadline_digest(tmp_path):
     path = tmp_path / "push.sqlite3"