109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy, main
|
|
|
|
|
|
@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_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():
|
|
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()
|
|
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}]
|