165 lines
6.6 KiB
Python
165 lines
6.6 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy, main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_detail_opens_the_newest_conversation_page_in_chronological_order(monkeypatch):
|
|
requested_pages = []
|
|
|
|
def comment(comment_id):
|
|
return {
|
|
"id": comment_id,
|
|
"user": {"login": f"user-{comment_id}"},
|
|
"body": f"message {comment_id}",
|
|
"created_at": f"2026-08-06T12:{comment_id:02d}:00Z",
|
|
"html_url": f"https://forge.example/stackchain/api/issues/7#issuecomment-{comment_id}",
|
|
}
|
|
|
|
def handler(request):
|
|
if request.url.path.endswith("/notifications/threads/42"):
|
|
return httpx.Response(200, json={
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"subject": {
|
|
"type": "Issue", "title": "Retry failed deploy", "state": "open",
|
|
"url": "https://forge.example/api/v1/repos/stackchain/api/issues/7",
|
|
"latest_comment_url": "https://forge.example/api/v1/repos/stackchain/api/issues/comments/47",
|
|
"html_url": "https://forge.example/stackchain/api/issues/7",
|
|
},
|
|
})
|
|
if request.url.path.endswith("/repos/stackchain/api/issues/7/comments"):
|
|
page = int(request.url.params["page"])
|
|
requested_pages.append(page)
|
|
comments = [comment(i) for i in (range(1, 21) if page == 1 else range(41, 48))]
|
|
return httpx.Response(200, json=comments, headers={"X-Total-Count": "47"})
|
|
if request.url.path.endswith("/repos/stackchain/api/issues/7"):
|
|
return httpx.Response(200, json={
|
|
"number": 7,
|
|
"state": "open",
|
|
"assignees": [],
|
|
"body": "Deploy fails after retries.",
|
|
})
|
|
if request.url.path.endswith("/repos/stackchain/api/issues/comments/47"):
|
|
return httpx.Response(200, json=comment(47))
|
|
raise AssertionError(f"unexpected request: {request.url}")
|
|
|
|
monkeypatch.setattr(gitea_proxy, "GITEA_URL", "https://forge.example")
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.notification_detail(42)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requested_pages == [1, 3]
|
|
assert [item["id"] for item in result["conversation"]["comments"]] == list(range(41, 48))
|
|
assert result["conversation"] == {
|
|
"comments": result["conversation"]["comments"],
|
|
"page": 3,
|
|
"older_page": 2,
|
|
"total": 47,
|
|
}
|
|
assert result["issue"] == {
|
|
"number": 7,
|
|
"assignees": [],
|
|
"claimable": True,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_detail_api_is_bounded_and_never_cacheable(monkeypatch):
|
|
calls = []
|
|
|
|
async def detail(thread_id):
|
|
calls.append(thread_id)
|
|
return {
|
|
"id": thread_id,
|
|
"repository": "stackchain/api",
|
|
"title": "Retry failed deploy",
|
|
"subject_type": "Issue",
|
|
"state": "open",
|
|
"url": "https://forge.example/stackchain/api/issues/7#issuecomment-9",
|
|
"subject_body": "Deploy fails after three retries.",
|
|
"latest_comment": {
|
|
"author": "alexander",
|
|
"body": "Logs point to the worker timeout.",
|
|
"created_at": "2026-08-06T12:30:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/7#issuecomment-9",
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(main, "notification_detail", detail, raising=False)
|
|
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/42")
|
|
invalid = await client.get("/api/v1/notifications/0")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["latest_comment"]["author"] == "alexander"
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert invalid.status_code == 422
|
|
assert calls == [42]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_conversation_api_loads_one_bounded_older_page(monkeypatch):
|
|
calls = []
|
|
|
|
async def conversation(thread_id, page, limit):
|
|
calls.append((thread_id, page, limit))
|
|
return {
|
|
"comments": [{"id": 21, "author": "timmy", "body": "Earlier context"}],
|
|
"page": page,
|
|
"older_page": 1,
|
|
"total": 47,
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "notification_conversation_page", conversation, raising=False)
|
|
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/42/conversation?page=2&limit=20")
|
|
invalid = await client.get("/api/v1/notifications/42/conversation?page=0&limit=51")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["comments"][0]["id"] == 21
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert invalid.status_code == 422
|
|
assert calls == [(42, 2, 20)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_detail_timeout_is_sanitized_and_retryable(monkeypatch):
|
|
async def detail(_thread_id):
|
|
await asyncio.sleep(0.05)
|
|
|
|
monkeypatch.setattr(main, "notification_detail", detail)
|
|
monkeypatch.setattr(main, "NOTIFICATION_DETAIL_TIMEOUT_SECONDS", 0.01)
|
|
transport = httpx.ASGITransport(app=main.app, raise_app_exceptions=False)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/notifications/42")
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"error": "Loading the update timed out. Please retry."}
|
|
assert response.headers["retry-after"] == "1"
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_detail_upstream_failure_does_not_leak_exception(monkeypatch):
|
|
async def detail(_thread_id):
|
|
raise httpx.HTTPError("token=secret upstream exploded")
|
|
|
|
monkeypatch.setattr(main, "notification_detail", detail)
|
|
transport = httpx.ASGITransport(app=main.app, raise_app_exceptions=False)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/notifications/42")
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {
|
|
"error": "The update is temporarily unavailable. Please retry."
|
|
}
|
|
assert "secret" not in response.text
|
|
assert response.headers["retry-after"] == "1"
|