2709 lines
107 KiB
Python
2709 lines
107 KiB
Python
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_comments_expose_one_bounded_line_suggestion_only():
|
|
comments = [
|
|
{
|
|
"id": 1,
|
|
"path": "src/api.py",
|
|
"new_position": 4,
|
|
"body": "Handle the empty state:\n```suggestion\nreturn empty_result()\n```",
|
|
},
|
|
{
|
|
"id": 2,
|
|
"path": "src/api.py",
|
|
"new_position": 7,
|
|
"body": "Choose one:\n```suggestion\nfirst()\n```\n```suggestion\nsecond()\n```",
|
|
},
|
|
{
|
|
"id": 3,
|
|
"path": "src/api.py",
|
|
"body": "No stable line:\n```suggestion\nunsafe()\n```",
|
|
},
|
|
]
|
|
|
|
normalized = gitea_proxy._normalize_review_comments(comments)
|
|
|
|
assert normalized[0]["suggestion"] == {"line": 4, "content": "return empty_result()"}
|
|
assert "suggestion" not in normalized[1]
|
|
assert "suggestion" not in normalized[2]
|
|
|
|
|
|
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="<script>csrfToken: 'csrf-123'</script>")
|
|
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="<script>csrfToken: 'csrf-123'</script>")
|
|
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
|