150 lines
5.6 KiB
Python
150 lines
5.6 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy, main
|
|
|
|
|
|
@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_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_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
|
|
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"),
|
|
]
|