126 lines
4.6 KiB
Python
126 lines
4.6 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy, main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_reply_returns_the_confirmed_comment_for_exact_once_append(monkeypatch):
|
|
def handler(request):
|
|
if request.method == "GET":
|
|
return httpx.Response(200, json={
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"subject": {
|
|
"type": "Issue",
|
|
"url": "https://forge.example/api/v1/repos/stackchain/api/issues/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",
|
|
})
|
|
|
|
monkeypatch.setattr(gitea_proxy, "GITEA_URL", "https://forge.example")
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.reply_to_notification(42, "Please retry the worker.")
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
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_notification_reply_api_validates_and_posts_without_marking_read(monkeypatch):
|
|
calls = []
|
|
|
|
async def reply(thread_id, body):
|
|
calls.append((thread_id, body))
|
|
return {
|
|
"id": 91,
|
|
"url": "https://forge.example/stackchain/api/issues/7#issuecomment-91",
|
|
}
|
|
|
|
async def must_not_mark_read(_thread_id):
|
|
raise AssertionError("replying must not mark the update read")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "reply_to_notification", reply)
|
|
monkeypatch.setattr(main, "mark_notification_read", must_not_mark_read)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/notifications/42/reply",
|
|
json={"body": " Please retry the worker. "},
|
|
)
|
|
empty = await client.post(
|
|
"/api/v1/notifications/42/reply", json={"body": " "}
|
|
)
|
|
oversized = await client.post(
|
|
"/api/v1/notifications/42/reply", json={"body": "x" * 10_001}
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json() == {
|
|
"id": 91,
|
|
"url": "https://forge.example/stackchain/api/issues/7#issuecomment-91",
|
|
}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert empty.status_code == 422
|
|
assert oversized.status_code == 422
|
|
assert calls == [(42, "Please retry the worker.")]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("failure", ["timeout", "upstream"])
|
|
async def test_notification_reply_failure_is_sanitized_retryable_and_no_store(
|
|
monkeypatch, failure
|
|
):
|
|
async def reply(_thread_id, _body):
|
|
if failure == "timeout":
|
|
await asyncio.sleep(0.05)
|
|
raise httpx.HTTPError("token=secret upstream exploded")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "reply_to_notification", reply)
|
|
monkeypatch.setattr(main, "NOTIFICATION_MUTATION_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.post(
|
|
"/api/v1/notifications/42/reply", json={"body": "Retry it"}
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {
|
|
"error": "The reply could not be posted. Your draft is safe; please retry."
|
|
}
|
|
assert response.headers["retry-after"] == "1"
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert "secret" not in response.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_reply_cannot_be_invoked_by_a_foreign_browser_origin():
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.options(
|
|
"/api/v1/notifications/42/reply",
|
|
headers={
|
|
"Origin": "https://evil.example",
|
|
"Access-Control-Request-Method": "POST",
|
|
"Access-Control-Request-Headers": "content-type",
|
|
},
|
|
)
|
|
|
|
assert response.status_code != 200
|
|
assert "access-control-allow-origin" not in response.headers
|