203 lines
7.0 KiB
Python
203 lines
7.0 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_page_preserves_upstream_total_without_loading_other_pages():
|
|
requests = []
|
|
|
|
def upstream(request):
|
|
requests.append(str(request.url))
|
|
return httpx.Response(
|
|
200,
|
|
headers={"X-Total-Count": "125"},
|
|
json=[{
|
|
"id": 42,
|
|
"unread": True,
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"subject": {
|
|
"title": "Retry failed deploy",
|
|
"type": "Issue",
|
|
"html_url": "https://forge.example/stackchain/api/issues/7",
|
|
},
|
|
}],
|
|
)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.notification_page(page=2)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests == [
|
|
"http://127.0.0.1:3000/api/v1/notifications?status-types=unread&limit=50&page=2"
|
|
]
|
|
assert result["page"] == 2
|
|
assert result["total"] == 125
|
|
assert result["has_more"] is True
|
|
assert [item["id"] for item in result["items"]] == [42]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_unread_notifications_are_normalized_for_mobile_handoff():
|
|
result = gitea_proxy._normalize_notifications(
|
|
[
|
|
{
|
|
"id": 42,
|
|
"unread": True,
|
|
"updated_at": "2026-08-06T12:30:00Z",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"subject": {
|
|
"title": "Retry failed deploy",
|
|
"type": "Issue",
|
|
"state": "open",
|
|
"html_url": "https://forge.example/stackchain/api/issues/7",
|
|
"latest_comment_html_url": "https://forge.example/stackchain/api/issues/7#issuecomment-9",
|
|
},
|
|
},
|
|
{"id": 43, "repository": None, "subject": None},
|
|
{
|
|
"id": 44,
|
|
"repository": {"full_name": "stackchain/web"},
|
|
"subject": {"title": "Unsafe", "html_url": "javascript:alert(1)"},
|
|
},
|
|
"malformed",
|
|
]
|
|
)
|
|
assert result == [
|
|
{
|
|
"id": 42,
|
|
"unread": True,
|
|
"updated_at": "2026-08-06T12:30:00Z",
|
|
"repository": "stackchain/api",
|
|
"number": 7,
|
|
"title": "Retry failed deploy",
|
|
"subject_type": "Issue",
|
|
"state": "open",
|
|
"url": "https://forge.example/stackchain/api/issues/7#issuecomment-9",
|
|
"subject_url": "https://forge.example/stackchain/api/issues/7",
|
|
},
|
|
{
|
|
"id": 43,
|
|
"unread": False,
|
|
"updated_at": "",
|
|
"repository": "",
|
|
"number": None,
|
|
"title": "Untitled update",
|
|
"subject_type": "Update",
|
|
"state": "",
|
|
"url": "",
|
|
"subject_url": "",
|
|
},
|
|
{
|
|
"id": 44,
|
|
"unread": False,
|
|
"updated_at": "",
|
|
"repository": "stackchain/web",
|
|
"number": None,
|
|
"title": "Unsafe",
|
|
"subject_type": "Update",
|
|
"state": "",
|
|
"url": "",
|
|
"subject_url": "",
|
|
},
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_collection_rejects_non_list_payload():
|
|
with pytest.raises(ValueError, match="notification response was not a list"):
|
|
gitea_proxy._normalize_notifications({"message": "unexpected"})
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_detail_loads_subject_and_latest_comment_for_inbox_reader():
|
|
requests = []
|
|
|
|
def upstream(request):
|
|
requests.append(str(request.url))
|
|
if request.url.path.endswith("/notifications/threads/42"):
|
|
return httpx.Response(200, json={
|
|
"id": 42,
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"subject": {
|
|
"title": "Retry failed deploy",
|
|
"type": "Issue",
|
|
"state": "open",
|
|
"url": "http://127.0.0.1:3000/api/v1/repos/stackchain/api/issues/7",
|
|
"latest_comment_url": "http://127.0.0.1:3000/api/v1/repos/stackchain/api/issues/comments/9",
|
|
"html_url": "https://forge.example/stackchain/api/issues/7",
|
|
"latest_comment_html_url": "https://forge.example/stackchain/api/issues/7#issuecomment-9",
|
|
},
|
|
})
|
|
if request.url.path.endswith("/issues/7"):
|
|
return httpx.Response(200, json={"body": "Deploy fails after **three** retries."})
|
|
if request.url.path.endswith("/issues/comments/9"):
|
|
return httpx.Response(200, json={
|
|
"body": "Logs point to the worker timeout.",
|
|
"created_at": "2026-08-06T12:30:00Z",
|
|
"user": {"login": "alexander"},
|
|
"html_url": "https://forge.example/stackchain/api/issues/7#issuecomment-9",
|
|
})
|
|
return httpx.Response(404)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.notification_detail(42)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests == [
|
|
"http://127.0.0.1:3000/api/v1/notifications/threads/42",
|
|
"http://127.0.0.1:3000/api/v1/repos/stackchain/api/issues/7",
|
|
"http://127.0.0.1:3000/api/v1/repos/stackchain/api/issues/comments/9",
|
|
]
|
|
assert result == {
|
|
"id": 42,
|
|
"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",
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_detail_never_follows_foreign_api_urls():
|
|
requests = []
|
|
|
|
def upstream(request):
|
|
requests.append(str(request.url))
|
|
return httpx.Response(200, json={
|
|
"id": 42,
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"subject": {
|
|
"title": "Suspicious update",
|
|
"url": "https://evil.example/api/v1/user",
|
|
"latest_comment_url": "https://evil.example/api/v1/admin/users",
|
|
"html_url": "javascript:alert(1)",
|
|
},
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.notification_detail(42)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests == [
|
|
"http://127.0.0.1:3000/api/v1/notifications/threads/42"
|
|
]
|
|
assert result["url"] == ""
|
|
assert result["subject_body"] == ""
|
|
assert result["latest_comment"]["body"] == ""
|