import asyncio import time import httpx import pytest from src import gitea_proxy, main from src.live_snapshot_store import LiveSnapshotStore @pytest.fixture(autouse=True) def isolated_live_snapshot_store(tmp_path, monkeypatch): monkeypatch.setattr( main, "_live_snapshot_store", LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: main._live_snapshot_clock()), ) monkeypatch.setattr(main, "_read_notification_ids", set()) @pytest.mark.anyio async def test_mark_notification_read_calls_supported_gitea_thread_endpoint(monkeypatch): calls = [] class Response: def raise_for_status(self): return None class Client: async def patch(self, path, headers): calls.append((path, headers)) return Response() monkeypatch.setattr(gitea_proxy, "_get_client", lambda: Client()) monkeypatch.setattr(gitea_proxy, "_auth", lambda: {"Authorization": "token test"}) await gitea_proxy.mark_notification_read(42) assert calls == [ ("/api/v1/notifications/threads/42?to-status=read", {"Authorization": "token test"}) ] @pytest.mark.anyio async def test_mark_notification_read_api_is_bounded_and_never_cacheable(monkeypatch): marked = [] async def mark(thread_id): marked.append(thread_id) monkeypatch.setattr(main, "mark_notification_read", mark) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.patch("/api/v1/notifications/42/read") invalid = await client.patch("/api/v1/notifications/0/read") assert response.status_code == 200 assert response.json() == {"id": 42, "status": "read"} assert response.headers["cache-control"] == "no-store" assert invalid.status_code == 422 assert marked == [42] @pytest.mark.anyio async def test_bulk_mark_read_reports_partial_progress_and_retains_only_failures(monkeypatch): calls = [] async def mark(thread_id): calls.append(thread_id) if thread_id == 43: raise httpx.HTTPError("upstream unavailable") monkeypatch.setattr(main, "mark_notification_read", mark) monkeypatch.setattr( main, "_live_snapshot_value", {"notifications": [{"id": 42}, {"id": 43}, {"id": 44}]}, ) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.patch( "/api/v1/notifications/read", json={"ids": [42, 43, 42, 44]} ) assert response.status_code == 200 assert response.json() == {"marked": [42, 44], "failed": [43]} assert response.headers["cache-control"] == "no-store" assert sorted(calls) == [42, 43, 44] assert main._live_snapshot_value == {"notifications": [{"id": 43}]} @pytest.mark.anyio async def test_bulk_mark_read_limits_upstream_concurrency(monkeypatch): active = 0 peak_active = 0 started = asyncio.Event() release = asyncio.Event() calls = [] monkeypatch.setattr(main, "_read_notification_ids", set()) monkeypatch.setattr(main, "_live_snapshot_value", None) async def mark(thread_id): nonlocal active, peak_active calls.append(thread_id) active += 1 peak_active = max(peak_active, active) if active == 5: started.set() try: await release.wait() finally: active -= 1 monkeypatch.setattr(main, "mark_notification_read", mark) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: request = asyncio.create_task( client.patch("/api/v1/notifications/read", json={"ids": list(range(1, 51))}) ) await asyncio.wait_for(started.wait(), timeout=1) await asyncio.sleep(0) assert peak_active == 5 assert len(calls) == 5 release.set() response = await request assert response.status_code == 200 assert response.json() == {"marked": list(range(1, 51)), "failed": []} assert sorted(calls) == list(range(1, 51)) @pytest.mark.anyio async def test_bulk_mark_read_cancels_unfinished_work_at_batch_deadline(monkeypatch): active = 0 cancelled = [] async def mark(thread_id): nonlocal active active += 1 try: await asyncio.sleep(0.08) except asyncio.CancelledError: cancelled.append(thread_id) raise finally: active -= 1 monkeypatch.setattr(main, "mark_notification_read", mark) monkeypatch.setattr(main, "BULK_NOTIFICATION_DEADLINE_SECONDS", 0.01, raising=False) monkeypatch.setattr(main, "_read_notification_ids", set()) monkeypatch.setattr(main, "_live_snapshot_value", None) transport = httpx.ASGITransport(app=main.app) started_at = time.monotonic() async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.patch( "/api/v1/notifications/read", json={"ids": list(range(1, 11))} ) elapsed = time.monotonic() - started_at assert elapsed < 0.06 assert response.status_code == 200 assert response.json() == {"marked": [], "failed": list(range(1, 11))} assert response.headers["retry-after"] == "1" assert active == 0 assert sorted(cancelled) == list(range(1, 6)) @pytest.mark.anyio async def test_bulk_mark_read_rejects_empty_invalid_and_oversized_batches(monkeypatch): marked = [] async def mark(thread_id): marked.append(thread_id) monkeypatch.setattr(main, "mark_notification_read", mark) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: empty = await client.patch("/api/v1/notifications/read", json={"ids": []}) invalid = await client.patch( "/api/v1/notifications/read", json={"ids": [42, 0]} ) oversized = await client.patch( "/api/v1/notifications/read", json={"ids": list(range(1, 52))} ) assert [empty.status_code, invalid.status_code, oversized.status_code] == [ 422, 422, 422, ] assert all(response.headers["cache-control"] == "no-store" for response in [empty, invalid, oversized]) assert marked == [] @pytest.mark.anyio async def test_mark_notification_read_removes_thread_from_retained_live_snapshot(monkeypatch): async def mark(_thread_id): return None monkeypatch.setattr(main, "mark_notification_read", mark) monkeypatch.setattr( main, "_live_snapshot_value", { "context": {}, "events": [], "notifications": [{"id": 42}, {"id": 43}], "sections": {"notifications": "fresh"}, }, ) response = await main.read_notification(42) assert response.status_code == 200 assert main._live_snapshot_value is not None assert main._live_snapshot_value["notifications"] == [{"id": 43}] @pytest.mark.anyio async def test_inflight_refresh_cannot_restore_a_notification_marked_read(monkeypatch): refresh_started = asyncio.Event() release_refresh = asyncio.Event() async def mark(_thread_id): return None async def stale_upstream_snapshot(_sections): refresh_started.set() await release_refresh.wait() return { "context": {}, "events": [], "notifications": [{"id": 42}, {"id": 43}], "sections": {"notifications": "fresh"}, } monkeypatch.setattr(main, "mark_notification_read", mark) monkeypatch.setattr( main, "_build_live_snapshot_before_deadline", stale_upstream_snapshot ) monkeypatch.setattr(main, "_live_snapshot_task", None) monkeypatch.setattr(main, "_live_snapshot_value", {"notifications": [{"id": 42}]}) refresh = main._start_live_snapshot_refresh(set(main.LIVE_SNAPSHOT_SECTIONS)) await refresh_started.wait() await main.read_notification(42) release_refresh.set() await refresh assert main._live_snapshot_value is not None assert main._live_snapshot_value["notifications"] == [{"id": 43}]