204 lines
7.5 KiB
Python
204 lines
7.5 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy
|
|
from src import main
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_pull_creation_operations():
|
|
main._authored_action_operations.clear()
|
|
main._idempotency_ledger.clear()
|
|
yield
|
|
main._authored_action_operations.clear()
|
|
main._idempotency_ledger.clear()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pull_creation_options_return_visible_repository_branches(monkeypatch):
|
|
async def access(repository):
|
|
assert repository == "stackchain/api"
|
|
return {"full_name": repository, "default_branch": "main", "permissions": {"push": True}}
|
|
|
|
async def branches(repository):
|
|
assert repository == "stackchain/api"
|
|
return [
|
|
{"name": "feature/mobile", "commit": {"id": "abc1234"}},
|
|
{"name": "main", "commit": {"id": "def5678"}},
|
|
]
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "repository_access", access)
|
|
monkeypatch.setattr(main.gitea_proxy, "repo_branches", branches, 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/pull-creation-options")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {
|
|
"default_branch": "main",
|
|
"branches": [
|
|
{"name": "feature/mobile", "sha": "abc1234"},
|
|
{"name": "main", "sha": "def5678"},
|
|
],
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_pull_endpoint_returns_confirmed_mobile_pull(monkeypatch):
|
|
calls = []
|
|
|
|
async def access(repository):
|
|
return {"full_name": repository, "permissions": {"push": True}}
|
|
|
|
async def create(repository, *, head, base, title, body, draft, expected_head_sha):
|
|
calls.append((repository, head, base, title, body, draft, expected_head_sha))
|
|
return {
|
|
"number": 42,
|
|
"repository": repository,
|
|
"title": title,
|
|
"body": body,
|
|
"head": {"ref": head, "sha": expected_head_sha},
|
|
"base": {"ref": base},
|
|
"draft": draft,
|
|
"state": "open",
|
|
"url": "https://forge.example/stackchain/api/pulls/42",
|
|
"existing": False,
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "repository_access", access)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_pull", create, 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",
|
|
headers={"Idempotency-Key": "mobile-create-42"},
|
|
json={
|
|
"head": "feature/mobile",
|
|
"base": "main",
|
|
"title": " Ship mobile create ",
|
|
"body": " Context ",
|
|
"draft": True,
|
|
"expected_head_sha": "abc1234",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["number"] == 42
|
|
assert calls == [
|
|
("stackchain/api", "feature/mobile", "main", "Ship mobile create", "Context", True, "abc1234")
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_create_pull_revalidates_head_and_confirms_created_pull():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.method == "GET" and request.url.path.endswith("/branches/feature/mobile"):
|
|
return httpx.Response(200, json={"name": "feature/mobile", "commit": {"id": "abc1234"}})
|
|
if request.method == "GET" and request.url.path.endswith("/pulls"):
|
|
return httpx.Response(200, json=[])
|
|
if request.method == "POST" and request.url.path.endswith("/pulls"):
|
|
return httpx.Response(201, json={
|
|
"number": 42,
|
|
"title": "Ship mobile create",
|
|
"body": "Context",
|
|
"state": "open",
|
|
"draft": True,
|
|
"head": {"ref": "feature/mobile", "sha": "abc1234"},
|
|
"base": {"ref": "main"},
|
|
"user": {"login": "timmy"},
|
|
"html_url": "https://forge.alexanderwhitestone.com/git/stackchain/api/pulls/42",
|
|
})
|
|
return httpx.Response(404)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.create_pull(
|
|
"stackchain/api",
|
|
head="feature/mobile",
|
|
base="main",
|
|
title="Ship mobile create",
|
|
body="Context",
|
|
draft=True,
|
|
expected_head_sha="abc1234",
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result["number"] == 42
|
|
assert result["head"] == {"ref": "feature/mobile", "sha": "abc1234"}
|
|
assert result["author"] == "timmy"
|
|
assert result["existing"] is False
|
|
mutation = next(request for request in requests if request.method == "POST")
|
|
assert mutation.read() == b'{"head":"feature/mobile","base":"main","title":"Ship mobile create","body":"Context","draft":true}'
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_create_pull_returns_existing_source_base_without_mutation():
|
|
requests = []
|
|
existing = {
|
|
"number": 41,
|
|
"title": "Already open",
|
|
"body": "Existing context",
|
|
"state": "open",
|
|
"draft": False,
|
|
"head": {"ref": "feature/mobile", "sha": "abc1234"},
|
|
"base": {"ref": "main"},
|
|
"user": {"login": "timmy"},
|
|
"html_url": "https://forge.alexanderwhitestone.com/git/stackchain/api/pulls/41",
|
|
}
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if "/branches/" in request.url.path:
|
|
return httpx.Response(200, json={"name": "feature/mobile", "commit": {"id": "abc1234"}})
|
|
if request.method == "GET" and request.url.path.endswith("/pulls"):
|
|
return httpx.Response(200, json=[existing])
|
|
return httpx.Response(500)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.create_pull(
|
|
"stackchain/api", head="feature/mobile", base="main",
|
|
title="Duplicate attempt", body="", draft=True,
|
|
expected_head_sha="abc1234",
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result["number"] == 41
|
|
assert result["existing"] is True
|
|
assert all(request.method != "POST" for request in requests)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_pull_endpoint_reports_stale_source_without_retrying(monkeypatch):
|
|
async def access(repository):
|
|
return {"full_name": repository, "permissions": {"push": True}}
|
|
|
|
async def stale(*args, **kwargs):
|
|
raise gitea_proxy.PullCreateConflictError("Source branch changed before pull creation")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "repository_access", access)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_pull", stale)
|
|
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",
|
|
json={
|
|
"head": "feature/mobile", "base": "main", "title": "Ship it",
|
|
"expected_head_sha": "abc1234",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 409
|
|
assert response.json() == {
|
|
"error": "The source branch changed. Refresh branches before creating the pull request."
|
|
}
|