86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_page_endpoint_loads_only_requested_page_and_is_not_cacheable(monkeypatch):
|
|
requested = []
|
|
|
|
async def page_loader(page):
|
|
requested.append(page)
|
|
return {
|
|
"items": [{"id": 51, "title": "Older update"}],
|
|
"page": page,
|
|
"total": 125,
|
|
"has_more": True,
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "notification_page", page_loader)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/notifications?page=2")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {
|
|
"items": [{"id": 51, "title": "Older update"}],
|
|
"page": 2,
|
|
"total": 125,
|
|
"has_more": True,
|
|
}
|
|
assert requested == [2]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_snapshot_endpoint_returns_one_complete_atomic_no_store_result(monkeypatch):
|
|
requested = []
|
|
|
|
async def snapshot_loader(*, deadline_seconds):
|
|
requested.append(deadline_seconds)
|
|
return {
|
|
"items": [
|
|
{"id": 1, "title": "Newest update"},
|
|
{"id": 51, "title": "Older update"},
|
|
],
|
|
"total": 2,
|
|
"complete": True,
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", snapshot_loader)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/notifications/snapshot")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {
|
|
"items": [
|
|
{"id": 1, "title": "Newest update"},
|
|
{"id": 51, "title": "Older update"},
|
|
],
|
|
"total": 2,
|
|
"complete": True,
|
|
}
|
|
assert requested == [main.NOTIFICATION_PAGE_TIMEOUT_SECONDS]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("failure", [TimeoutError(), ValueError("pagination changed")])
|
|
async def test_notification_snapshot_endpoint_is_bounded_and_retryable(monkeypatch, failure):
|
|
async def snapshot_loader(*, deadline_seconds):
|
|
raise failure
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", snapshot_loader)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/notifications/snapshot")
|
|
|
|
assert response.status_code == 503
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.headers["retry-after"] == "1"
|
|
assert response.json() == {
|
|
"error": "Complete unread updates are temporarily unavailable. Please retry."
|
|
}
|