import asyncio import base64 import json import httpx import pytest from src import gitea_proxy, main from src.security_event_store import SecurityEventStoreError def _content_payload(text: str, sha: str) -> dict: return { "type": "file", "encoding": "base64", "content": base64.b64encode(text.encode()).decode(), "sha": sha, } @pytest.mark.anyio async def test_gitea_loads_bounded_text_file_from_authored_pull_head(): pull = { "number": 7, "state": "open", "merged": False, "user": {"login": "alex"}, "head": { "sha": "abc1234", "ref": "alex/review-fix", "repo": {"full_name": "stackchain/api"}, }, "base": {"repo": {"full_name": "stackchain/api"}}, } async def handler(request): if request.url.path == "/api/v1/user": return httpx.Response(200, json={"login": "alex"}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7": return httpx.Response(200, json=pull) if request.url.path == "/api/v1/repos/stackchain/api/contents/src/api.py": assert request.url.params["ref"] == "abc1234" return httpx.Response(200, json=_content_payload("return empty\n", "blob123")) raise AssertionError(f"unexpected request: {request.method} {request.url}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.authored_pull_feedback_file( "stackchain/api", 7, "src/api.py", "abc1234" ) finally: await gitea_proxy.stop_client() assert result == { "repository": "stackchain/api", "number": 7, "path": "src/api.py", "head_sha": "abc1234", "blob_sha": "blob123", "content": "return empty\n", } @pytest.mark.anyio async def test_gitea_commits_feedback_fix_and_verifies_advanced_pull_head(): requests = [] pull = { "number": 7, "state": "open", "merged": False, "user": {"login": "alex"}, "head": { "sha": "abc1234", "ref": "alex/review-fix", "repo": {"full_name": "stackchain/api"}, }, "base": {"repo": {"full_name": "stackchain/api"}}, } async def handler(request): requests.append((request.method, request.url.path)) if request.url.path == "/api/v1/user": return httpx.Response(200, json={"login": "alex"}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7": return httpx.Response(200, json=pull) if request.url.path == "/api/v1/repos/stackchain/api/contents/src/api.py" and request.method == "GET": ref = request.url.params["ref"] return httpx.Response(200, json=_content_payload( "return handled\n" if ref == "def5678" else "return empty\n", "blob456" if ref == "def5678" else "blob123", )) if request.url.path == "/api/v1/repos/stackchain/api/contents/src/api.py" and request.method == "PUT": body = json.loads(request.content) assert body == { "branch": "alex/review-fix", "sha": "blob123", "message": "fix: handle empty state", "content": base64.b64encode(b"return handled\n").decode(), } pull["head"]["sha"] = "def5678" return httpx.Response(200, json={"commit": {"sha": "def5678"}}) raise AssertionError(f"unexpected request: {request.method} {request.url}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.commit_authored_pull_feedback_fix( "stackchain/api", 7, "src/api.py", "return handled\n", "fix: handle empty state", "abc1234", "blob123", ) finally: await gitea_proxy.stop_client() assert result["previous_head_sha"] == "abc1234" assert result["head_sha"] == "def5678" assert result["path"] == "src/api.py" assert requests[-2:] == [ ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ("GET", "/api/v1/repos/stackchain/api/contents/src/api.py"), ] @pytest.mark.anyio async def test_feedback_file_api_loads_and_commits_an_authored_pull_fix(monkeypatch): calls = [] async def load(repository, number, path, head): calls.append(("load", repository, number, path, head)) return {"path": path, "head_sha": head, "blob_sha": "blob123", "content": "before\n"} async def commit(repository, number, path, content, message, head, blob): calls.append(("commit", repository, number, path, content, message, head, blob)) return {"path": path, "previous_head_sha": head, "head_sha": "def5678", "blob_sha": "blob456"} monkeypatch.setattr(main.gitea_proxy, "authored_pull_feedback_file", load, raising=False) monkeypatch.setattr(main.gitea_proxy, "commit_authored_pull_feedback_fix", commit, raising=False) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: loaded = await client.get( "/api/v1/repos/stackchain/api/pulls/7/feedback-file", params={"path": "src/api.py", "expected_head_sha": "abc1234"}, ) committed = await client.patch( "/api/v1/repos/stackchain/api/pulls/7/feedback-file", json={ "path": "src/api.py", "content": "after\n", "message": "fix: address review", "expected_head_sha": "abc1234", "expected_blob_sha": "blob123", }, ) assert loaded.status_code == 200 assert loaded.json()["blob_sha"] == "blob123" assert committed.status_code == 200 assert committed.json()["head_sha"] == "def5678" assert calls == [ ("load", "stackchain/api", 7, "src/api.py", "abc1234"), ("commit", "stackchain/api", 7, "src/api.py", "after\n", "fix: address review", "abc1234", "blob123"), ] @pytest.mark.anyio async def test_gitea_updates_authored_pull_branch_and_confirms_new_head(): requests = [] pull = { "number": 7, "title": "Ship mobile flow", "state": "open", "draft": False, "merged": False, "mergeable": False, "user": {"login": "alex"}, "head": {"sha": "abc1234"}, } async def handler(request): requests.append((request.method, request.url.path, request.content)) if request.url.path == "/api/v1/user": return httpx.Response(200, json={"login": "alex"}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7" and request.method == "GET": return httpx.Response(200, json=pull) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7/update": assert request.method == "POST" pull["head"] = {"sha": "def5678"} return httpx.Response(200, json={"message": "updated"}) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.update_authored_pull_branch( "stackchain/api", 7, "abc1234" ) finally: await gitea_proxy.stop_client() assert result == { "repository": "stackchain/api", "number": 7, "title": "Ship mobile flow", "previous_head_sha": "abc1234", "head_sha": "def5678", "state": "open", "draft": False, } assert [item[:2] for item in requests] == [ ("GET", "/api/v1/user"), ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ("POST", "/api/v1/repos/stackchain/api/pulls/7/update"), ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ] @pytest.mark.anyio async def test_update_pull_branch_api_maps_stale_head_and_conflict(monkeypatch): outcomes = [ gitea_proxy.IssueNotAvailableError("stale"), gitea_proxy.PullUpdateConflictError("conflict"), ] async def update(*_args): raise outcomes.pop(0) monkeypatch.setattr(main.gitea_proxy, "update_authored_pull_branch", update, raising=False) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: stale = await client.post( "/api/v1/repos/stackchain/api/pulls/7/update-branch", json={"expected_head_sha": "abc1234"}, ) conflict = await client.post( "/api/v1/repos/stackchain/api/pulls/7/update-branch", json={"expected_head_sha": "abc1234"}, ) assert stale.status_code == 409 assert "Reload" in stale.json()["error"] assert conflict.status_code == 422 assert "conflicts" in conflict.json()["error"] @pytest.mark.anyio async def test_gitea_closes_authored_pull_at_expected_head_and_verifies_state(): requests = [] pull = { "number": 7, "title": "Obsolete experiment", "state": "open", "merged": False, "user": {"login": "alex"}, "head": {"sha": "abc1234"}, } async def handler(request): requests.append((request.method, request.url.path, request.content)) if request.url.path == "/api/v1/user": return httpx.Response(200, json={"login": "alex"}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7" and request.method == "GET": return httpx.Response(200, json=pull) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7" and request.method == "PATCH": assert json.loads(request.content) == {"state": "closed"} pull["state"] = "closed" return httpx.Response(200, json=pull) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.close_authored_pull("stackchain/api", 7, "abc1234") finally: await gitea_proxy.stop_client() assert result == { "repository": "stackchain/api", "number": 7, "title": "Obsolete experiment", "head_sha": "abc1234", "state": "closed", "merged": False, } assert [item[:2] for item in requests] == [ ("GET", "/api/v1/user"), ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ("PATCH", "/api/v1/repos/stackchain/api/pulls/7"), ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ] @pytest.mark.anyio async def test_gitea_reopens_authored_unmerged_pull_at_expected_head(): pull = { "number": 7, "title": "Obsolete experiment", "state": "closed", "merged": False, "user": {"login": "alex"}, "head": {"sha": "abc1234"}, } async def handler(request): if request.url.path == "/api/v1/user": return httpx.Response(200, json={"login": "alex"}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7" and request.method == "GET": return httpx.Response(200, json=pull) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7" and request.method == "PATCH": assert json.loads(request.content) == {"state": "open"} pull["state"] = "open" return httpx.Response(200, json=pull) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.reopen_authored_pull("stackchain/api", 7, "abc1234") finally: await gitea_proxy.stop_client() assert result["state"] == "open" assert result["head_sha"] == "abc1234" @pytest.mark.anyio async def test_author_can_close_then_reopen_pull_through_api(monkeypatch): calls = [] async def transition(repository, number, expected_head_sha, state): calls.append((state, repository, number, expected_head_sha)) return { "repository": repository, "number": number, "head_sha": expected_head_sha, "state": state, "merged": False, } monkeypatch.setattr( main.gitea_proxy, "close_authored_pull", lambda repository, number, head: transition(repository, number, head, "closed"), ) monkeypatch.setattr( main.gitea_proxy, "reopen_authored_pull", lambda repository, number, head: transition(repository, number, head, "open"), ) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: closed = await client.patch( "/api/v1/repos/stackchain/api/pulls/7/close", json={"expected_head_sha": "abc1234"}, ) reopened = await client.patch( "/api/v1/repos/stackchain/api/pulls/7/reopen", json={"expected_head_sha": "abc1234"}, ) assert closed.status_code == 200 assert closed.json()["state"] == "closed" assert reopened.status_code == 200 assert reopened.json()["state"] == "open" assert calls == [ ("closed", "stackchain/api", 7, "abc1234"), ("open", "stackchain/api", 7, "abc1234"), ] @pytest.mark.anyio async def test_pull_author_can_publish_assigned_draft_at_expected_head(monkeypatch): calls = [] async def publish(repository, number, expected_head_sha): calls.append((repository, number, expected_head_sha)) return { "repository": repository, "number": number, "title": "Ship mobile flow", "head_sha": expected_head_sha, "state": "open", "draft": False, } monkeypatch.setattr( main.gitea_proxy, "publish_authored_assigned_pull", publish, raising=False ) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.patch( "/api/v1/repos/stackchain/api/pulls/7/ready", json={"expected_head_sha": "abc1234"}, ) assert response.status_code == 200 assert response.json()["draft"] is False assert calls == [("stackchain/api", 7, "abc1234")] @pytest.mark.anyio async def test_gitea_publishes_authored_unassigned_draft_and_verifies_ready_head(): requests = [] pull = { "number": 7, "title": "WIP: Ship mobile flow", "state": "open", "draft": True, "merged": False, "user": {"login": "alex"}, "assignees": [], "head": {"sha": "abc1234"}, } async def handler(request): requests.append((request.method, request.url.path, request.content)) if request.url.path == "/api/v1/user": return httpx.Response(200, json={"login": "alex"}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7" and request.method == "GET": return httpx.Response(200, json=pull) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7" and request.method == "PATCH": assert json.loads(request.content) == {"title": "Ship mobile flow"} pull.update({"title": "Ship mobile flow", "draft": False}) return httpx.Response(201, json=pull) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.publish_authored_assigned_pull( "stackchain/api", 7, "abc1234" ) finally: await gitea_proxy.stop_client() assert result == { "repository": "stackchain/api", "number": 7, "title": "Ship mobile flow", "head_sha": "abc1234", "state": "open", "draft": False, } assert [item[:2] for item in requests] == [ ("GET", "/api/v1/user"), ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ("PATCH", "/api/v1/repos/stackchain/api/pulls/7"), ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ] @pytest.mark.anyio async def test_pull_author_can_update_assigned_open_pull_context(monkeypatch): calls = [] async def update(repository, number, title, body, expected_head_sha): calls.append((repository, number, title, body, expected_head_sha)) return { "repository": repository, "number": number, "title": title, "body": body, "head_sha": expected_head_sha, "state": "open", } monkeypatch.setattr( main.gitea_proxy, "update_authored_assigned_pull", update, raising=False ) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.patch( "/api/v1/repos/stackchain/api/pulls/7/content", json={ "title": "Clarify mobile handoff", "body": "Explain the reviewer path.", "expected_head_sha": "abc1234", }, ) assert response.status_code == 200 assert response.json()["title"] == "Clarify mobile handoff" assert calls == [ ( "stackchain/api", 7, "Clarify mobile handoff", "Explain the reviewer path.", "abc1234", ) ] @pytest.mark.anyio async def test_gitea_pull_context_update_allows_author_without_assignment_at_matching_head(): requests = [] async def handler(request): requests.append((request.method, request.url.path, request.content)) if request.url.path == "/api/v1/user": return httpx.Response(200, json={"login": "alex"}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7" and request.method == "GET": return httpx.Response(200, json={ "number": 7, "title": "Old context", "body": "Old body", "state": "open", "merged": False, "user": {"login": "alex"}, "assignees": [], "head": {"sha": "abc1234"}, "html_url": "https://forge.example/stackchain/api/pulls/7", }) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7" and request.method == "PATCH": assert json.loads(request.content) == { "title": "Clear review brief", "body": "Review the mobile handoff." } return httpx.Response(200, json={ "number": 7, "title": "Clear review brief", "body": "Review the mobile handoff.", "state": "open", "head": {"sha": "abc1234"}, "html_url": "https://forge.example/stackchain/api/pulls/7", }) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.update_authored_assigned_pull( "stackchain/api", 7, "Clear review brief", "Review the mobile handoff.", "abc1234" ) finally: await gitea_proxy.stop_client() assert result["title"] == "Clear review brief" assert [request[:2] for request in requests] == [ ("GET", "/api/v1/user"), ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ("PATCH", "/api/v1/repos/stackchain/api/pulls/7"), ] @pytest.mark.anyio async def test_gitea_authored_unassigned_pull_has_workspace_access(): requests = [] async def handler(request): requests.append((request.method, request.url.path)) if request.url.path == "/api/v1/user": return httpx.Response(200, json={"login": "alex"}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7": return httpx.Response(200, json={ "number": 7, "state": "open", "merged": False, "user": {"login": "alex"}, "assignees": [], }) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: capabilities = await gitea_proxy.pull_workspace_capabilities("stackchain/api", 7) finally: await gitea_proxy.stop_client() assert capabilities == {"authored": True, "assigned": False} assert requests == [ ("GET", "/api/v1/user"), ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ] @pytest.mark.anyio async def test_authored_pull_detail_reuses_one_authorized_snapshot(): requests = [] async def handler(request): requests.append((request.method, request.url.path)) if request.url.path == "/api/v1/user": return httpx.Response(200, json={"login": "alex"}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7": return httpx.Response(200, json={ "number": 7, "title": "Ship mobile flow", "body": "Ready for review", "state": "open", "merged": False, "head": {"sha": "abc1234"}, "user": {"login": "alex"}, "assignees": [], }) if request.url.path == "/api/v1/repos/stackchain/api/issues/7/comments": return httpx.Response(200, json=[], headers={"X-Total-Count": "0"}) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: 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/detail") finally: await gitea_proxy.stop_client() assert response.status_code == 200 assert response.json()["capabilities"] == {"authored": True, "assigned": False} assert requests.count(("GET", "/api/v1/user")) == 1 assert requests.count(("GET", "/api/v1/repos/stackchain/api/pulls/7")) == 1 assert requests.count(("GET", "/api/v1/repos/stackchain/api/issues/7/comments")) == 1 @pytest.mark.anyio async def test_authored_unassigned_pull_detail_returns_capabilities(monkeypatch): pull = {"number": 7} async def snapshot(repository, number): assert (repository, number) == ("stackchain/api", 7) return {"pull": pull, "capabilities": {"authored": True, "assigned": False}} async def detail(repository, number, authorized_pull): assert authorized_pull is pull return { "repository": repository, "number": number, "title": "Ship mobile flow", "body": "Ready for review", "author": "alex", "head_sha": "abc1234", "state": "open", "conversation": {"comments": [], "page": 1, "older_page": None, "total": 0}, } monkeypatch.setattr(main.gitea_proxy, "pull_workspace_snapshot", snapshot) monkeypatch.setattr(main.gitea_proxy, "pull_completion_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/detail") assert response.status_code == 200 assert response.json()["capabilities"] == {"authored": True, "assigned": False} @pytest.mark.anyio async def test_assigned_pull_detail_reports_completion_state(monkeypatch): pull = {"number": 7} async def snapshot(repository, number): assert (repository, number) == ("stackchain/api", 7) return {"pull": pull, "capabilities": {"authored": False, "assigned": True}} async def detail(repository, number, authorized_pull): assert (repository, number) == ("stackchain/api", 7) assert authorized_pull is pull return { "repository": repository, "number": number, "title": "Ship mobile flow", "body": "Ready to merge", "url": "https://forge.example/stackchain/api/pulls/7", "author": "alex", "head_sha": "abc123", "state": "open", "draft": False, "mergeable": True, "merged": False, "ci_state": "success", "files": [{"filename": "src/api.py", "additions": 8, "deletions": 2}], "comments": [{"id": 9, "author": "sam", "body": "Ship it"}], } monkeypatch.setattr(main.gitea_proxy, "pull_workspace_snapshot", snapshot) monkeypatch.setattr(main.gitea_proxy, "pull_completion_detail", detail, raising=False) 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/detail") assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" assert response.json()["head_sha"] == "abc123" assert response.json()["mergeable"] is True @pytest.mark.anyio async def test_assigned_pull_detail_rejects_unassigned_pull(monkeypatch): async def snapshot(repository, number): return {"pull": {}, "capabilities": {"authored": False, "assigned": False}} monkeypatch.setattr(main.gitea_proxy, "pull_workspace_snapshot", snapshot) 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/detail") assert response.status_code == 404 @pytest.mark.anyio async def test_gitea_assigned_pull_detail_loads_reading_without_review_resources(): requests = [] async def handler(request): requests.append((request.method, request.url.path)) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "number": 7, "title": "Read this first", "body": "The discussion should not wait for the diff.", "state": "open", "mergeable": True, "head": {"sha": "abc123"}, "user": {"login": "alex"}, }) if request.url.path.endswith("/issues/7/comments"): return httpx.Response(200, json=[{ "id": 9, "body": "Question", "user": {"login": "sam"}, "created_at": "2026-08-07T18:00:00Z", }], headers={"X-Total-Count": "1"}) raise AssertionError(f"review resource requested during read load: {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: detail = await gitea_proxy.pull_completion_detail("stackchain/api", 7) finally: await gitea_proxy.stop_client() assert requests == [ ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ("GET", "/api/v1/repos/stackchain/api/issues/7/comments"), ] assert detail["head_sha"] == "abc123" assert detail["conversation"]["comments"][0]["body"] == "Question" assert "files" not in detail assert "ci_state" not in detail @pytest.mark.anyio async def test_assigned_pull_review_endpoint_loads_review_payload_on_demand(monkeypatch): calls = [] pull = {"number": 7} async def snapshot(repository, number): calls.append(("snapshot", repository, number)) return {"pull": pull, "capabilities": {"authored": False, "assigned": True}} async def review(repository, number, authorized_pull): assert authorized_pull is pull calls.append(("review", repository, number)) return { "repository": repository, "number": number, "head_sha": "abc123", "state": "open", "draft": False, "mergeable": True, "merged": False, "ci_state": "success", "files": [{"filename": "src/api.py"}], } monkeypatch.setattr(main.gitea_proxy, "pull_workspace_snapshot", snapshot) monkeypatch.setattr(main.gitea_proxy, "pull_completion_review", review, raising=False) 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-data") assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" assert response.json()["files"] == [{"filename": "src/api.py"}] assert calls == [ ("snapshot", "stackchain/api", 7), ("review", "stackchain/api", 7), ] @pytest.mark.anyio async def test_assigned_pull_review_reuses_one_authorized_snapshot(): requests = [] async def handler(request): requests.append((request.method, request.url.path)) if request.url.path == "/api/v1/user": return httpx.Response(200, json={"login": "alex"}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7": return httpx.Response(200, json={ "number": 7, "state": "open", "merged": False, "mergeable": True, "head": {"sha": "abc1234"}, "user": {"login": "sam"}, "assignees": [{"login": "alex"}], "requested_reviewers": [], }) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7/files": return httpx.Response(200, json=[]) if request.url.path == "/api/v1/repos/stackchain/api/commits/abc1234/status": return httpx.Response(200, json={"state": "pending", "statuses": []}) if request.url.path == "/api/v1/repos/stackchain/api/pulls/7.diff": return httpx.Response(200, text="") if request.url.path == "/api/v1/repos/stackchain/api/pulls/7/reviews": return httpx.Response(200, json=[]) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: 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-data") finally: await gitea_proxy.stop_client() assert response.status_code == 200 assert response.json()["capabilities"] == {"authored": False, "assigned": True} assert requests.count(("GET", "/api/v1/user")) == 1 assert requests.count(("GET", "/api/v1/repos/stackchain/api/pulls/7")) == 1 @pytest.mark.anyio async def test_assigned_pull_feedback_endpoint_authorizes_and_returns_one_review(monkeypatch): calls = [] async def assigned(repository, number): calls.append(("assigned", repository, number)) return True async def feedback(repository, number, review_id, expected_head_sha): calls.append(("feedback", repository, number, review_id, expected_head_sha)) return { "review_id": review_id, "head_sha": expected_head_sha, "reviewed_head_sha": expected_head_sha, "comments": [], } monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "pull_review_feedback", feedback, raising=False) 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/reviews/8/feedback?expected_head_sha=abc1234" ) assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" assert response.json()["review_id"] == 8 assert calls == [ ("assigned", "stackchain/api", 7), ("feedback", "stackchain/api", 7, 8, "abc1234"), ] def test_reviewer_statuses_distinguish_waiting_current_and_outdated_decisions(): pull = { "requested_reviewers": [{"login": "casey"}], "head": {"sha": "new-head"}, } reviews = [ {"id": 1, "user": {"login": "casey"}, "state": "APPROVED", "commit_id": "old-head"}, {"id": 2, "user": {"login": "sam"}, "state": "REQUEST_CHANGES", "commit_id": "new-head"}, {"id": 3, "user": {"login": "lee"}, "state": "COMMENT", "commit_id": "new-head"}, {"id": 4, "user": {"login": "pat"}, "state": "APPROVED", "commit_id": "new-head"}, ] assert gitea_proxy._normalize_reviewer_statuses(pull, reviews, "new-head") == [ {"login": "casey", "status": "waiting", "head_sha": "new-head", "blocking": True}, {"review_id": 3, "login": "lee", "status": "commented", "head_sha": "new-head", "blocking": False}, {"review_id": 4, "login": "pat", "status": "approved", "head_sha": "new-head", "blocking": False}, {"review_id": 2, "login": "sam", "status": "changes_requested", "head_sha": "new-head", "blocking": True}, ] pull["requested_reviewers"] = [] assert gitea_proxy._normalize_reviewer_statuses(pull, reviews[:1], "new-head") == [ {"review_id": 1, "login": "casey", "status": "outdated", "head_sha": "old-head", "blocking": True}, ] def test_reviewer_status_exposes_only_bounded_nonempty_feedback_summary(): pull = {"requested_reviewers": []} reviews = [ { "id": 8, "user": {"login": "sam"}, "state": "REQUEST_CHANGES", "commit_id": "abc123", "body": "Please split this helper. " + ("x" * 900), }, { "id": 9, "user": {"login": "lee"}, "state": "COMMENT", "commit_id": "abc123", "body": " ", }, ] statuses = gitea_proxy._normalize_reviewer_statuses(pull, reviews, "abc123") sam = next(item for item in statuses if item["login"] == "sam") lee = next(item for item in statuses if item["login"] == "lee") assert sam["summary"].startswith("Please split this helper.") assert len(sam["summary"]) == 500 assert "summary" not in lee def test_review_comments_are_bounded_and_require_a_file_and_body(): comments = [ {"id": 1, "path": "src/api.py", "body": "Handle the empty state", "new_position": 4}, {"id": 2, "path": "src/api.py", "body": "x" * 900, "old_position": 7}, {"id": 3, "path": "", "body": "missing file"}, {"id": 4, "path": "src/ignored.py", "body": " "}, "malformed", ] + [ {"id": index, "path": f"src/{index}.py", "body": "bounded"} for index in range(5, 40) ] normalized = gitea_proxy._normalize_review_comments(comments) assert len(normalized) == 20 assert normalized[0] == { "id": 1, "path": "src/api.py", "body": "Handle the empty state", "line": 4, } assert normalized[1]["line"] == 7 assert len(normalized[1]["body"]) == 500 assert all(item["path"] and item["body"] for item in normalized) def test_review_feedback_preserves_stable_review_and_comment_identities(): pull = {"requested_reviewers": []} reviews = [{ "id": 42, "user": {"login": "sam"}, "state": "REQUEST_CHANGES", "commit_id": "abc123", }] statuses = gitea_proxy._normalize_reviewer_statuses(pull, reviews, "abc123") comments = gitea_proxy._normalize_review_comments([ {"id": 99, "path": "src/api.py", "body": "Handle the empty state", "new_position": 4}, {"id": -1, "path": "src/other.py", "body": "Invalid identity is omitted"}, ]) assert statuses == [{ "review_id": 42, "login": "sam", "status": "changes_requested", "head_sha": "abc123", "blocking": True, }] assert comments[0]["id"] == 99 assert "id" not in comments[1] def test_inline_feedback_is_loaded_only_for_each_reviewers_latest_decision(): reviews = [ {"id": 7, "user": {"login": "sam"}, "state": "REQUEST_CHANGES"}, {"id": 8, "user": {"login": "sam"}, "state": "APPROVED"}, {"id": 9, "user": {"login": "lee"}, "state": "COMMENT"}, {"id": 10, "user": {"login": "pat"}, "state": "PENDING"}, ] assert gitea_proxy._latest_feedback_review_ids(reviews) == {"lee": 9} @pytest.mark.anyio async def test_gitea_pull_check_status_refreshes_reviewer_decisions_without_files_or_diff(): requests = [] async def handler(request): requests.append((request.method, request.url.path)) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "state": "open", "draft": False, "mergeable": True, "merged": False, "head": {"sha": "abc123"}, "requested_reviewers": [{"login": "casey"}], }) if request.url.path.endswith("/pulls/7/reviews"): return httpx.Response(200, json=[]) if request.url.path.endswith("/commits/abc123/status"): return httpx.Response(200, json={ "state": "success", "statuses": [{"context": "tests", "status": "success", "description": "Passed"}], }) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: status = await gitea_proxy.pull_check_status("stackchain/api", 7) finally: await gitea_proxy.stop_client() assert requests == [ ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ("GET", "/api/v1/repos/stackchain/api/commits/abc123/status"), ("GET", "/api/v1/repos/stackchain/api/pulls/7/reviews"), ] assert status == { "repository": "stackchain/api", "number": 7, "head_sha": "abc123", "state": "open", "draft": False, "mergeable": True, "merged": False, "ci_state": "success", "checks": [{"name": "tests", "state": "success", "description": "Passed", "url": ""}], "reviewers": [ {"login": "casey", "status": "waiting", "head_sha": "abc123", "blocking": True}, ], } @pytest.mark.anyio async def test_assigned_pull_checks_endpoint_authorizes_and_returns_status_only(monkeypatch): calls = [] async def assigned(repository, number): calls.append(("assigned", repository, number)) return True async def checks(repository, number): calls.append(("checks", repository, number)) return {"head_sha": "abc123", "ci_state": "pending", "checks": []} monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "pull_check_status", checks) 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/checks") assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" assert response.json() == {"head_sha": "abc123", "ci_state": "pending", "checks": []} assert calls == [ ("assigned", "stackchain/api", 7), ("checks", "stackchain/api", 7), ] def test_commit_checks_expose_only_same_forge_actions_recovery_metadata(monkeypatch): monkeypatch.setattr( gitea_proxy, "GITEA_URL", "https://forge.example/git" ) checks = gitea_proxy._normalize_commit_checks({ "statuses": [ { "context": "CI / lint (pull_request)", "status": "failure", "description": "Failed", "target_url": "/git/acme/mobile/actions/runs/91/jobs/3", }, { "context": "external", "status": "failure", "target_url": "https://checks.example/jobs/4", }, ] }) assert checks[0] == { "name": "CI / lint (pull_request)", "state": "failure", "description": "Failed", "url": "https://forge.example/git/acme/mobile/actions/runs/91/jobs/3", "recovery": {"run_id": 91, "job_index": 3}, } assert checks[1]["url"] == "" assert "recovery" not in checks[1] @pytest.mark.anyio async def test_action_failure_excerpt_is_bounded_and_redacts_credentials(monkeypatch): monkeypatch.setattr(gitea_proxy, "GITEA_URL", "http://test/git") requests = [] noisy_log = "old output\n" * 4_000 + ( "Authorization: token super-secret\n" "GITEA_TOKEN=another-secret\n" "AssertionError: expected ready, got blocked\n" ) async def handler(request): requests.append((request.method, request.url.path)) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={"head": {"sha": "abc1234"}}) if request.url.path.endswith("/commits/abc1234/status"): return httpx.Response(200, json={"statuses": [{ "context": "CI / lint (pull_request)", "status": "failure", "target_url": "/git/stackchain/api/actions/runs/91/jobs/3", }]}) if request.url.path.endswith("/actions/runs/91/jobs/3/logs"): return httpx.Response(200, text=noisy_log) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.action_failure_excerpt( "stackchain/api", 7, "abc1234", 91, 3 ) finally: await gitea_proxy.stop_client() assert result["head_sha"] == "abc1234" assert result["run_id"] == 91 assert result["job_index"] == 3 assert len(result["excerpt"].encode()) <= 24 * 1024 assert "AssertionError: expected ready, got blocked" in result["excerpt"] assert "super-secret" not in result["excerpt"] assert "another-secret" not in result["excerpt"] assert "Authorization: [redacted]" in result["excerpt"] assert requests == [ ("GET", "/git/api/v1/repos/stackchain/api/pulls/7"), ("GET", "/git/api/v1/repos/stackchain/api/commits/abc1234/status"), ("GET", "/git/stackchain/api/actions/runs/91/jobs/3/logs"), ] @pytest.mark.anyio async def test_action_failure_endpoint_authorizes_and_returns_no_store(monkeypatch): calls = [] async def capabilities(repository, number): calls.append(("access", repository, number)) return {"authored": True, "assigned": False} async def excerpt(repository, number, head_sha, run_id, job_index): calls.append(("excerpt", repository, number, head_sha, run_id, job_index)) return { "head_sha": head_sha, "run_id": run_id, "job_index": job_index, "name": "CI / lint", "excerpt": "AssertionError: failed", } monkeypatch.setattr(main, "_pull_workspace_capabilities", capabilities) monkeypatch.setattr(main.gitea_proxy, "action_failure_excerpt", excerpt, raising=False) 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/checks/91/jobs/3/failure", params={"expected_head_sha": "abc1234"}, ) assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" assert response.json()["excerpt"] == "AssertionError: failed" assert calls == [ ("access", "stackchain/api", 7), ("excerpt", "stackchain/api", 7, "abc1234", 91, 3), ] @pytest.mark.anyio async def test_retry_failed_action_job_validates_head_and_uses_gitea_csrf(monkeypatch): monkeypatch.setattr(gitea_proxy, "GITEA_URL", "http://test/git") requests = [] async def handler(request): requests.append((request.method, request.url.path, request.headers.get("x-csrf-token"))) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={"head": {"sha": "abc1234"}}) if request.url.path.endswith("/commits/abc1234/status"): return httpx.Response(200, json={"statuses": [{ "context": "CI / lint (pull_request)", "status": "failure", "target_url": "/git/stackchain/api/actions/runs/91/jobs/3", }]}) if request.method == "GET" and request.url.path.endswith("/actions/runs/91/jobs/3"): return httpx.Response(200, text="") if request.method == "POST" and request.url.path.endswith("/actions/runs/91/jobs/3/rerun"): assert request.headers["x-csrf-token"] == "csrf-123" assert b"_csrf=csrf-123" in request.content return httpx.Response(303, headers={"location": "/git/stackchain/api/actions/runs/92"}) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.retry_action_job( "stackchain/api", 7, "abc1234", 91, 3 ) finally: await gitea_proxy.stop_client() assert result == { "head_sha": "abc1234", "run_id": 91, "job_index": 3, "status": "queued", } assert requests == [ ("GET", "/git/api/v1/repos/stackchain/api/pulls/7", None), ("GET", "/git/api/v1/repos/stackchain/api/commits/abc1234/status", None), ("GET", "/git/stackchain/api/actions/runs/91/jobs/3", None), ("POST", "/git/stackchain/api/actions/runs/91/jobs/3/rerun", "csrf-123"), ] @pytest.mark.anyio async def test_retry_action_endpoint_requires_workspace_access_and_exact_head(monkeypatch): calls = [] async def capabilities(repository, number): calls.append(("access", repository, number)) return {"authored": True, "assigned": False} async def retry(repository, number, head_sha, run_id, job_index): calls.append(("retry", repository, number, head_sha, run_id, job_index)) return { "head_sha": head_sha, "run_id": run_id, "job_index": job_index, "status": "queued", } monkeypatch.setattr(main, "_pull_workspace_capabilities", capabilities) monkeypatch.setattr(main.gitea_proxy, "retry_action_job", retry, raising=False) 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/checks/91/jobs/3/retry", json={"expected_head_sha": "abc1234"}, ) assert response.status_code == 202 assert response.headers["cache-control"] == "no-store" assert response.json()["status"] == "queued" assert calls == [ ("access", "stackchain/api", 7), ("retry", "stackchain/api", 7, "abc1234", 91, 3), ] @pytest.mark.anyio async def test_gitea_assigned_pull_review_includes_bounded_diff_previews(): async def handler(request): path = request.url.path if path.endswith("/pulls/7"): return httpx.Response(200, json={ "number": 7, "title": "Review this patch", "state": "open", "mergeable": True, "head": {"sha": "abc123"}, "requested_reviewers": [{"login": "casey"}], }) if path.endswith("/pulls/7/reviews"): return httpx.Response(200, json=[ {"id": 8, "user": {"login": "sam"}, "state": "APPROVED", "commit_id": "abc123"}, ]) if path.endswith("/pulls/7/files"): return httpx.Response(200, json=[ {"filename": "src/api.py", "status": "modified", "additions": 1, "deletions": 1}, {"filename": "static/logo.png", "status": "modified"}, ]) if path.endswith("/commits/abc123/status"): return httpx.Response(200, json={"state": "success"}) if path.endswith("/issues/7/comments"): return httpx.Response(200, json=[]) if path.endswith("/pulls/7.diff"): return httpx.Response(200, text=( "diff --git a/src/api.py b/src/api.py\n" "--- a/src/api.py\n+++ b/src/api.py\n" "@@ -1 +1 @@\n-old\n+new\n" "diff --git a/static/logo.png b/static/logo.png\n" "Binary files a/static/logo.png and b/static/logo.png differ\n" )) raise AssertionError(f"unexpected request: {request.method} {path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: detail = await gitea_proxy.pull_completion_review("stackchain/api", 7) finally: await gitea_proxy.stop_client() assert detail["files"][0]["diff_available"] is True assert "+new" in detail["files"][0]["diff_lines"] assert detail["files"][1]["diff_binary"] is True assert detail["files"][1]["diff_available"] is False assert detail["reviewers"] == [ {"login": "casey", "status": "waiting", "head_sha": "abc123", "blocking": True}, {"review_id": 8, "login": "sam", "status": "approved", "head_sha": "abc123", "blocking": False}, ] assert "conversation" not in detail @pytest.mark.anyio async def test_assigned_pull_review_does_not_block_on_inline_feedback(): requests = [] async def handler(request): path = request.url.path requests.append(path) if path.endswith("/pulls/7"): return httpx.Response(200, json={ "state": "open", "mergeable": True, "head": {"sha": "abc123"}, "requested_reviewers": [], }) if path.endswith("/pulls/7/reviews"): return httpx.Response(200, json=[ {"id": 7, "user": {"login": "sam"}, "state": "COMMENT", "commit_id": "old-head"}, {"id": 8, "user": {"login": "sam"}, "state": "REQUEST_CHANGES", "commit_id": "abc123", "body": "Please handle the empty state."}, ]) if path.endswith("/pulls/7/files"): return httpx.Response(200, json=[{"filename": "src/api.py", "status": "modified"}]) if path.endswith("/commits/abc123/status"): return httpx.Response(200, json={"state": "success"}) if path.endswith("/pulls/7.diff"): return httpx.Response(200, text="") raise AssertionError(f"optional feedback must not block core review: {request.method} {path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: detail = await gitea_proxy.pull_completion_review("stackchain/api", 7) finally: await gitea_proxy.stop_client() assert not any(path.endswith("/comments") for path in requests) assert detail["reviewers"] == [{ "review_id": 8, "login": "sam", "status": "changes_requested", "head_sha": "abc123", "blocking": True, "summary": "Please handle the empty state.", }] @pytest.mark.anyio async def test_pull_review_feedback_loads_one_current_bounded_review(): requests = [] async def handler(request): path = request.url.path requests.append(path) if path.endswith("/pulls/7"): return httpx.Response(200, json={"head": {"sha": "abc123"}}) if path.endswith("/pulls/7/reviews"): return httpx.Response(200, json=[ {"id": 7, "user": {"login": "sam"}, "state": "COMMENT", "commit_id": "old-head"}, {"id": 8, "user": {"login": "sam"}, "state": "REQUEST_CHANGES", "commit_id": "abc123"}, ]) if path.endswith("/pulls/7/reviews/8/comments"): return httpx.Response(200, json=[ {"id": 99, "path": "src/api.py", "body": "Return before parsing.", "new_position": 12}, ]) raise AssertionError(f"unexpected request: {request.method} {path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: feedback = await gitea_proxy.pull_review_feedback( "stackchain/api", 7, 8, "abc123" ) finally: await gitea_proxy.stop_client() assert requests[-1].endswith("/pulls/7/reviews/8/comments") assert feedback == { "review_id": 8, "head_sha": "abc123", "reviewed_head_sha": "abc123", "comments": [{"id": 99, "path": "src/api.py", "body": "Return before parsing.", "line": 12}], } @pytest.mark.anyio async def test_assigned_pull_conversation_endpoint_reuses_issue_thread_with_pull_authorization(monkeypatch): calls = [] async def assigned(repository, number): calls.append(("assigned", repository, number)) return True async def conversation(repository, number, page, limit): calls.append(("conversation", repository, number, page, limit)) return {"comments": [{"id": 41}], "page": 3, "older_page": 2, "total": 47} monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "issue_conversation_page", conversation) 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/comments?page=3&limit=20" ) assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" assert response.json() == { "comments": [{"id": 41}], "page": 3, "older_page": 2, "total": 47 } assert calls == [ ("assigned", "stackchain/api", 7), ("conversation", "stackchain/api", 7, 3, 20), ] @pytest.mark.anyio async def test_assigned_pull_comment_posts_only_after_assignment_check(monkeypatch): calls = [] async def assigned(repository, number): return True async def comment(repository, number, body): calls.append((repository, number, body)) return {"id": 91, "author": "timmy", "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) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.post( "/api/v1/repos/stackchain/api/pulls/7/comments", json={"body": " Ready to ship. "}, ) assert response.status_code == 201 assert calls == [("stackchain/api", 7, "Ready to ship.")] @pytest.mark.anyio async def test_assigned_pull_merge_requires_current_eligible_head(monkeypatch): calls = [] async def assigned(repository, number): return True async def merge(repository, number, expected_head_sha): calls.append((repository, number, expected_head_sha)) return {"number": number, "merged": True, "state": "closed"} monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge, raising=False) 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/merge", json={"expected_head_sha": "abc123"}, ) assert response.status_code == 200 assert response.json() == {"number": 7, "merged": True, "state": "closed"} assert calls == [("stackchain/api", 7, "abc123")] @pytest.mark.anyio async def test_assigned_pull_merge_fails_before_gitea_when_audit_reservation_fails( monkeypatch, ): merge_calls = [] class UnavailableJournal: def reserve(self, *_args, **_kwargs): raise SecurityEventStoreError("unavailable") async def assigned(*_args): return True async def merge(*args): merge_calls.append(args) return {"number": 7, "merged": True, "state": "closed"} monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal()) monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge) 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/merge", json={"expected_head_sha": "abc123"}, ) assert response.status_code == 503 assert response.json() == { "error": "Security activity is temporarily unavailable. Nothing was merged." } assert merge_calls == [] @pytest.mark.anyio async def test_assigned_pull_merge_reports_success_and_retains_pending_audit_when_finalize_fails( monkeypatch, ): lifecycle = [] class InterruptedJournal: def reserve(self, kind, *, target): lifecycle.append(("reserve", kind, target)) return "merge-operation" def finalize(self, operation_id): lifecycle.append(("finalize", operation_id)) raise SecurityEventStoreError("unavailable") def discard(self, operation_id): lifecycle.append(("discard", operation_id)) async def assigned(*_args): return True async def merge(*_args): return {"number": 7, "merged": True, "state": "closed"} monkeypatch.setattr(main, "_security_event_store", lambda: InterruptedJournal()) monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge) 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/merge", json={"expected_head_sha": "abc123"}, ) assert response.status_code == 200 assert response.json() == {"number": 7, "merged": True, "state": "closed"} assert lifecycle == [ ("reserve", "pull_merged", "stackchain/api#7"), ("finalize", "merge-operation"), ] @pytest.mark.anyio async def test_assigned_pull_merge_reconciles_acceptance_before_timeout(monkeypatch): calls = [] merged = False class LifecycleJournal: def reserve(self, kind, *, target): calls.append(("reserve", kind, target)) return "merge-operation" def finalize(self, operation_id): calls.append(("finalize", operation_id)) def discard(self, operation_id): calls.append(("discard", operation_id)) async def assigned(repository, number): return True async def merge(repository, number, expected_head_sha): nonlocal merged calls.append(("merge", repository, number, expected_head_sha)) merged = True await asyncio.sleep(0.05) return {"number": number, "merged": True, "state": "closed"} async def confirm(repository, number, expected_head_sha): calls.append(("confirm", repository, number, expected_head_sha)) return merged monkeypatch.setattr(main, "ISSUE_ACTION_TIMEOUT_SECONDS", 0.01) monkeypatch.setattr(main, "_security_event_store", lambda: LifecycleJournal()) monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge) monkeypatch.setattr(main.gitea_proxy, "is_pull_merged_at_head", confirm, raising=False) 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/merge", json={"expected_head_sha": "abc123"}, ) assert response.status_code == 200 assert response.json() == {"number": 7, "merged": True, "state": "closed"} assert calls == [ ("reserve", "pull_merged", "stackchain/api#7"), ("merge", "stackchain/api", 7, "abc123"), ("confirm", "stackchain/api", 7, "abc123"), ("finalize", "merge-operation"), ] @pytest.mark.anyio async def test_gitea_merge_confirmation_requires_merged_expected_head(): requests = [] async def handler(request): requests.append((request.method, request.url.path)) return httpx.Response(200, json={ "number": 7, "state": "closed", "merged": True, "head": {"sha": "abc123"}, }) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: confirmed = await gitea_proxy.is_pull_merged_at_head( "stackchain/api", 7, "abc123" ) finally: await gitea_proxy.stop_client() assert confirmed is True assert requests == [("GET", "/api/v1/repos/stackchain/api/pulls/7")] @pytest.mark.anyio async def test_assigned_pull_merge_reports_unresolved_confirmation_without_false_state(monkeypatch): calls = [] async def assigned(repository, number): return True async def merge(repository, number, expected_head_sha): calls.append(("merge", repository, number, expected_head_sha)) raise httpx.ReadTimeout("upstream response was lost") async def confirm(repository, number, expected_head_sha): calls.append(("confirm", repository, number, expected_head_sha)) return False monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge) monkeypatch.setattr(main.gitea_proxy, "is_pull_merged_at_head", confirm) 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/merge", json={"expected_head_sha": "abc123"}, ) assert response.status_code == 202 assert response.json() == { "number": 7, "merged": False, "state": "unknown", "confirmation_pending": True, "error": "Merge confirmation is pending. Check its status before retrying.", } assert calls == [ ("merge", "stackchain/api", 7, "abc123"), ("confirm", "stackchain/api", 7, "abc123"), ] @pytest.mark.anyio async def test_assigned_pull_merge_returns_conflict_without_mutating_stale_head(monkeypatch): async def assigned(repository, number): return True async def merge(*args): raise gitea_proxy.StalePullError("changed") monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge, raising=False) 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/merge", json={"expected_head_sha": "abc123"}, ) assert response.status_code == 409 assert "New commits" in response.json()["error"] @pytest.mark.anyio @pytest.mark.parametrize( "rejection", [gitea_proxy.StalePullError("changed"), gitea_proxy.PullNotMergeableError("unsafe")], ) async def test_assigned_pull_merge_discards_audit_reservation_after_definite_rejection( monkeypatch, rejection ): lifecycle = [] class LifecycleJournal: def reserve(self, kind, *, target): lifecycle.append(("reserve", kind, target)) return "merge-operation" def discard(self, operation_id): lifecycle.append(("discard", operation_id)) async def assigned(*_args): return True async def merge(*_args): raise rejection monkeypatch.setattr(main, "_security_event_store", lambda: LifecycleJournal()) monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge) 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/merge", json={"expected_head_sha": "abc123"}, ) assert response.status_code == 409 assert lifecycle == [ ("reserve", "pull_merged", "stackchain/api#7"), ("discard", "merge-operation"), ] @pytest.mark.anyio async def test_gitea_merge_rejects_failed_ci_without_upstream_mutation(): requests = [] async def handler(request): requests.append((request.method, request.url.path)) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "number": 7, "state": "open", "draft": False, "mergeable": True, "merged": False, "head": {"sha": "abc123"}, }) return httpx.Response(200, json={"state": "failure"}) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: with pytest.raises(gitea_proxy.PullNotMergeableError): await gitea_proxy.merge_assigned_pull("stackchain/api", 7, "abc123") finally: await gitea_proxy.stop_client() assert requests == [ ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ("GET", "/api/v1/repos/stackchain/api/commits/abc123/status"), ] @pytest.mark.anyio @pytest.mark.parametrize("requested,reviews", [ ([{"login": "casey"}], []), ([], [{"id": 9, "user": {"login": "casey"}, "state": "REQUEST_CHANGES", "commit_id": "abc123"}]), ]) async def test_gitea_merge_rejects_unresolved_reviewer_status_without_mutation(requested, reviews): requests = [] async def handler(request): requests.append((request.method, request.url.path)) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "number": 7, "state": "open", "draft": False, "mergeable": True, "merged": False, "head": {"sha": "abc123"}, "requested_reviewers": requested, }) if request.url.path.endswith("/commits/abc123/status"): return httpx.Response(200, json={"state": "success"}) if request.url.path.endswith("/pulls/7/reviews"): return httpx.Response(200, json=reviews) raise AssertionError(f"unexpected mutation: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: with pytest.raises(gitea_proxy.PullNotMergeableError): await gitea_proxy.merge_assigned_pull("stackchain/api", 7, "abc123") finally: await gitea_proxy.stop_client() assert requests[-1] == ("GET", "/api/v1/repos/stackchain/api/pulls/7/reviews") assert all(method == "GET" for method, _path in requests) @pytest.mark.anyio async def test_assigned_pull_ownership_endpoints_list_handoff_and_release(monkeypatch): calls = [] async def assigned(repository, number): calls.append(("assigned", repository, number)) return True async def candidates(repository): calls.append(("candidates", repository)) return [{"login": "alex", "name": "Alexander"}] async def handoff(repository, number, recipient): calls.append(("handoff", repository, number, recipient)) return {"repository": repository, "number": number, "assignees": [recipient], "recipient": recipient} async def release(repository, number): calls.append(("release", repository, number)) return {"repository": repository, "number": number, "assignees": []} monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "pull_handoff_candidates", candidates, raising=False) monkeypatch.setattr(main.gitea_proxy, "handoff_assigned_pull", handoff, raising=False) monkeypatch.setattr(main.gitea_proxy, "release_assigned_pull", release, raising=False) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: listed = await client.get("/api/v1/repos/stackchain/api/pulls/7/handoff-candidates") transferred = await client.patch( "/api/v1/repos/stackchain/api/pulls/7/handoff", json={"recipient": "alex"} ) released = await client.patch("/api/v1/repos/stackchain/api/pulls/7/release") assert listed.status_code == 200 assert listed.headers["cache-control"] == "no-store" assert listed.json() == [{"login": "alex", "name": "Alexander"}] assert transferred.status_code == 200 assert transferred.json()["recipient"] == "alex" assert released.status_code == 200 assert released.json()["assignees"] == [] assert calls == [ ("assigned", "stackchain/api", 7), ("candidates", "stackchain/api"), ("handoff", "stackchain/api", 7, "alex"), ("release", "stackchain/api", 7), ] @pytest.mark.anyio async def test_assigned_pull_review_request_endpoints_list_and_submit(monkeypatch): calls = [] async def assigned(repository, number): calls.append(("assigned", repository, number)) return True async def candidates(repository, number): calls.append(("candidates", repository, number)) return [{"login": "casey", "name": "Casey"}] async def request_review(repository, number, reviewer, expected_head_sha): calls.append(("request", repository, number, reviewer, expected_head_sha)) return {"number": number, "reviewer": reviewer, "requested_reviewers": [reviewer]} monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "pull_review_candidates", candidates, raising=False) monkeypatch.setattr(main.gitea_proxy, "request_assigned_pull_review", request_review, raising=False) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: listed = await client.get("/api/v1/repos/stackchain/api/pulls/7/review-candidates") requested = await client.post( "/api/v1/repos/stackchain/api/pulls/7/request-review", json={"reviewer": "casey", "expected_head_sha": "abc1234"}, ) assert listed.status_code == 200 assert listed.headers["cache-control"] == "no-store" assert listed.json() == [{"login": "casey", "name": "Casey"}] assert requested.status_code == 200 assert requested.json()["requested_reviewers"] == ["casey"] assert calls == [ ("assigned", "stackchain/api", 7), ("candidates", "stackchain/api", 7), ("request", "stackchain/api", 7, "casey", "abc1234"), ] @pytest.mark.anyio async def test_assigned_pull_cancel_review_endpoint_forwards_head_scoped_request(monkeypatch): calls = [] async def cancel(repository, number, reviewer, expected_head_sha): calls.append((repository, number, reviewer, expected_head_sha)) return { "repository": repository, "number": number, "reviewer": reviewer, "head_sha": expected_head_sha, "requested_reviewers": ["casey"], } monkeypatch.setattr(main.gitea_proxy, "cancel_assigned_pull_review", cancel, raising=False) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.request( "DELETE", "/api/v1/repos/stackchain/api/pulls/7/request-review", json={"reviewer": "sam", "expected_head_sha": "abc1234"}, ) assert response.status_code == 200 assert response.json()["requested_reviewers"] == ["casey"] assert calls == [("stackchain/api", 7, "sam", "abc1234")] @pytest.mark.anyio async def test_gitea_pull_handoff_preserves_coassignees_and_confirms_ownership_exit(): requests = [] async def handler(request): requests.append((request.method, request.url.path, request.content)) if request.url.path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "number": 7, "state": "open", "merged": False, "assignees": [{"login": "timmy"}, {"login": "sam"}], }) if request.url.path.endswith("/assignees"): return httpx.Response(200, json=[ {"login": "timmy"}, {"login": "sam"}, {"login": "alex", "full_name": "Alexander"}, ]) if request.method == "PATCH" and request.url.path.endswith("/issues/7"): assert request.read() return httpx.Response(200, json={ "number": 7, "state": "open", "assignees": [{"login": "sam"}, {"login": "alex"}], }) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.handoff_assigned_pull("stackchain/api", 7, "alex") finally: await gitea_proxy.stop_client() assert result == { "repository": "stackchain/api", "number": 7, "state": "open", "assignees": ["sam", "alex"], "recipient": "alex", } assert [(method, path) for method, path, _body in requests] == [ ("GET", "/api/v1/user"), ("GET", "/api/v1/repos/stackchain/api/pulls/7"), ("GET", "/api/v1/user"), ("GET", "/api/v1/repos/stackchain/api/assignees"), ("PATCH", "/api/v1/repos/stackchain/api/issues/7"), ] @pytest.mark.anyio async def test_pull_handoff_candidates_are_bounded_and_exclude_invalid_or_current_logins(): async def handler(request): if request.url.path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if request.url.path.endswith("/assignees"): return httpx.Response(200, json=( [{"login": "timmy"}, {"login": "bad login"}, {"login": ""}] + [{"login": f"user-{index:02d}"} for index in range(30)] )) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: candidates = await gitea_proxy.pull_handoff_candidates("stackchain/api") finally: await gitea_proxy.stop_client() assert len(candidates) == 25 assert [candidate["login"] for candidate in candidates] == [ f"user-{index:02d}" for index in range(25) ] @pytest.mark.anyio async def test_gitea_authored_unassigned_pull_review_request_filters_candidates_and_confirms_reviewer(): requests = [] review_requested = False async def handler(request): nonlocal review_requested requests.append((request.method, request.url.path, request.content)) if request.url.path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if request.url.path.endswith("/assignees"): return httpx.Response(200, json=[ {"login": "timmy", "full_name": "Timmy"}, {"login": "alex", "full_name": "Alexander"}, {"login": "sam", "full_name": "Sam"}, {"login": "casey", "full_name": "Casey"}, ]) if request.url.path.endswith("/pulls/7"): requested = [{"login": "sam"}] if review_requested: requested.append({"login": "casey"}) return httpx.Response(200, json={ "number": 7, "state": "open", "merged": False, "head": {"sha": "abc123"}, "user": {"login": "timmy"}, "assignees": [], "requested_reviewers": requested, }) if request.method == "POST" and request.url.path.endswith("/requested_reviewers"): review_requested = True return httpx.Response(201, json={}) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: candidates = await gitea_proxy.pull_review_candidates("stackchain/api", 7) result = await gitea_proxy.request_assigned_pull_review( "stackchain/api", 7, "casey", "abc123" ) finally: await gitea_proxy.stop_client() assert candidates == [ {"login": "alex", "name": "Alexander"}, {"login": "casey", "name": "Casey"}, ] assert result == { "repository": "stackchain/api", "number": 7, "head_sha": "abc123", "requested_reviewers": ["sam", "casey"], "reviewer": "casey", } mutation = next(item for item in requests if item[0] == "POST") assert mutation[1] == "/api/v1/repos/stackchain/api/pulls/7/requested_reviewers" assert mutation[2] == b'{"reviewers":["casey"]}' @pytest.mark.anyio async def test_gitea_authored_unassigned_pull_can_cancel_pending_review(): requests = [] cancelled = False async def handler(request): nonlocal cancelled requests.append((request.method, request.url.path, request.content)) if request.url.path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "number": 7, "state": "open", "merged": False, "head": {"sha": "abc123"}, "user": {"login": "timmy"}, "assignees": [], "requested_reviewers": [{"login": "casey"}] if cancelled else [{"login": "sam"}, {"login": "casey"}], }) if request.method == "DELETE" and request.url.path.endswith("/requested_reviewers"): assert request.content == b'{"reviewers":["sam"]}' cancelled = True return httpx.Response(204) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.cancel_assigned_pull_review( "stackchain/api", 7, "sam", "abc123" ) finally: await gitea_proxy.stop_client() assert result == { "repository": "stackchain/api", "number": 7, "head_sha": "abc123", "requested_reviewers": ["casey"], "reviewer": "sam", } assert [method for method, _path, _body in requests] == ["GET", "GET", "DELETE", "GET"] @pytest.mark.anyio async def test_gitea_cancel_pending_pull_review_rejects_head_drift_before_mutation(): requests = [] async def handler(request): requests.append(request.method) if request.url.path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "number": 7, "state": "open", "merged": False, "head": {"sha": "new-head"}, "assignees": [{"login": "timmy"}], "requested_reviewers": [{"login": "sam"}], }) if request.method == "DELETE": return httpx.Response(204) raise AssertionError(request.url.path) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: with pytest.raises(gitea_proxy.IssueNotAvailableError): await gitea_proxy.cancel_assigned_pull_review( "stackchain/api", 7, "sam", "old-head" ) finally: await gitea_proxy.stop_client() assert requests == ["GET", "GET"] @pytest.mark.anyio async def test_gitea_pull_release_removes_current_login_case_insensitively(): async def handler(request): if request.url.path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "number": 7, "state": "open", "merged": False, "assignees": [{"login": "Timmy"}, {"login": "sam"}], }) if request.method == "PATCH" and request.url.path.endswith("/issues/7"): assert request.content == b'{"assignees":["sam"]}' return httpx.Response(200, json={ "number": 7, "state": "open", "assignees": [{"login": "sam"}], }) raise AssertionError(f"unexpected request: {request.method} {request.url.path}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: released = await gitea_proxy.release_assigned_pull("stackchain/api", 7) finally: await gitea_proxy.stop_client() assert released["assignees"] == ["sam"] @pytest.mark.anyio async def test_gitea_merge_returns_the_exact_merge_commit_for_release_tracking(): async def handler(request): if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "number": 7, "state": "open", "draft": False, "mergeable": True, "merged": False, "head": { "sha": "abc123", "ref": "timmy/feature", "repo": {"full_name": "stackchain/api"}, }, }) if request.url.path.endswith("/commits/abc123/status"): return httpx.Response(200, json={"state": "success"}) if request.url.path.endswith("/pulls/7/reviews"): return httpx.Response(200, json=[]) if request.url.path.endswith("/pulls/7/merge"): return httpx.Response(200, json={"merged": True, "sha": "merge456"}) raise AssertionError(request.url.path) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.merge_assigned_pull("stackchain/api", 7, "abc123") finally: await gitea_proxy.stop_client() assert result == { "number": 7, "merged": True, "state": "closed", "merge_commit_sha": "merge456", "source_branch": "timmy/feature", "source_head_sha": "abc123", "source_repository": "stackchain/api", } @pytest.mark.anyio async def test_source_branch_cleanup_endpoint_audits_and_deletes_the_exact_merged_head(monkeypatch): lifecycle = [] class Journal: def reserve(self, kind, *, target): lifecycle.append(("reserve", kind, target)) return "branch-cleanup" def finalize(self, operation_id): lifecycle.append(("finalize", operation_id)) def discard(self, operation_id): lifecycle.append(("discard", operation_id)) async def cleanup(repository, number, source_branch, expected_head_sha): lifecycle.append(("delete", repository, number, source_branch, expected_head_sha)) return { "number": number, "deleted": True, "source_branch": source_branch, "source_head_sha": expected_head_sha, } monkeypatch.setattr(main, "_security_event_store", lambda: Journal()) monkeypatch.setattr(main.gitea_proxy, "delete_merged_source_branch", cleanup, raising=False) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: response = await client.request( "DELETE", "/api/v1/repos/stackchain/api/pulls/7/source-branch", json={"source_branch": "timmy/feature", "expected_head_sha": "abc1234"}, ) assert response.status_code == 200 assert response.json()["deleted"] is True assert lifecycle == [ ("reserve", "source_branch_deleted", "stackchain/api#7@abc1234"), ("delete", "stackchain/api", 7, "timmy/feature", "abc1234"), ("finalize", "branch-cleanup"), ] @pytest.mark.anyio async def test_gitea_deletes_only_the_merged_authors_same_repository_exact_source_branch(): requests = [] async def handler(request): path = request.url.raw_path.decode() requests.append((request.method, path)) if request.url.path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "state": "closed", "merged": True, "user": {"login": "timmy"}, "head": { "ref": "timmy/feature", "sha": "abc123", "repo": {"full_name": "stackchain/api"}, }, }) if request.url.path.endswith("/repos/stackchain/api"): return httpx.Response(200, json={"default_branch": "main"}) if path.endswith("/branches/timmy%2Ffeature"): branch_reads = sum( method == "GET" and seen_path.endswith("/branches/timmy%2Ffeature") for method, seen_path in requests ) if request.method == "DELETE": return httpx.Response(204) if branch_reads == 1: return httpx.Response(200, json={ "name": "timmy/feature", "protected": False, "commit": {"id": "abc123"}, }) return httpx.Response(404) raise AssertionError(request.url.path) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.delete_merged_source_branch( "stackchain/api", 7, "timmy/feature", "abc123" ) finally: await gitea_proxy.stop_client() assert result == { "number": 7, "deleted": True, "source_branch": "timmy/feature", "source_head_sha": "abc123", } assert requests[-2:] == [ ("DELETE", "/api/v1/repos/stackchain/api/branches/timmy%2Ffeature"), ("GET", "/api/v1/repos/stackchain/api/branches/timmy%2Ffeature"), ] @pytest.mark.anyio async def test_release_receipt_matches_only_the_captured_commit_and_reports_checks(monkeypatch): async def receipt(repository, commit_sha): assert (repository, commit_sha) == ("stackchain/api", "merge456") return { "commit_sha": "merge456", "ci_state": "pending", "checks": [{"name": "browser", "state": "pending", "url": ""}], "release": None, } monkeypatch.setattr(main.gitea_proxy, "release_receipt_status", receipt, raising=False) 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/release-receipt/merge456" ) assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" assert response.json()["checks"] == [ {"name": "browser", "state": "pending", "url": ""} ] assert response.json()["release"] is None @pytest.mark.anyio async def test_release_receipt_preserves_only_safe_actions_recovery_for_failed_exact_commit(monkeypatch): monkeypatch.setattr(gitea_proxy, "GITEA_URL", "https://forge.example/git") async def handler(request): if request.url.path.endswith("/commits/merge456/status"): return httpx.Response(200, json={ "state": "failure", "statuses": [ { "context": "CI / browser (push)", "status": "failure", "description": "Playwright failed", "target_url": "/git/stackchain/api/actions/runs/91/jobs/3", }, { "context": "external", "status": "failure", "description": "External check failed", "target_url": "https://checks.example/jobs/4", }, ], }) if request.url.path.endswith("/releases"): return httpx.Response(200, json=[]) raise AssertionError(request.url.path) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.release_receipt_status("stackchain/api", "merge456") finally: await gitea_proxy.stop_client() assert result["checks"] == [ { "name": "CI / browser (push)", "state": "failure", "description": "Playwright failed", "url": "https://forge.example/git/stackchain/api/actions/runs/91/jobs/3", "recovery": {"run_id": 91, "job_index": 3}, }, { "name": "external", "state": "failure", "description": "External check failed", "url": "", }, ] @pytest.mark.anyio async def test_release_action_failure_excerpt_is_bound_to_exact_merge_commit(monkeypatch): monkeypatch.setattr(gitea_proxy, "GITEA_URL", "http://test/git") requests = [] async def handler(request): requests.append((request.method, request.url.path)) if request.url.path.endswith("/commits/merge456/status"): return httpx.Response(200, json={"statuses": [{ "context": "CI / browser (push)", "status": "failure", "target_url": "/git/stackchain/api/actions/runs/91/jobs/3", }]}) if request.url.path.endswith("/actions/runs/91/jobs/3/logs"): return httpx.Response(200, text="GITEA_TOKEN=secret\nAssertionError: mobile journey failed") raise AssertionError(request.url.path) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.release_action_failure_excerpt( "stackchain/api", "merge456", 91, 3 ) finally: await gitea_proxy.stop_client() assert result == { "commit_sha": "merge456", "run_id": 91, "job_index": 3, "name": "CI / browser (push)", "excerpt": "GITEA_TOKEN=[redacted]\nAssertionError: mobile journey failed", } assert requests == [ ("GET", "/git/api/v1/repos/stackchain/api/commits/merge456/status"), ("GET", "/git/stackchain/api/actions/runs/91/jobs/3/logs"), ] @pytest.mark.anyio async def test_retry_release_action_job_revalidates_failed_exact_commit(monkeypatch): monkeypatch.setattr(gitea_proxy, "GITEA_URL", "http://test/git") requests = [] async def handler(request): requests.append((request.method, request.url.path)) if request.url.path.endswith("/commits/merge456/status"): return httpx.Response(200, json={"statuses": [{ "context": "CI / browser (push)", "status": "failure", "target_url": "/git/stackchain/api/actions/runs/91/jobs/3", }]}) if request.method == "GET" and request.url.path.endswith("/actions/runs/91/jobs/3"): return httpx.Response(200, text="") if request.method == "POST" and request.url.path.endswith("/actions/runs/91/jobs/3/rerun"): assert request.headers["x-csrf-token"] == "csrf-123" return httpx.Response(303) raise AssertionError(request.url.path) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.retry_release_action_job( "stackchain/api", "merge456", 91, 3 ) finally: await gitea_proxy.stop_client() assert result == { "commit_sha": "merge456", "run_id": 91, "job_index": 3, "status": "queued" } assert requests == [ ("GET", "/git/api/v1/repos/stackchain/api/commits/merge456/status"), ("GET", "/git/stackchain/api/actions/runs/91/jobs/3"), ("POST", "/git/stackchain/api/actions/runs/91/jobs/3/rerun"), ] @pytest.mark.anyio async def test_release_failure_recovery_endpoints_authorize_pull_and_bind_merge_commit(monkeypatch): calls = [] async def release_access(repository, number, commit_sha): calls.append(("access", repository, number, commit_sha)) return True async def excerpt(repository, commit_sha, run_id, job_index): calls.append(("excerpt", repository, commit_sha, run_id, job_index)) return { "commit_sha": commit_sha, "run_id": run_id, "job_index": job_index, "name": "CI / browser", "excerpt": "AssertionError: failed", } async def retry(repository, commit_sha, run_id, job_index): calls.append(("retry", repository, commit_sha, run_id, job_index)) return { "commit_sha": commit_sha, "run_id": run_id, "job_index": job_index, "status": "queued", } monkeypatch.setattr(main.gitea_proxy, "can_recover_merged_release", release_access, raising=False) monkeypatch.setattr(main.gitea_proxy, "release_action_failure_excerpt", excerpt, raising=False) monkeypatch.setattr(main.gitea_proxy, "retry_release_action_job", retry, raising=False) path = "/api/v1/repos/stackchain/api/pulls/7/release-receipt/abc1234/checks/91/jobs/3" transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: failure = await client.get(path + "/failure") retried = await client.post(path + "/retry") assert failure.status_code == 200 assert failure.headers["cache-control"] == "no-store" assert failure.json()["commit_sha"] == "abc1234" assert retried.status_code == 202 assert retried.headers["cache-control"] == "no-store" assert retried.json()["status"] == "queued" assert calls == [ ("access", "stackchain/api", 7, "abc1234"), ("excerpt", "stackchain/api", "abc1234", 91, 3), ("access", "stackchain/api", 7, "abc1234"), ("retry", "stackchain/api", "abc1234", 91, 3), ] @pytest.mark.anyio async def test_merged_release_recovery_access_requires_exact_commit_and_participant(): async def handler(request): if request.url.path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "state": "closed", "merged": True, "merge_commit_sha": "abc1234", "user": {"login": "timmy"}, "assignees": [], }) raise AssertionError(request.url.path) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: exact = await gitea_proxy.can_recover_merged_release("stackchain/api", 7, "abc1234") other = await gitea_proxy.can_recover_merged_release("stackchain/api", 7, "def5678") finally: await gitea_proxy.stop_client() assert exact is True assert other is False @pytest.mark.anyio async def test_gitea_release_receipt_ignores_other_commits_and_normalizes_matching_assets(): forge = gitea_proxy.GITEA_URL.rstrip("/") async def handler(request): if request.url.path.endswith("/commits/merge456/status"): return httpx.Response(200, json={ "state": "success", "statuses": [{"context": "browser", "status": "success", "target_url": ""}], }) if request.url.path.endswith("/releases"): return httpx.Response(200, json=[ {"tag_name": "newer", "target_commitish": "other789", "html_url": f"{forge}/stackchain/api/releases/tag/newer"}, { "tag_name": "rc-42", "target_commitish": "merge456", "html_url": f"{forge}/stackchain/api/releases/tag/rc-42", "assets": [ {"name": "manifest.json", "browser_download_url": f"{forge}/stackchain/api/releases/download/rc-42/manifest.json"}, ], }, ]) raise AssertionError(request.url.path) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.release_receipt_status("stackchain/api", "merge456") finally: await gitea_proxy.stop_client() assert result["commit_sha"] == "merge456" assert result["ci_state"] == "success" assert result["checks"] == [{"name": "browser", "state": "success", "url": ""}] assert result["release"] == { "tag": "rc-42", "url": f"{forge}/stackchain/api/releases/tag/rc-42", "assets": [{ "name": "manifest.json", "url": f"{forge}/stackchain/api/releases/download/rc-42/manifest.json", }], } @pytest.mark.anyio async def test_source_branch_cleanup_resolves_a_lost_delete_response_with_one_absence_check(): requests = [] async def handler(request): raw_path = request.url.raw_path.decode() requests.append((request.method, raw_path)) if request.url.path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if request.url.path.endswith("/pulls/7"): return httpx.Response(200, json={ "state": "closed", "merged": True, "user": {"login": "timmy"}, "head": {"ref": "timmy/feature", "sha": "abc123", "repo": {"full_name": "stackchain/api"}}, }) if request.url.path.endswith("/repos/stackchain/api"): return httpx.Response(200, json={"default_branch": "main"}) if raw_path.endswith("/branches/timmy%2Ffeature"): if request.method == "DELETE": raise httpx.ReadTimeout("delete response was lost") reads = sum(method == "GET" and path.endswith("/branches/timmy%2Ffeature") for method, path in requests) return httpx.Response(200, json={"protected": False, "commit": {"id": "abc123"}}) if reads == 1 else httpx.Response(404) raise AssertionError(raw_path) gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.delete_merged_source_branch( "stackchain/api", 7, "timmy/feature", "abc123" ) finally: await gitea_proxy.stop_client() assert result["deleted"] is True assert sum(method == "GET" and path.endswith("/branches/timmy%2Ffeature") for method, path in requests) == 2 @pytest.mark.anyio async def test_prepare_release_rollback_creates_one_atomic_draft_pull_from_exact_merge_parent(): requests = [] branch_created = False merge_sha = "abc1234def5678" parent_sha = "1111111aaaaaaa" current_sha = "2222222bbbbbbb" rollback_sha = "3333333ccccccc" branch = "timmy/rollback-7-abc1234def56" async def handler(request): nonlocal branch_created path = request.url.raw_path.decode().split("?", 1)[0] requests.append((request.method, path)) if path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if path.endswith("/pulls/7") and request.method == "GET": return httpx.Response(200, json={ "number": 7, "state": "closed", "merged": True, "merge_commit_sha": merge_sha, "title": "Ship release", "user": {"login": "timmy"}, "assignees": [], }) if path.endswith("/repos/stackchain/api"): return httpx.Response(200, json={ "default_branch": "main", "permissions": {"push": True}, }) if path.endswith(f"/git/commits/{merge_sha}"): return httpx.Response(200, json={ "sha": merge_sha, "parents": [{"sha": parent_sha}, {"sha": "feature-parent"}], "files": [{"filename": "app.txt", "status": "modified"}], }) if path.endswith("/contents/app.txt"): ref = request.url.params.get("ref") if ref == merge_sha: return httpx.Response(200, json=_content_payload("broken\n", "merge-blob")) if ref == parent_sha: return httpx.Response(200, json=_content_payload("working\n", "parent-blob")) if ref == "main": return httpx.Response(200, json=_content_payload("broken\n", "current-blob")) if path.endswith("/branches/timmy%2Frollback-7-abc1234def56"): if request.method == "GET": return httpx.Response( 200, json={"commit": {"id": rollback_sha}} ) if branch_created else httpx.Response(404) if path.endswith("/contents") and request.method == "POST": body = json.loads(request.content) assert body == { "branch": "main", "new_branch": branch, "message": f"Revert {merge_sha} from pull #7", "files": [{ "operation": "update", "path": "app.txt", "sha": "current-blob", "content": base64.b64encode(b"working\n").decode(), }], } branch_created = True return httpx.Response(201, json={"commit": {"sha": rollback_sha}}) if path.endswith("/pulls") and request.method == "GET": return httpx.Response(200, json=[]) if path.endswith("/pulls") and request.method == "POST": body = json.loads(request.content) assert body["head"] == branch assert body["base"] == "main" assert body["draft"] is True assert f"Rollback of #{7} at `{merge_sha}`" in body["body"] return httpx.Response(201, json={ "number": 8, "title": body["title"], "body": body["body"], "state": "open", "draft": True, "head": {"ref": branch, "sha": rollback_sha}, "base": {"ref": "main"}, "user": {"login": "timmy"}, "html_url": "https://forge.example/git/stackchain/api/pulls/8", }) raise AssertionError(f"unexpected request: {request.method} {request.url}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.prepare_release_rollback( "stackchain/api", 7, merge_sha ) finally: await gitea_proxy.stop_client() assert result["number"] == 8 assert result["draft"] is True assert result["head"] == {"ref": branch, "sha": rollback_sha} assert result["rollback_of"] == merge_sha assert result["files_changed"] == 1 assert sum(method == "POST" and path.endswith("/contents") for method, path in requests) == 1 assert sum(method == "POST" and path.endswith("/pulls") for method, path in requests) == 1 @pytest.mark.anyio async def test_release_rollback_endpoint_requires_failed_exact_commit_and_audits_draft_pull(monkeypatch): lifecycle = [] class Journal: def reserve(self, kind, *, target): lifecycle.append(("reserve", kind, target)) return "rollback-operation" def finalize(self, operation_id): lifecycle.append(("finalize", operation_id)) def discard(self, operation_id): lifecycle.append(("discard", operation_id)) async def status(repository, commit_sha): lifecycle.append(("status", repository, commit_sha)) return {"commit_sha": commit_sha, "ci_state": "failure", "release": None} async def prepare(repository, number, commit_sha): lifecycle.append(("prepare", repository, number, commit_sha)) return { "repository": repository, "number": 9, "state": "open", "draft": True, "head": {"ref": "timmy/rollback-7-abc1234", "sha": "def5678"}, "base": {"ref": "main"}, "url": "https://forge.example/git/stackchain/api/pulls/9", "rollback_of": commit_sha, "files_changed": 2, "existing": False, } monkeypatch.setattr(main, "_security_event_store", lambda: Journal()) monkeypatch.setattr(main.gitea_proxy, "release_receipt_status", status) monkeypatch.setattr(main.gitea_proxy, "prepare_release_rollback", prepare, raising=False) 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/release-receipt/abc1234/rollback" ) assert response.status_code == 201 assert response.headers["cache-control"] == "no-store" assert response.json()["rollback_of"] == "abc1234" assert lifecycle == [ ("status", "stackchain/api", "abc1234"), ("reserve", "release_rollback_prepared", "stackchain/api#7@abc1234"), ("prepare", "stackchain/api", 7, "abc1234"), ("finalize", "rollback-operation"), ] @pytest.mark.anyio async def test_prepare_release_rollback_reconciles_a_concurrent_branch_creation_without_duplicate_commit(): merge_sha = "abc1234def5678" rollback_sha = "3333333ccccccc" branch_reads = 0 pull_created = False async def handler(request): nonlocal branch_reads, pull_created path = request.url.raw_path.decode().split("?", 1)[0] if path.endswith("/user"): return httpx.Response(200, json={"login": "timmy"}) if path.endswith("/pulls/7"): return httpx.Response(200, json={ "state": "closed", "merged": True, "merge_commit_sha": merge_sha, "title": "Ship release", "user": {"login": "timmy"}, "assignees": [], }) if path.endswith("/repos/stackchain/api"): return httpx.Response(200, json={"default_branch": "main", "permissions": {"push": True}}) if path.endswith(f"/git/commits/{merge_sha}"): return httpx.Response(200, json={ "sha": merge_sha, "parents": [{"sha": "parent123"}, {"sha": "feature123"}], "files": [{"filename": "app.txt", "status": "modified"}], }) if path.endswith("/contents/app.txt"): ref = request.url.params["ref"] return httpx.Response(200, json=_content_payload( "working\n" if ref == "parent123" else "broken\n", "current-blob" if ref == "main" else ref + "-blob", )) if path.endswith("/branches/timmy%2Frollback-7-abc1234def56"): branch_reads += 1 return httpx.Response(404) if branch_reads == 1 else httpx.Response( 200, json={"commit": {"id": rollback_sha}} ) if path.endswith("/contents") and request.method == "POST": return httpx.Response(422, json={"message": "branch already exists"}) if path.endswith("/pulls") and request.method == "GET": return httpx.Response(200, json=[]) if path.endswith("/pulls") and request.method == "POST": pull_created = True body = json.loads(request.content) return httpx.Response(201, json={ "number": 9, "title": body["title"], "body": body["body"], "state": "open", "draft": True, "head": {"ref": body["head"], "sha": rollback_sha}, "base": {"ref": body["base"]}, "user": {"login": "timmy"}, "html_url": "https://forge.example/git/stackchain/api/pulls/9", }) raise AssertionError(f"unexpected request {request.method} {request.url}") gitea_proxy.start_client(transport=httpx.MockTransport(handler)) try: result = await gitea_proxy.prepare_release_rollback("stackchain/api", 7, merge_sha) finally: await gitea_proxy.stop_client() assert result["number"] == 9 assert result["head"]["sha"] == rollback_sha assert branch_reads == 3 # initial check, race reconciliation, pull-creation revalidation assert pull_created is True