89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_review_detail_endpoint_returns_normalized_no_store_payload(monkeypatch):
|
|
async def requested(repository, number):
|
|
return (repository, number) == ("stackchain/api", 7)
|
|
|
|
async def detail(repository, number):
|
|
assert (repository, number) == ("stackchain/api", 7)
|
|
return {
|
|
"repository": repository,
|
|
"number": number,
|
|
"title": "Review API",
|
|
"body": "Check retries",
|
|
"url": "https://forge.example/stackchain/api/pulls/7",
|
|
"author": "alex",
|
|
"ci_state": "success",
|
|
"files": [{"filename": "src/api.py", "status": "modified", "additions": 8, "deletions": 2}],
|
|
"reviews": [{"user": {"login": "sam"}, "state": "APPROVED", "body": "Good"}],
|
|
}
|
|
|
|
monkeypatch.setattr(main, "is_requested_review", requested)
|
|
monkeypatch.setattr(main, "pull_review_detail", detail)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/repos/stackchain/api/pulls/7/review")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["files"][0]["filename"] == "src/api.py"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_review_detail_rejects_pulls_not_requested_from_service_user(monkeypatch):
|
|
async def requested(repository, number):
|
|
return False
|
|
|
|
monkeypatch.setattr(main, "is_requested_review", requested)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/repos/private/secret/pulls/9/review")
|
|
|
|
assert response.status_code == 404
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_review_detail_has_one_retryable_deadline_and_cancels_pending_work(monkeypatch):
|
|
cancelled = asyncio.Event()
|
|
|
|
async def requested(repository, number):
|
|
try:
|
|
await asyncio.Event().wait()
|
|
finally:
|
|
cancelled.set()
|
|
|
|
monkeypatch.setattr(main, "is_requested_review", requested)
|
|
monkeypatch.setattr(main, "REVIEW_DETAIL_TIMEOUT_SECONDS", 0.01)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/repos/stackchain/api/pulls/7/review")
|
|
|
|
assert response.status_code == 503
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.headers["retry-after"] == "1"
|
|
assert response.json() == {
|
|
"error": "Pull request review details timed out. Please retry."
|
|
}
|
|
assert cancelled.is_set()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_public_review_endpoint_does_not_expose_service_token_mutations():
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/review",
|
|
json={"action": "approve"},
|
|
)
|
|
|
|
assert response.status_code == 405
|
|
|