520 lines
17 KiB
Python
520 lines
17 KiB
Python
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_unread_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_unread(42)
|
|
|
|
assert calls == [
|
|
("/api/v1/notifications/threads/42?to-status=unread", {"Authorization": "token test"})
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_acknowledge_notification_resolves_latest_comment_and_adds_one_reaction(monkeypatch):
|
|
calls = []
|
|
|
|
async def fake_fetch(path):
|
|
calls.append(("fetch", path))
|
|
if path == "notifications/threads/42":
|
|
return {
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"subject": {
|
|
"type": "Issue",
|
|
"url": "https://forge.example/api/v1/repos/stackchain/api/issues/7",
|
|
"latest_comment_url": (
|
|
"https://forge.example/api/v1/repos/stackchain/api/issues/comments/91"
|
|
),
|
|
},
|
|
}
|
|
if path == "user":
|
|
return {"login": "timmy"}
|
|
raise AssertionError(path)
|
|
|
|
class Response:
|
|
def __init__(self, payload=None):
|
|
self._payload = payload
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
class Client:
|
|
async def get(self, path, headers):
|
|
calls.append(("get", path))
|
|
return Response([])
|
|
|
|
async def post(self, path, headers, json):
|
|
calls.append(("post", path, json))
|
|
return Response({"content": "+1"})
|
|
|
|
async def patch(self, path, headers):
|
|
calls.append(("patch", path))
|
|
return Response()
|
|
|
|
monkeypatch.setattr(gitea_proxy, "GITEA_URL", "https://forge.example")
|
|
monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
|
|
monkeypatch.setattr(gitea_proxy, "_get_client", lambda: Client())
|
|
monkeypatch.setattr(gitea_proxy, "_auth", lambda: {"Authorization": "token test"})
|
|
|
|
result = await gitea_proxy.acknowledge_notification(42)
|
|
|
|
reaction_path = "/api/v1/repos/stackchain/api/issues/comments/91/reactions"
|
|
assert result == {"id": 42, "reaction": "created", "status": "read"}
|
|
assert calls == [
|
|
("fetch", "notifications/threads/42"),
|
|
("fetch", "user"),
|
|
("get", reaction_path),
|
|
("post", reaction_path, {"content": "+1"}),
|
|
("patch", "/api/v1/notifications/threads/42?to-status=read"),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_acknowledge_notification_reuses_existing_operator_reaction_on_retry(monkeypatch):
|
|
posts = []
|
|
|
|
async def fake_fetch(path):
|
|
if path == "user":
|
|
return {"login": "timmy"}
|
|
return {
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"subject": {
|
|
"type": "Pull",
|
|
"url": "https://forge.example/api/v1/repos/stackchain/api/pulls/7",
|
|
"latest_comment_url": (
|
|
"https://forge.example/api/v1/repos/stackchain/api/issues/comments/91"
|
|
),
|
|
},
|
|
}
|
|
|
|
class Response:
|
|
def __init__(self, payload=None):
|
|
self.payload = payload
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return self.payload
|
|
|
|
class Client:
|
|
async def get(self, path, headers):
|
|
return Response([{"content": "+1", "user": {"login": "timmy"}}])
|
|
|
|
async def post(self, path, headers, json):
|
|
posts.append((path, json))
|
|
return Response()
|
|
|
|
async def patch(self, path, headers):
|
|
return Response()
|
|
|
|
monkeypatch.setattr(gitea_proxy, "GITEA_URL", "https://forge.example")
|
|
monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
|
|
monkeypatch.setattr(gitea_proxy, "_get_client", lambda: Client())
|
|
|
|
result = await gitea_proxy.acknowledge_notification(42)
|
|
|
|
assert result == {"id": 42, "reaction": "existing", "status": "read"}
|
|
assert posts == []
|
|
|
|
|
|
@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_mark_notification_unread_api_is_bounded_and_never_cacheable(monkeypatch):
|
|
restored = []
|
|
|
|
async def mark_unread(thread_id):
|
|
restored.append(thread_id)
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "mark_notification_unread", mark_unread, raising=False)
|
|
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/unread")
|
|
invalid = await client.patch("/api/v1/notifications/0/unread")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"id": 42, "status": "unread"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert invalid.status_code == 422
|
|
assert restored == [42]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_mute_notification_then_marks_read_and_reports_partial_success(monkeypatch):
|
|
calls = []
|
|
|
|
async def mute(thread_id):
|
|
calls.append(("mute", thread_id))
|
|
return {"id": thread_id, "muted": True}
|
|
|
|
async def fail_read(thread_id):
|
|
calls.append(("read", thread_id))
|
|
raise httpx.HTTPError("unavailable")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "mute_notification", mute, raising=False)
|
|
monkeypatch.setattr(main, "mark_notification_read", fail_read)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post("/api/v1/notifications/42/mute")
|
|
|
|
assert response.status_code == 409
|
|
assert response.json() == {
|
|
"id": 42,
|
|
"muted": True,
|
|
"status": "unread",
|
|
"error": "Future updates are muted; current item is still unread. Retry mark read & next.",
|
|
}
|
|
assert calls == [("mute", 42), ("read", 42)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_acknowledge_notification_api_confirms_reaction_and_removes_snapshot_item(monkeypatch):
|
|
calls = []
|
|
|
|
async def acknowledge(thread_id):
|
|
calls.append(thread_id)
|
|
return {"id": thread_id, "reaction": "created", "status": "read"}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "acknowledge_notification", acknowledge, raising=False)
|
|
monkeypatch.setattr(
|
|
main,
|
|
"_live_snapshot_value",
|
|
{"notifications": [{"id": 42}, {"id": 43}]},
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post("/api/v1/notifications/42/acknowledge")
|
|
invalid = await client.post("/api/v1/notifications/0/acknowledge")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"id": 42, "reaction": "created", "status": "read"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert invalid.status_code == 422
|
|
assert calls == [42]
|
|
assert main._live_snapshot_value == {"notifications": [{"id": 43}]}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_snapshot_maintenance_does_not_block_the_event_loop(monkeypatch):
|
|
async def mark(_thread_id):
|
|
return None
|
|
|
|
calls = []
|
|
|
|
class BlockingStore:
|
|
def remove_notification(self, thread_id):
|
|
calls.append([thread_id])
|
|
time.sleep(0.08)
|
|
|
|
def remove_notifications(self, thread_ids):
|
|
calls.append(list(thread_ids))
|
|
time.sleep(0.08)
|
|
|
|
monkeypatch.setattr(main, "mark_notification_read", mark)
|
|
monkeypatch.setattr(main, "_live_snapshot_store", BlockingStore())
|
|
monkeypatch.setattr(main, "_live_snapshot_value", None)
|
|
started_at = asyncio.get_running_loop().time()
|
|
|
|
request = asyncio.create_task(main.read_notification(42))
|
|
await asyncio.sleep(0.01)
|
|
heartbeat_delay = asyncio.get_running_loop().time() - started_at
|
|
response = await request
|
|
|
|
assert heartbeat_delay < 0.05
|
|
assert response.status_code == 200
|
|
assert calls == [[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_updates_shared_snapshot_once_for_successes(monkeypatch):
|
|
async def mark(thread_id):
|
|
if thread_id == 43:
|
|
raise httpx.HTTPError("upstream unavailable")
|
|
|
|
calls = []
|
|
|
|
class RecordingStore:
|
|
def remove_notifications(self, thread_ids):
|
|
calls.append(tuple(sorted(thread_ids)))
|
|
|
|
monkeypatch.setattr(main, "mark_notification_read", mark)
|
|
monkeypatch.setattr(main, "_live_snapshot_store", RecordingStore())
|
|
monkeypatch.setattr(
|
|
main,
|
|
"_live_snapshot_value",
|
|
{"notifications": [{"id": 42}, {"id": 43}, {"id": 44}]},
|
|
)
|
|
|
|
response = await main.read_notifications(main.NotificationReadBatch(ids=[42, 43, 44]))
|
|
|
|
assert response.status_code == 200
|
|
assert calls == [(42, 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)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await asyncio.wait_for(
|
|
client.patch(
|
|
"/api/v1/notifications/read", json={"ids": list(range(1, 11))}
|
|
),
|
|
timeout=0.5,
|
|
)
|
|
|
|
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}]
|