427 lines
15 KiB
Python
427 lines
15 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_notification_snapshot_loads_every_page_with_bounded_concurrency():
|
|
active = 0
|
|
peak = 0
|
|
requested_pages = []
|
|
|
|
async def upstream(request):
|
|
nonlocal active, peak
|
|
page = int(request.url.params["page"])
|
|
requested_pages.append(page)
|
|
active += 1
|
|
peak = max(peak, active)
|
|
await __import__("asyncio").sleep(0)
|
|
active -= 1
|
|
start = (page - 1) * 50 + 1
|
|
end = min(start + 50, 126)
|
|
return httpx.Response(
|
|
200,
|
|
headers={"X-Total-Count": "125"},
|
|
json=[{
|
|
"id": thread_id,
|
|
"updated_at": f"2026-08-11T12:{thread_id % 60:02d}:00Z",
|
|
"subject": {},
|
|
"repository": {},
|
|
} for thread_id in range(start, end)],
|
|
)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.unread_notification_snapshot(
|
|
max_pages=3, max_concurrency=2, deadline_seconds=1
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requested_pages[0] == 1
|
|
assert sorted(requested_pages) == [1, 2, 3]
|
|
assert peak == 2
|
|
assert result["complete"] is True
|
|
assert result["total"] == 125
|
|
assert [item["id"] for item in result["items"]] == list(range(1, 126))
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_unread_notification_snapshot_rejects_pagination_that_changes_mid_scan():
|
|
def upstream(request):
|
|
page = int(request.url.params["page"])
|
|
total = 51 if page == 1 else 50
|
|
start = (page - 1) * 50 + 1
|
|
return httpx.Response(
|
|
200,
|
|
headers={"X-Total-Count": str(total)},
|
|
json=[{
|
|
"id": thread_id,
|
|
"subject": {},
|
|
"repository": {},
|
|
} for thread_id in range(start, min(start + 50, total + 1))],
|
|
)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
with pytest.raises(ValueError, match="changed during the scan"):
|
|
await gitea_proxy.unread_notification_snapshot(deadline_seconds=1)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_unread_notification_snapshot_rejects_missing_or_duplicate_threads():
|
|
def upstream(_request):
|
|
return httpx.Response(
|
|
200,
|
|
headers={"X-Total-Count": "3"},
|
|
json=[
|
|
{"id": 7, "subject": {}, "repository": {}},
|
|
{"id": 7, "subject": {}, "repository": {}},
|
|
"malformed",
|
|
],
|
|
)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
with pytest.raises(ValueError, match="incomplete thread set"):
|
|
await gitea_proxy.unread_notification_snapshot(deadline_seconds=1)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_unread_notification_snapshot_fails_before_exceeding_the_scan_limit():
|
|
requested_pages = []
|
|
|
|
def upstream(request):
|
|
page = int(request.url.params["page"])
|
|
requested_pages.append(page)
|
|
start = (page - 1) * 50 + 1
|
|
return httpx.Response(
|
|
200,
|
|
headers={"X-Total-Count": "101"},
|
|
json=[
|
|
{"id": thread_id, "subject": {}, "repository": {}}
|
|
for thread_id in range(start, min(start + 50, 102))
|
|
],
|
|
)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
with pytest.raises(ValueError, match="scan limit"):
|
|
await gitea_proxy.unread_notification_snapshot(
|
|
max_pages=2, deadline_seconds=1
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requested_pages == [1]
|
|
|
|
|
|
@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={
|
|
"number": 7, "state": "open", "assignees": [],
|
|
"body": "Deploy fails after **three** retries.",
|
|
})
|
|
if request.url.path.endswith("/issues/7/subscriptions/check"):
|
|
return httpx.Response(200, json={"subscribed": True, "ignored": False})
|
|
if request.url.path.endswith("/issues/comments/9"):
|
|
return httpx.Response(200, json={
|
|
"id": 9,
|
|
"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",
|
|
})
|
|
if request.url.path.endswith("/issues/7/comments"):
|
|
return httpx.Response(200, json=[{
|
|
"id": 9,
|
|
"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",
|
|
}], headers={"X-Total-Count": "1"})
|
|
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",
|
|
"http://127.0.0.1:3000/api/v1/repos/stackchain/api/issues/7/subscriptions/check",
|
|
]
|
|
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",
|
|
},
|
|
"issue": {"number": 7, "assignees": [], "claimable": True},
|
|
"acknowledge_supported": True,
|
|
"mute_supported": True,
|
|
"conversation_available": True,
|
|
}
|
|
|
|
|
|
@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"] == ""
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("subject_kind", ["issues", "pulls"])
|
|
async def test_reply_to_notification_posts_to_its_issue_conversation(subject_kind):
|
|
requests = []
|
|
|
|
def upstream(request):
|
|
requests.append((request.method, str(request.url), request.content))
|
|
if request.method == "GET":
|
|
return httpx.Response(200, json={
|
|
"id": 42,
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"subject": {
|
|
"type": "Pull" if subject_kind == "pulls" else "Issue",
|
|
"url": (
|
|
"http://127.0.0.1:3000/api/v1/repos/stackchain/api/"
|
|
f"{subject_kind}/7"
|
|
),
|
|
},
|
|
})
|
|
return httpx.Response(201, json={
|
|
"id": 91,
|
|
"user": {"login": "timmy"},
|
|
"body": "Please retry the worker.",
|
|
"created_at": "2026-08-07T19:00:00Z",
|
|
"html_url": "https://forge.example/stackchain/api/issues/7#issuecomment-91",
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.reply_to_notification(42, "Please retry the worker.")
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests == [
|
|
(
|
|
"GET",
|
|
"http://127.0.0.1:3000/api/v1/notifications/threads/42",
|
|
b"",
|
|
),
|
|
(
|
|
"POST",
|
|
"http://127.0.0.1:3000/api/v1/repos/stackchain/api/issues/7/comments",
|
|
b'{"body":"Please retry the worker."}',
|
|
),
|
|
]
|
|
assert result == {
|
|
"id": 91,
|
|
"author": "timmy",
|
|
"body": "Please retry the worker.",
|
|
"created_at": "2026-08-07T19:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/7#issuecomment-91",
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_mute_notification_unsubscribes_current_user_from_trusted_thread_target():
|
|
requests = []
|
|
|
|
def upstream(request):
|
|
requests.append((request.method, request.url.path))
|
|
if request.url.path.endswith("/notifications/threads/42"):
|
|
return httpx.Response(200, json={
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"subject": {
|
|
"type": "Pull",
|
|
"url": "http://127.0.0.1:3000/api/v1/repos/stackchain/api/pulls/7",
|
|
},
|
|
})
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.method == "DELETE":
|
|
return httpx.Response(204)
|
|
return httpx.Response(404)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.mute_notification(42)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result == {"id": 42, "repository": "stackchain/api", "number": 7, "muted": True}
|
|
assert requests == [
|
|
("GET", "/api/v1/notifications/threads/42"),
|
|
("GET", "/api/v1/user"),
|
|
("DELETE", "/api/v1/repos/stackchain/api/issues/7/subscriptions/timmy"),
|
|
]
|