231 lines
8.0 KiB
Python
231 lines
8.0 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.idempotency import IdempotencyLedger
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_authored_operations():
|
|
main._authored_action_operations.clear()
|
|
main._idempotency_ledger.clear()
|
|
yield
|
|
main._authored_action_operations.clear()
|
|
main._idempotency_ledger.clear()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_comment_replays_one_upstream_result_for_concurrent_requests(monkeypatch):
|
|
calls = 0
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def assigned(_repository, _number):
|
|
return True
|
|
|
|
async def comment(_repository, _number, body):
|
|
nonlocal calls
|
|
calls += 1
|
|
started.set()
|
|
await release.wait()
|
|
return {"id": 82, "body": body}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "issue-comment-185"}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first = asyncio.create_task(client.post(
|
|
"/api/v1/repos/stackchain/api/issues/7/comments",
|
|
json={"body": "Ship it"}, headers=headers,
|
|
))
|
|
await started.wait()
|
|
second = asyncio.create_task(client.post(
|
|
"/api/v1/repos/stackchain/api/issues/7/comments",
|
|
json={"body": "Ship it"}, headers=headers,
|
|
))
|
|
await asyncio.sleep(0)
|
|
release.set()
|
|
responses = await asyncio.gather(first, second)
|
|
|
|
assert [response.status_code for response in responses] == [201, 201]
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pull_comment_rejects_changed_payload_for_same_key(monkeypatch):
|
|
calls = []
|
|
|
|
async def assigned(_repository, _number):
|
|
return True
|
|
|
|
async def comment(_repository, _number, body):
|
|
calls.append(body)
|
|
return {"id": len(calls), "body": body}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "pull-comment-185"}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
created = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/comments",
|
|
json={"body": "Ship it"}, headers=headers,
|
|
)
|
|
conflict = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/comments",
|
|
json={"body": "Changed"}, headers=headers,
|
|
)
|
|
|
|
assert created.status_code == 201
|
|
assert conflict.status_code == 409
|
|
assert calls == ["Ship it"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_notification_reply_retry_recovers_after_caller_timeout(monkeypatch):
|
|
calls = 0
|
|
|
|
async def reply(_thread_id, body):
|
|
nonlocal calls
|
|
calls += 1
|
|
await asyncio.sleep(0.03)
|
|
return {"id": 91, "body": body}
|
|
|
|
monkeypatch.setattr(main, "NOTIFICATION_MUTATION_TIMEOUT_SECONDS", 0.01)
|
|
monkeypatch.setattr(main.gitea_proxy, "reply_to_notification", reply)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "notification-reply-185"}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
timed_out = await client.post(
|
|
"/api/v1/notifications/42/reply", json={"body": "Retry"}, headers=headers,
|
|
)
|
|
await asyncio.sleep(0.03)
|
|
recovered = await client.post(
|
|
"/api/v1/notifications/42/reply", json={"body": "Retry"}, headers=headers,
|
|
)
|
|
|
|
assert timed_out.status_code == 503
|
|
assert recovered.status_code == 201
|
|
assert recovered.json()["id"] == 91
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_review_replays_same_key_and_rejects_changed_head(monkeypatch):
|
|
calls = []
|
|
|
|
async def requested(_repository, _number):
|
|
return True
|
|
|
|
async def submit(_repository, _number, head, decision, body):
|
|
calls.append((head, decision, body))
|
|
return {"id": 93, "state": "APPROVED"}
|
|
|
|
monkeypatch.setattr(main, "is_requested_review", requested)
|
|
monkeypatch.setattr(main.gitea_proxy, "submit_pull_review", submit)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "review-185"}
|
|
payload = {"expected_head_sha": "abc", "decision": "approve", "body": "Good"}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
created = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/review", json=payload, headers=headers,
|
|
)
|
|
replayed = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/review", json=payload, headers=headers,
|
|
)
|
|
conflict = await client.post(
|
|
"/api/v1/repos/stackchain/api/pulls/7/review",
|
|
json={**payload, "expected_head_sha": "def"}, headers=headers,
|
|
)
|
|
|
|
assert [created.status_code, replayed.status_code, conflict.status_code] == [201, 201, 409]
|
|
assert calls == [("abc", "approve", "Good")]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_completed_comment_replays_after_ledger_reconstruction(monkeypatch, tmp_path):
|
|
calls = 0
|
|
|
|
async def assigned(_repository, _number):
|
|
return True
|
|
|
|
async def comment(_repository, _number, body):
|
|
nonlocal calls
|
|
calls += 1
|
|
return {"id": 201, "body": body}
|
|
|
|
database = tmp_path / "actions.sqlite3"
|
|
monkeypatch.setattr(
|
|
main,
|
|
"_idempotency_ledger",
|
|
IdempotencyLedger(database, ttl_seconds=600, max_entries=256),
|
|
raising=False,
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "restart-comment-201"}
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
created = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/7/comments",
|
|
json={"body": "Persist me"},
|
|
headers=headers,
|
|
)
|
|
main._authored_action_operations.clear()
|
|
monkeypatch.setattr(
|
|
main,
|
|
"_idempotency_ledger",
|
|
IdempotencyLedger(database, ttl_seconds=600, max_entries=256),
|
|
)
|
|
replayed = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/7/comments",
|
|
json={"body": "Persist me"},
|
|
headers=headers,
|
|
)
|
|
|
|
assert [created.status_code, replayed.status_code] == [201, 201]
|
|
assert replayed.json() == {"id": 201, "body": "Persist me"}
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_orphaned_pending_comment_fails_closed_without_upstream_retry(
|
|
monkeypatch, tmp_path
|
|
):
|
|
calls = 0
|
|
|
|
async def assigned(_repository, _number):
|
|
return True
|
|
|
|
async def comment(_repository, _number, body):
|
|
nonlocal calls
|
|
calls += 1
|
|
return {"id": 202, "body": body}
|
|
|
|
ledger = IdempotencyLedger(
|
|
tmp_path / "orphan.sqlite3", ttl_seconds=600, max_entries=256
|
|
)
|
|
fingerprint = ("issue-comment", "stackchain/api", 7, "Verify first")
|
|
assert ledger.reserve("orphan-comment-201", fingerprint).state == "reserved"
|
|
monkeypatch.setattr(main, "_idempotency_ledger", ledger)
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
|
|
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/issues/7/comments",
|
|
json={"body": "Verify first"},
|
|
headers={"Idempotency-Key": "orphan-comment-201"},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert response.json()["detail"] == (
|
|
"This action may still be completing; verify its result before retrying"
|
|
)
|
|
assert calls == 0
|