91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
|
|
|
|
@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
|