1186 lines
47 KiB
Python
1186 lines
47 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy, main
|
|
from src.security_event_store import SecurityEventStoreError
|
|
|
|
|
|
@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_requires_author_assignment_and_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": [{"login": "alex"}],
|
|
"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_assigned_pull_detail_reports_completion_state(monkeypatch):
|
|
async def assigned(repository, number):
|
|
return (repository, number) == ("stackchain/api", 7)
|
|
|
|
async def detail(repository, number):
|
|
assert (repository, number) == ("stackchain/api", 7)
|
|
return {
|
|
"repository": repository,
|
|
"number": number,
|
|
"title": "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, "is_assigned_pull", assigned, raising=False)
|
|
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 assigned(repository, number):
|
|
return False
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned, 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/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 = []
|
|
|
|
async def assigned(repository, number):
|
|
calls.append(("assigned", repository, number))
|
|
return True
|
|
|
|
async def review(repository, number):
|
|
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, "is_assigned_pull", assigned)
|
|
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 == [
|
|
("assigned", "stackchain/api", 7),
|
|
("review", "stackchain/api", 7),
|
|
]
|
|
|
|
|
|
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},
|
|
{"login": "lee", "status": "commented", "head_sha": "new-head", "blocking": False},
|
|
{"login": "pat", "status": "approved", "head_sha": "new-head", "blocking": False},
|
|
{"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") == [
|
|
{"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] == {
|
|
"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_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),
|
|
]
|
|
|
|
|
|
@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},
|
|
{"login": "sam", "status": "approved", "head_sha": "abc123", "blocking": False},
|
|
]
|
|
assert "conversation" not in detail
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_assigned_pull_review_includes_latest_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/reviews/8/comments"):
|
|
return httpx.Response(200, json=[
|
|
{"path": "src/api.py", "body": "Return before parsing.", "new_position": 12},
|
|
])
|
|
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"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 "/api/v1/repos/stackchain/api/pulls/7/reviews/7/comments" not in requests
|
|
assert detail["reviewers"] == [{
|
|
"login": "sam", "status": "changes_requested", "head_sha": "abc123", "blocking": True,
|
|
"summary": "Please handle the empty state.",
|
|
"comments": [{"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_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_pull_review_request_filters_candidates_and_confirms_requested_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": "alex"},
|
|
"assignees": [{"login": "timmy"}],
|
|
"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": "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_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"},
|
|
})
|
|
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"
|
|
}
|
|
|
|
|
|
@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_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",
|
|
}],
|
|
}
|