604 lines
24 KiB
Python
604 lines
24 KiB
Python
import asyncio
|
|
|
|
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_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),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_pull_check_status_skips_files_reviews_and_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"},
|
|
})
|
|
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"),
|
|
]
|
|
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": ""}],
|
|
}
|
|
|
|
|
|
@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"},
|
|
})
|
|
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 "conversation" not in detail
|
|
|
|
|
|
@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_reconciles_acceptance_before_timeout(monkeypatch):
|
|
calls = []
|
|
merged = False
|
|
|
|
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.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 == [
|
|
("merge", "stackchain/api", 7, "abc123"),
|
|
("confirm", "stackchain/api", 7, "abc123"),
|
|
]
|
|
|
|
|
|
@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
|
|
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
|
|
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_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_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"]
|