2378 lines
90 KiB
Python
2378 lines
90 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy, main
|
|
from src.idempotency import IdempotencyLedger
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_owned_comment_mutations_verify_thread_and_author_before_writing():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.method == "GET" and request.url.path.endswith("/issues/comments/42"):
|
|
return httpx.Response(200, json={
|
|
"id": 42,
|
|
"body": "Original",
|
|
"user": {"login": "timmy"},
|
|
"issue_url": f"{gitea_proxy.GITEA_URL}/api/v1/repos/stackchain/api/issues/17",
|
|
"html_url": f"{gitea_proxy.GITEA_URL}/stackchain/api/issues/17#issuecomment-42",
|
|
})
|
|
if request.method == "PATCH":
|
|
return httpx.Response(200, json={
|
|
"id": 42, "body": "Corrected", "user": {"login": "timmy"},
|
|
"html_url": "https://gitea.example/stackchain/api/issues/17#issuecomment-42",
|
|
})
|
|
return httpx.Response(204)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
edited = await gitea_proxy.edit_owned_comment("stackchain/api", 17, 42, "Corrected")
|
|
deleted = await gitea_proxy.delete_owned_comment("stackchain/api", 17, 42)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert edited["body"] == "Corrected"
|
|
assert deleted == {"id": 42, "deleted": True}
|
|
assert [(request.method, request.url.path) for request in requests] == [
|
|
("GET", "/api/v1/user"),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/comments/42"),
|
|
("PATCH", "/api/v1/repos/stackchain/api/issues/comments/42"),
|
|
("GET", "/api/v1/user"),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/comments/42"),
|
|
("DELETE", "/api/v1/repos/stackchain/api/issues/comments/42"),
|
|
]
|
|
assert json.loads(requests[2].content) == {"body": "Corrected"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize(
|
|
"comment",
|
|
[
|
|
{"id": 42, "user": {"login": "alex"}, "issue_url": "https://gitea.example/api/v1/repos/stackchain/api/issues/17"},
|
|
{"id": 42, "user": {"login": "timmy"}, "issue_url": f"{gitea_proxy.GITEA_URL}/api/v1/repos/stackchain/api/issues/99"},
|
|
{"id": 42, "user": {"login": "timmy"}, "issue_url": "https://evil.example/api/v1/repos/stackchain/api/issues/17"},
|
|
],
|
|
)
|
|
async def test_owned_comment_mutations_reject_other_author_or_thread_without_writing(comment):
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json=comment)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
with pytest.raises(gitea_proxy.CommentMutationForbiddenError):
|
|
await gitea_proxy.edit_owned_comment("stackchain/api", 17, 42, "Nope")
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert [request.method for request in requests] == ["GET", "GET"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize(
|
|
("base_path", "availability_call"),
|
|
[
|
|
("/api/v1/repos/stackchain/api/issues/17", "issue"),
|
|
("/api/v1/repos/stackchain/api/pulls/17", "pull"),
|
|
("/api/v1/notifications/91", "notification"),
|
|
],
|
|
)
|
|
async def test_comment_mutation_routes_cover_issue_pull_and_unread_update(
|
|
monkeypatch, base_path, availability_call
|
|
):
|
|
calls = []
|
|
|
|
async def available_issue(repository, number):
|
|
calls.append(("issue", repository, number))
|
|
return True
|
|
|
|
async def available_pull(repository, number):
|
|
calls.append(("pull", repository, number))
|
|
return True
|
|
|
|
async def notification_target(thread_id):
|
|
calls.append(("notification", thread_id))
|
|
return "stackchain/api", 17
|
|
|
|
async def edit(repository, number, comment_id, body):
|
|
calls.append(("edit", repository, number, comment_id, body))
|
|
return {"id": comment_id, "author": "timmy", "body": body}
|
|
|
|
async def delete(repository, number, comment_id):
|
|
calls.append(("delete", repository, number, comment_id))
|
|
return {"id": comment_id, "deleted": True}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", available_issue)
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", available_pull)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "notification_conversation_target", notification_target,
|
|
raising=False,
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "edit_owned_comment", edit)
|
|
monkeypatch.setattr(main.gitea_proxy, "delete_owned_comment", delete)
|
|
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
edited = await client.patch(base_path + "/comments/42", json={"body": " Corrected "})
|
|
deleted = await client.delete(base_path + "/comments/42")
|
|
|
|
assert edited.status_code == 200
|
|
assert edited.json()["body"] == "Corrected"
|
|
assert deleted.status_code == 200
|
|
assert deleted.json() == {"id": 42, "deleted": True}
|
|
if availability_call == "notification":
|
|
assert calls == [
|
|
("notification", 91), ("edit", "stackchain/api", 17, 42, "Corrected"),
|
|
("notification", 91), ("delete", "stackchain/api", 17, 42),
|
|
]
|
|
else:
|
|
assert calls == [
|
|
(availability_call, "stackchain/api", 17),
|
|
("edit", "stackchain/api", 17, 42, "Corrected"),
|
|
(availability_call, "stackchain/api", 17),
|
|
("delete", "stackchain/api", 17, 42),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_comment_mutation_route_returns_forbidden_without_hiding_saved_edit(monkeypatch):
|
|
async def available(repository, number):
|
|
return True
|
|
|
|
async def forbidden(*args):
|
|
raise gitea_proxy.CommentMutationForbiddenError("not owned")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", available)
|
|
monkeypatch.setattr(main.gitea_proxy, "edit_owned_comment", forbidden)
|
|
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/issues/17/comments/42",
|
|
json={"body": "Keep this draft"},
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"] == "You can only change your own comments"
|
|
|
|
|
|
async def repository_access_from(loader, repository):
|
|
repositories = await loader()
|
|
return next(
|
|
(item for item in repositories if item.get("full_name") == repository), None
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_milestone_update_revalidates_assignment_and_open_repository_milestone():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.method == "GET" and request.url.path.endswith("/issues/17"):
|
|
return httpx.Response(200, json={
|
|
"number": 17, "state": "open", "pull_request": None,
|
|
"assignees": [{"login": "timmy"}],
|
|
})
|
|
if request.method == "GET" and request.url.path.endswith("/milestones"):
|
|
return httpx.Response(200, json=[
|
|
{"id": 9, "title": "August RC", "state": "open"},
|
|
{"id": 10, "title": "Old release", "state": "closed"},
|
|
{"id": "bad", "title": "Invalid", "state": "open"},
|
|
])
|
|
return httpx.Response(200, json={
|
|
"number": 17, "state": "open",
|
|
"milestone": {"id": 9, "title": "August RC"},
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
options = await gitea_proxy.repo_milestones("stackchain/api")
|
|
result = await gitea_proxy.update_assigned_issue_milestone(
|
|
"stackchain/api", 17, 9
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert options == [{"id": 9, "title": "August RC"}]
|
|
patch_request = next(request for request in requests if request.method == "PATCH")
|
|
assert patch_request.url.path == "/api/v1/repos/stackchain/api/issues/17"
|
|
assert json.loads(patch_request.content) == {"milestone": 9}
|
|
assert result == {
|
|
"repository": "stackchain/api", "number": 17, "state": "open",
|
|
"milestone": {"id": 9, "title": "August RC"},
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_milestone_routes_are_repository_bounded_and_no_store(monkeypatch):
|
|
calls = []
|
|
|
|
async def available_repos():
|
|
return [{"full_name": "stackchain/api"}]
|
|
|
|
async def milestones(repository):
|
|
calls.append(("list", repository))
|
|
return [{"id": 9, "title": "August RC"}]
|
|
|
|
async def update(repository, number, milestone_id):
|
|
calls.append(("update", repository, number, milestone_id))
|
|
return {
|
|
"repository": repository, "number": number, "state": "open",
|
|
"milestone": {"id": milestone_id, "title": "August RC"} if milestone_id else None,
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "repository_access",
|
|
lambda repository: repository_access_from(available_repos, repository),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "repo_milestones", milestones, raising=False)
|
|
monkeypatch.setattr(main.gitea_proxy, "update_assigned_issue_milestone", update, 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/milestones")
|
|
updated = await client.patch(
|
|
"/api/v1/repos/stackchain/api/issues/17/milestone", json={"milestone_id": 9}
|
|
)
|
|
missing = await client.get("/api/v1/repos/other/private/milestones")
|
|
|
|
assert listed.status_code == 200
|
|
assert listed.headers["cache-control"] == "no-store"
|
|
assert listed.json() == [{"id": 9, "title": "August RC"}]
|
|
assert updated.status_code == 200
|
|
assert updated.headers["cache-control"] == "no-store"
|
|
assert updated.json()["milestone"] == {"id": 9, "title": "August RC"}
|
|
assert missing.status_code == 404
|
|
assert calls == [
|
|
("list", "stackchain/api"),
|
|
("update", "stackchain/api", 17, 9),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize(
|
|
("due_date", "expected_payload", "confirmed_due_date"),
|
|
[
|
|
("2026-08-09T23:59:59Z", {"due_date": "2026-08-09T23:59:59Z"}, "2026-08-09T23:59:59Z"),
|
|
(None, {"unset_due_date": True}, None),
|
|
],
|
|
)
|
|
async def test_gitea_due_date_update_revalidates_assignment_and_confirms_result(
|
|
due_date, expected_payload, confirmed_due_date
|
|
):
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.method == "GET":
|
|
return httpx.Response(200, json={
|
|
"number": 17, "state": "open", "pull_request": None,
|
|
"assignees": [{"login": "timmy"}],
|
|
})
|
|
return httpx.Response(200, json={
|
|
"number": 17, "state": "open", "due_date": confirmed_due_date,
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.update_assigned_issue_due_date(
|
|
"stackchain/api", 17, due_date
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert [(request.method, request.url.path) for request in requests] == [
|
|
("GET", "/api/v1/user"),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/17"),
|
|
("PATCH", "/api/v1/repos/stackchain/api/issues/17"),
|
|
]
|
|
assert json.loads(requests[2].content) == expected_payload
|
|
assert result == {
|
|
"repository": "stackchain/api", "number": 17,
|
|
"state": "open", "due_date": confirmed_due_date,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_due_date_endpoint_sets_or_clears_assigned_issue(monkeypatch):
|
|
calls = []
|
|
|
|
async def update(repository, number, due_date):
|
|
calls.append((repository, number, due_date))
|
|
return {
|
|
"repository": repository, "number": number, "state": "open",
|
|
"due_date": due_date,
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "update_assigned_issue_due_date", update, raising=False
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
set_response = await client.patch(
|
|
"/api/v1/repos/stackchain/api/issues/17/due-date",
|
|
json={"due_date": "2026-08-09T23:59:59Z"},
|
|
)
|
|
clear_response = await client.patch(
|
|
"/api/v1/repos/stackchain/api/issues/17/due-date",
|
|
json={"due_date": None},
|
|
)
|
|
|
|
assert [set_response.status_code, clear_response.status_code] == [200, 200]
|
|
assert set_response.headers["cache-control"] == "no-store"
|
|
assert set_response.json()["due_date"] == "2026-08-09T23:59:59Z"
|
|
assert clear_response.json()["due_date"] is None
|
|
assert calls == [
|
|
("stackchain/api", 17, "2026-08-09T23:59:59Z"),
|
|
("stackchain/api", 17, None),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_edit_assigned_issue_updates_title_and_body_at_expected_revision(monkeypatch):
|
|
calls = []
|
|
|
|
async def update(repository, number, title, body, expected_updated_at):
|
|
calls.append((repository, number, title, body, expected_updated_at))
|
|
return {
|
|
"repository": repository,
|
|
"number": number,
|
|
"title": title,
|
|
"body": body,
|
|
"state": "open",
|
|
"updated_at": "2026-08-07T10:01:00Z",
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "update_assigned_issue", 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/issues/17/content",
|
|
json={
|
|
"title": " Clarified scope ",
|
|
"body": " Updated acceptance criteria ",
|
|
"expected_updated_at": "2026-08-07T10:00:00Z",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["title"] == "Clarified scope"
|
|
assert calls == [(
|
|
"stackchain/api", 17, "Clarified scope", "Updated acceptance criteria",
|
|
"2026-08-07T10:00:00Z",
|
|
)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_edit_issue_rejects_stale_revision_without_patch():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"number": 17,
|
|
"title": "Changed upstream",
|
|
"body": "Newer body",
|
|
"state": "open",
|
|
"updated_at": "2026-08-07T10:02:00Z",
|
|
"assignees": [{"login": "timmy"}],
|
|
"pull_request": None,
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
with pytest.raises(gitea_proxy.IssueEditConflictError):
|
|
await gitea_proxy.update_assigned_issue(
|
|
"stackchain/api", 17, "My draft", "Draft body",
|
|
"2026-08-07T10:00:00Z",
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert [(request.method, request.url.path) for request in requests] == [
|
|
("GET", "/api/v1/user"),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/17"),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_edit_issue_revalidates_assignment_and_confirms_content():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.method == "GET":
|
|
return httpx.Response(200, json={
|
|
"number": 17, "title": "Old", "body": "Old body", "state": "open",
|
|
"updated_at": "2026-08-07T10:00:00Z",
|
|
"assignees": [{"login": "timmy"}], "pull_request": None,
|
|
})
|
|
return httpx.Response(200, json={
|
|
"number": 17, "title": "Clarified", "body": "New body", "state": "open",
|
|
"updated_at": "2026-08-07T10:01:00Z",
|
|
"html_url": "https://forge.example/stackchain/api/issues/17",
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.update_assigned_issue(
|
|
"stackchain/api", 17, "Clarified", "New body", "2026-08-07T10:00:00Z"
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert [(request.method, request.url.path) for request in requests] == [
|
|
("GET", "/api/v1/user"),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/17"),
|
|
("PATCH", "/api/v1/repos/stackchain/api/issues/17"),
|
|
]
|
|
assert requests[2].content == b'{"title":"Clarified","body":"New body"}'
|
|
assert result == {
|
|
"repository": "stackchain/api", "number": 17, "title": "Clarified",
|
|
"body": "New body", "state": "open", "updated_at": "2026-08-07T10:01:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/17",
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_edit_assigned_issue_reports_revision_conflict_without_mutation(monkeypatch):
|
|
async def update(*_args):
|
|
raise gitea_proxy.IssueEditConflictError("changed")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "update_assigned_issue", update)
|
|
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/issues/17/content",
|
|
json={
|
|
"title": "My draft",
|
|
"body": "Still safe",
|
|
"expected_updated_at": "2026-08-07T10:00:00Z",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 409
|
|
assert response.json() == {
|
|
"error": "This issue changed in Gitea. Your draft is safe; reload the latest issue before saving."
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_issue_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_create_issue_endpoint_derives_self_assignment_and_returns_confirmed_issue(monkeypatch):
|
|
calls = []
|
|
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
async def available_repos():
|
|
return [{"full_name": "stackchain/api"}]
|
|
|
|
async def labels(repository):
|
|
assert repository == "stackchain/api"
|
|
return [
|
|
{"id": 3, "name": "P0", "color": "d73a4a"},
|
|
{"id": 8, "name": "frontend", "color": "1d76db"},
|
|
]
|
|
|
|
async def create(repository, title, body, assignee, label_ids):
|
|
calls.append((repository, title, body, assignee, label_ids))
|
|
return {
|
|
"id": 81,
|
|
"number": 17,
|
|
"title": title,
|
|
"state": "open",
|
|
"repository": repository,
|
|
"labels": ["P0"],
|
|
"assignees": [assignee],
|
|
"updated_at": "2026-08-07T03:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/17",
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "repository_access",
|
|
lambda repository: repository_access_from(available_repos, repository),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "repo_labels", labels, raising=False)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_issue", 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/issues",
|
|
json={
|
|
"title": " Capture mobile work ",
|
|
"body": " Context ",
|
|
"label_ids": [3],
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["number"] == 17
|
|
assert response.json()["assignees"] == ["timmy"]
|
|
assert response.json()["labels"] == ["P0"]
|
|
assert calls == [("stackchain/api", "Capture mobile work", "Context", "timmy", [3])]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_repository_page_reports_more_results_and_targeted_access_uses_repository_route():
|
|
calls = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
calls.append((request.url.path, dict(request.url.params)))
|
|
if request.url.path.endswith("/user/repos"):
|
|
return httpx.Response(
|
|
200,
|
|
json=[{"id": 51, "name": "later", "full_name": "stackchain/later"}],
|
|
headers={"X-Total-Count": "51"},
|
|
)
|
|
if request.url.path.endswith("/repos/stackchain/later"):
|
|
return httpx.Response(200, json={"id": 51, "full_name": "stackchain/later"})
|
|
return httpx.Response(404)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
page = await gitea_proxy.repo_page(page=2, limit=50)
|
|
first_page = await gitea_proxy.repos()
|
|
accessible = await gitea_proxy.repository_access("stackchain/later")
|
|
missing = await gitea_proxy.repository_access("stackchain/missing")
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert page == {
|
|
"items": [{"id": 51, "name": "later", "full_name": "stackchain/later"}],
|
|
"page": 2,
|
|
"total": 51,
|
|
"has_more": False,
|
|
}
|
|
assert accessible["full_name"] == "stackchain/later"
|
|
assert missing is None
|
|
assert first_page.pagination == {"page": 1, "total": 51, "has_more": True}
|
|
assert calls == [
|
|
("/api/v1/user/repos", {"page": "2", "limit": "50"}),
|
|
("/api/v1/user/repos", {"page": "1", "limit": "50"}),
|
|
("/api/v1/repos/stackchain/later", {}),
|
|
("/api/v1/repos/stackchain/missing", {}),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_repository_search_finds_visible_repositories_without_loading_pages():
|
|
requests = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append((request.url.path, dict(request.url.params)))
|
|
return httpx.Response(200, json={
|
|
"ok": True,
|
|
"data": [
|
|
{"id": 151, "name": "mobile", "full_name": "stackchain/mobile", "private": True},
|
|
"invalid",
|
|
],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.search_repositories(" mobile ", limit=20)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result == [{
|
|
"id": 151, "name": "mobile", "full_name": "stackchain/mobile", "private": True,
|
|
}]
|
|
assert requests == [(
|
|
"/api/v1/repos/search", {"q": "mobile", "limit": "20", "page": "1"},
|
|
)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_repository_search_endpoint_returns_minimal_visible_matches(monkeypatch):
|
|
calls = []
|
|
|
|
async def search(query, limit):
|
|
calls.append((query, limit))
|
|
return [{
|
|
"id": 151,
|
|
"name": "mobile",
|
|
"full_name": "stackchain/mobile",
|
|
"html_url": "https://forge.example/stackchain/mobile",
|
|
"description": "secret context that the picker does not need",
|
|
"private": True,
|
|
}]
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "search_repositories", search)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/repositories/search?q=%20mobile%20&limit=20")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {"items": [{
|
|
"id": 151, "name": "mobile", "full_name": "stackchain/mobile",
|
|
}]}
|
|
assert calls == [("mobile", 20)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_repository_page_endpoint_and_issue_creation_support_later_repository(monkeypatch):
|
|
access_calls = []
|
|
create_calls = []
|
|
|
|
async def page(page, limit):
|
|
assert (page, limit) == (2, 50)
|
|
return {
|
|
"items": [{
|
|
"id": 51, "name": "later", "full_name": "stackchain/later",
|
|
"description": "", "html_url": "https://forge.example/stackchain/later",
|
|
"updated_at": "2026-08-09T12:00:00Z",
|
|
}],
|
|
"page": 2, "total": 51, "has_more": False,
|
|
}
|
|
|
|
async def access(repository):
|
|
access_calls.append(repository)
|
|
return {"id": 51, "full_name": repository} if repository == "stackchain/later" else None
|
|
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
async def create(repository, title, body, assignee, label_ids):
|
|
create_calls.append((repository, title, assignee))
|
|
return {
|
|
"id": 403, "number": 403, "title": title, "state": "open",
|
|
"repository": repository, "labels": [], "assignees": [assignee],
|
|
"updated_at": "2026-08-09T12:00:00Z",
|
|
"url": "https://forge.example/stackchain/later/issues/403",
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "repo_page", page)
|
|
monkeypatch.setattr(main.gitea_proxy, "repository_access", access)
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_issue", create)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
listed = await client.get("/api/v1/repositories?page=2&limit=50")
|
|
created = await client.post(
|
|
"/api/v1/repos/stackchain/later/issues", json={"title": "Later-page work"}
|
|
)
|
|
|
|
assert listed.status_code == 200
|
|
assert listed.json()["items"][0]["full_name"] == "stackchain/later"
|
|
assert listed.json()["has_more"] is False
|
|
assert created.status_code == 201
|
|
assert access_calls == ["stackchain/later"]
|
|
assert create_calls == [("stackchain/later", "Later-page work", "timmy")]
|
|
|
|
|
|
def test_context_exposes_truthful_repository_pagination():
|
|
repositories = gitea_proxy.RepositoryItems(
|
|
[{
|
|
"id": 1, "name": "api", "full_name": "stackchain/api",
|
|
"description": "", "html_url": "https://forge.example/stackchain/api",
|
|
"updated_at": "2026-08-09T12:00:00Z",
|
|
}],
|
|
{"page": 1, "total": 51, "has_more": True},
|
|
)
|
|
|
|
payload = main._context_payload(
|
|
{"id": 2, "login": "timmy"}, repositories, [], []
|
|
)
|
|
|
|
assert payload["repository_pagination"] == {
|
|
"page": 1, "total": 51, "has_more": True
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_issue_atomically_validates_and_sends_release_plan(monkeypatch):
|
|
calls = []
|
|
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
async def available_repos():
|
|
return [{"full_name": "stackchain/api"}]
|
|
|
|
async def milestones(repository):
|
|
assert repository == "stackchain/api"
|
|
return [{"id": 9, "title": "August RC"}]
|
|
|
|
async def create(repository, title, body, assignee, label_ids, milestone_id, due_date):
|
|
calls.append((repository, title, body, assignee, label_ids, milestone_id, due_date))
|
|
return {
|
|
"id": 221, "number": 221, "title": title, "state": "open",
|
|
"repository": repository, "labels": [], "assignees": [assignee],
|
|
"milestone": {"id": 9, "title": "August RC"},
|
|
"due_date": "2026-08-31T23:59:59Z",
|
|
"updated_at": "2026-08-07T20:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/221",
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "repository_access",
|
|
lambda repository: repository_access_from(available_repos, repository),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "repo_milestones", milestones)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_issue", create)
|
|
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/issues",
|
|
json={
|
|
"title": "Ship mobile plan",
|
|
"milestone_id": 9,
|
|
"due_date": "2026-08-31T23:59:59Z",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["milestone"] == {"id": 9, "title": "August RC"}
|
|
assert response.json()["due_date"] == "2026-08-31T23:59:59Z"
|
|
assert calls == [(
|
|
"stackchain/api", "Ship mobile plan", "", "timmy", [], 9,
|
|
"2026-08-31T23:59:59Z",
|
|
)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_issue_replays_one_upstream_result_for_concurrent_idempotent_requests(monkeypatch):
|
|
calls = 0
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
async def available_repos():
|
|
return [{"full_name": "stackchain/api"}]
|
|
|
|
async def create(repository, title, body, assignee, label_ids):
|
|
nonlocal calls
|
|
calls += 1
|
|
started.set()
|
|
await release.wait()
|
|
return {
|
|
"id": 81, "number": 17, "title": title, "state": "open",
|
|
"repository": repository, "labels": [], "assignees": [assignee],
|
|
"updated_at": "2026-08-07T03:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/17",
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "repository_access",
|
|
lambda repository: repository_access_from(available_repos, repository),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_issue", create)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "capture-177-concurrent"}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first = asyncio.create_task(client.post(
|
|
"/api/v1/repos/stackchain/api/issues", json={"title": "Capture work"}, headers=headers
|
|
))
|
|
await started.wait()
|
|
second = asyncio.create_task(client.post(
|
|
"/api/v1/repos/stackchain/api/issues", json={"title": "Capture work"}, headers=headers
|
|
))
|
|
await asyncio.sleep(0)
|
|
release.set()
|
|
responses = await asyncio.gather(first, second)
|
|
|
|
assert [response.status_code for response in responses] == [201, 201]
|
|
assert [response.json()["number"] for response in responses] == [17, 17]
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_completed_issue_creation_replays_after_ledger_reconstruction(monkeypatch, tmp_path):
|
|
calls = 0
|
|
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
async def available_repos():
|
|
return [{"full_name": "stackchain/api"}]
|
|
|
|
async def create(repository, title, body, assignee, label_ids):
|
|
nonlocal calls
|
|
calls += 1
|
|
return {
|
|
"id": 201, "number": 201, "title": title, "state": "open",
|
|
"repository": repository, "labels": [], "assignees": [assignee],
|
|
"updated_at": "2026-08-07T15:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/201",
|
|
}
|
|
|
|
database = tmp_path / "issues.sqlite3"
|
|
monkeypatch.setattr(
|
|
main,
|
|
"_idempotency_ledger",
|
|
IdempotencyLedger(database, ttl_seconds=600, max_entries=256),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "repository_access",
|
|
lambda repository: repository_access_from(available_repos, repository),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_issue", create)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "restart-create-201"}
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
created = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues",
|
|
json={"title": "Durable capture"},
|
|
headers=headers,
|
|
)
|
|
main._authored_action_operations.clear()
|
|
monkeypatch.setattr(
|
|
main,
|
|
"_idempotency_ledger",
|
|
IdempotencyLedger(database, ttl_seconds=600, max_entries=256),
|
|
)
|
|
replayed = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues",
|
|
json={"title": "Durable capture"},
|
|
headers=headers,
|
|
)
|
|
|
|
assert [created.status_code, replayed.status_code] == [201, 201]
|
|
assert replayed.json()["number"] == 201
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_issue_rejects_changed_payload_for_an_existing_idempotency_key(monkeypatch):
|
|
calls = []
|
|
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
async def available_repos():
|
|
return [{"full_name": "stackchain/api"}]
|
|
|
|
async def create(repository, title, body, assignee, label_ids):
|
|
calls.append(title)
|
|
return {
|
|
"id": 81, "number": 17, "title": title, "state": "open",
|
|
"repository": repository, "labels": [], "assignees": [assignee],
|
|
"updated_at": "2026-08-07T03:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/17",
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "repository_access",
|
|
lambda repository: repository_access_from(available_repos, repository),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_issue", create)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "capture-177-payload-conflict"}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
created = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues", json={"title": "First title"}, headers=headers
|
|
)
|
|
conflict = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues", json={"title": "Changed title"}, headers=headers
|
|
)
|
|
|
|
assert created.status_code == 201
|
|
assert conflict.status_code == 409
|
|
assert conflict.json()["detail"] == "Idempotency key already used"
|
|
assert calls == ["First title"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_issue_retry_recovers_result_after_the_first_request_times_out(monkeypatch):
|
|
calls = 0
|
|
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
async def available_repos():
|
|
return [{"full_name": "stackchain/api"}]
|
|
|
|
async def create(repository, title, body, assignee, label_ids):
|
|
nonlocal calls
|
|
calls += 1
|
|
await asyncio.sleep(0.03)
|
|
return {
|
|
"id": 81, "number": 17, "title": title, "state": "open",
|
|
"repository": repository, "labels": [], "assignees": [assignee],
|
|
"updated_at": "2026-08-07T03:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/17",
|
|
}
|
|
|
|
monkeypatch.setattr(main, "ISSUE_ACTION_TIMEOUT_SECONDS", 0.01)
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "repository_access",
|
|
lambda repository: repository_access_from(available_repos, repository),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_issue", create)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "capture-177-timeout-recovery"}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
timed_out = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues", json={"title": "Slow capture"}, headers=headers
|
|
)
|
|
await asyncio.sleep(0.03)
|
|
recovered = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues", json={"title": "Slow capture"}, headers=headers
|
|
)
|
|
|
|
assert timed_out.status_code == 503
|
|
assert recovered.status_code == 201
|
|
assert recovered.json()["number"] == 17
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_issue_durable_ledger_evicts_completed_entry_at_size_limit(
|
|
monkeypatch, tmp_path
|
|
):
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
async def available_repos():
|
|
return [{"full_name": "stackchain/api"}]
|
|
|
|
async def create(repository, title, body, assignee, label_ids):
|
|
return {
|
|
"id": 81, "number": 17, "title": title, "state": "open",
|
|
"repository": repository, "labels": [], "assignees": [assignee],
|
|
"updated_at": "2026-08-07T03:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/17",
|
|
}
|
|
|
|
ledger = IdempotencyLedger(
|
|
tmp_path / "bounded-issues.sqlite3", ttl_seconds=600, max_entries=2
|
|
)
|
|
monkeypatch.setattr(main, "_idempotency_ledger", ledger)
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "repository_access",
|
|
lambda repository: repository_access_from(available_repos, repository),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_issue", create)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
for index in range(3):
|
|
response = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues",
|
|
json={"title": f"Capture {index}"},
|
|
headers={"Idempotency-Key": f"capture-177-bounded-{index}"},
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
replay = ledger.reserve(
|
|
"capture-177-bounded-0",
|
|
("issue-create", "stackchain/api", "Capture 0", "", ()),
|
|
)
|
|
assert replay.state == "reserved"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_create_issue_posts_self_assignment_and_normalizes_confirmation():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
return httpx.Response(201, json={
|
|
"id": 81, "number": 17, "title": "Capture mobile work", "state": "open",
|
|
"updated_at": "2026-08-07T03:00:00Z",
|
|
"html_url": "https://forge.example/stackchain/api/issues/17",
|
|
"assignees": [{"login": "timmy"}],
|
|
"labels": [{"id": 3, "name": "P0"}],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.create_issue(
|
|
"stackchain/api", "Capture mobile work", "Context", "timmy", [3]
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(requests) == 1
|
|
assert requests[0].method == "POST"
|
|
assert requests[0].url.path == "/api/v1/repos/stackchain/api/issues"
|
|
assert requests[0].content == (
|
|
b'{"title":"Capture mobile work","body":"Context","assignee":"timmy","labels":[3]}'
|
|
)
|
|
assert result["repository"] == "stackchain/api"
|
|
assert result["number"] == 17
|
|
assert result["assignees"] == ["timmy"]
|
|
assert result["labels"] == ["P0"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_create_issue_requires_confirmed_release_plan():
|
|
async def handler(_request):
|
|
return httpx.Response(201, json={
|
|
"id": 221, "number": 221, "title": "Planned work", "state": "open",
|
|
"assignees": [{"login": "timmy"}], "labels": [],
|
|
"milestone": None, "due_date": "2026-09-01T23:59:59Z",
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
with pytest.raises(ValueError, match="release plan"):
|
|
await gitea_proxy.create_issue(
|
|
"stackchain/api", "Planned work", "", "timmy", [], 9,
|
|
"2026-08-31T23:59:59Z",
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_repo_labels_returns_safe_touch_picker_options():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
return httpx.Response(200, json=[
|
|
{"id": 3, "name": "P0", "color": "d73a4a", "description": "Urgent"},
|
|
{"id": "bad", "name": "invalid", "color": "000000"},
|
|
])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.repo_labels("stackchain/api")
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests[0].url.path == "/api/v1/repos/stackchain/api/labels"
|
|
assert requests[0].url.params["limit"] == "50"
|
|
assert result == [{
|
|
"id": 3, "name": "P0", "color": "d73a4a", "description": "Urgent"
|
|
}]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_repo_labels_endpoint_only_loads_labels_for_accessible_repository(monkeypatch):
|
|
calls = []
|
|
|
|
async def available_repos():
|
|
return [{"full_name": "stackchain/api"}]
|
|
|
|
async def labels(repository):
|
|
calls.append(repository)
|
|
return [{"id": 3, "name": "P0", "color": "d73a4a", "description": "Urgent"}]
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "repository_access",
|
|
lambda repository: repository_access_from(available_repos, repository),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "repo_labels", labels)
|
|
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/labels")
|
|
missing = await client.get("/api/v1/repos/other/private/labels")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == [{
|
|
"id": 3, "name": "P0", "color": "d73a4a", "description": "Urgent"
|
|
}]
|
|
assert missing.status_code == 404
|
|
assert calls == ["stackchain/api"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_issue_rejects_blank_title_before_upstream(monkeypatch):
|
|
called = False
|
|
|
|
async def user():
|
|
nonlocal called
|
|
called = True
|
|
return {"login": "timmy"}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
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/issues", json={"title": " ", "body": "Context"}
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_issue_rejects_nonexistent_due_date_before_upstream(monkeypatch):
|
|
called = False
|
|
|
|
async def user():
|
|
nonlocal called
|
|
called = True
|
|
return {"login": "timmy"}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
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/issues",
|
|
json={"title": "Impossible plan", "due_date": "2026-02-31T23:59:59Z"},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_issue_rejects_label_not_in_target_repository(monkeypatch):
|
|
created = False
|
|
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
async def available_repos():
|
|
return [{"full_name": "stackchain/api"}]
|
|
|
|
async def labels(_repository):
|
|
return [{"id": 3, "name": "P0"}]
|
|
|
|
async def create(*_args):
|
|
nonlocal created
|
|
created = True
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "current_user", user)
|
|
monkeypatch.setattr(main.gitea_proxy, "repos", available_repos)
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "repository_access",
|
|
lambda repository: repository_access_from(available_repos, repository),
|
|
)
|
|
monkeypatch.setattr(main.gitea_proxy, "repo_labels", labels)
|
|
monkeypatch.setattr(main.gitea_proxy, "create_issue", create)
|
|
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/issues",
|
|
json={"title": "Urgent work", "label_ids": [999]},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
assert response.json()["detail"] == "Unknown repository label"
|
|
assert created is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_create_issue_requires_confirmed_self_assignment():
|
|
async def handler(_request):
|
|
return httpx.Response(201, json={
|
|
"id": 81, "number": 17, "title": "Capture mobile work", "state": "open",
|
|
"html_url": "https://forge.example/stackchain/api/issues/17", "assignees": [],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
with pytest.raises(ValueError, match="self-assignment"):
|
|
await gitea_proxy.create_issue(
|
|
"stackchain/api", "Capture mobile work", "", "timmy"
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_claim_available_issue_rechecks_then_confirms_authenticated_assignee():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.method == "GET" and request.url.path.endswith("/issues/17"):
|
|
return httpx.Response(200, json={
|
|
"id": 81, "number": 17, "title": "Available", "state": "open",
|
|
"assignees": None, "pull_request": None, "labels": [],
|
|
"html_url": "https://forge.example/stackchain/api/issues/17",
|
|
})
|
|
if request.method == "GET" and request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"id": 81, "number": 17, "title": "Available", "state": "open",
|
|
"assignees": [{"login": "timmy"}], "labels": [],
|
|
"html_url": "https://forge.example/stackchain/api/issues/17",
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.claim_available_issue("stackchain/api", 17)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert [(request.method, request.url.path) for request in requests] == [
|
|
("GET", "/api/v1/repos/stackchain/api/issues/17"),
|
|
("GET", "/api/v1/user"),
|
|
("PATCH", "/api/v1/repos/stackchain/api/issues/17"),
|
|
]
|
|
assert requests[2].content == b'{"assignee":"timmy"}'
|
|
assert result["repository"] == "stackchain/api"
|
|
assert result["assignees"] == ["timmy"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_reopen_issue_rechecks_closed_state_and_confirms_self_assignment():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.method == "GET" and request.url.path.endswith("/issues/17"):
|
|
return httpx.Response(200, json={
|
|
"id": 81, "number": 17, "title": "Resume work", "state": "closed",
|
|
"assignees": [], "pull_request": None, "labels": [{"name": "P1"}],
|
|
"html_url": "https://forge.example/stackchain/api/issues/17",
|
|
})
|
|
if request.method == "GET" and request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"id": 81, "number": 17, "title": "Resume work", "state": "open",
|
|
"assignees": [{"login": "timmy"}], "labels": [{"name": "P1"}],
|
|
"html_url": "https://forge.example/stackchain/api/issues/17",
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.reopen_issue("stackchain/api", 17)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert [(request.method, request.url.path) for request in requests] == [
|
|
("GET", "/api/v1/repos/stackchain/api/issues/17"),
|
|
("GET", "/api/v1/user"),
|
|
("PATCH", "/api/v1/repos/stackchain/api/issues/17"),
|
|
]
|
|
assert requests[2].content == b'{"state":"open","assignee":"timmy"}'
|
|
assert result == {
|
|
"id": 81, "number": 17, "title": "Resume work", "state": "open",
|
|
"repository": "stackchain/api", "labels": ["P1"],
|
|
"assignees": ["timmy"], "updated_at": "",
|
|
"url": "https://forge.example/stackchain/api/issues/17",
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_reopen_issue_retry_accepts_already_open_self_assigned_issue():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"id": 81, "number": 17, "title": "Resume work", "state": "open",
|
|
"assignees": [{"login": "timmy"}], "pull_request": None, "labels": [],
|
|
"html_url": "https://forge.example/stackchain/api/issues/17",
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.reopen_issue("stackchain/api", 17)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result["state"] == "open"
|
|
assert result["assignees"] == ["timmy"]
|
|
assert all(request.method == "GET" for request in requests)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_claim_available_issue_endpoint_returns_confirmed_work_item(monkeypatch):
|
|
calls = []
|
|
|
|
async def claim(repository, number):
|
|
calls.append((repository, number))
|
|
return {
|
|
"id": 81, "number": number, "title": "Available", "state": "open",
|
|
"repository": repository, "labels": [], "assignees": ["timmy"],
|
|
"url": "https://forge.example/stackchain/api/issues/17",
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "claim_available_issue", claim)
|
|
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/issues/17/claim")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["assignees"] == ["timmy"]
|
|
assert calls == [("stackchain/api", 17)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_claim_available_issue_endpoint_reports_assignment_race_as_conflict(monkeypatch):
|
|
async def claim(_repository, _number):
|
|
raise gitea_proxy.IssueNotAvailableError("claimed")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "claim_available_issue", claim)
|
|
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/issues/17/claim")
|
|
|
|
assert response.status_code == 409
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {
|
|
"error": "This issue was already claimed or is no longer open."
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_reopen_issue_endpoint_returns_confirmed_resumable_work_item(monkeypatch):
|
|
calls = []
|
|
|
|
async def reopen(repository, number):
|
|
calls.append((repository, number))
|
|
return {
|
|
"id": 81, "number": number, "title": "Resume work", "state": "open",
|
|
"repository": repository, "labels": ["P1"], "assignees": ["timmy"],
|
|
"url": "https://forge.example/stackchain/api/issues/17",
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "reopen_issue", reopen, 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/issues/17/reopen")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["state"] == "open"
|
|
assert response.json()["assignees"] == ["timmy"]
|
|
assert calls == [("stackchain/api", 17)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_release_issue_removes_only_authenticated_user_and_confirms_peers():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.method == "GET" and request.url.path.endswith("/issues/17"):
|
|
return httpx.Response(200, json={
|
|
"id": 81, "number": 17, "title": "Shared work", "state": "open",
|
|
"pull_request": None,
|
|
"assignees": [{"login": "timmy"}, {"login": "alex"}],
|
|
"labels": [],
|
|
})
|
|
if request.method == "GET" and request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"id": 81, "number": 17, "title": "Shared work", "state": "open",
|
|
"assignees": [{"login": "alex"}], "labels": [],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.release_assigned_issue("stackchain/api", 17)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert [(request.method, request.url.path) for request in requests] == [
|
|
("GET", "/api/v1/user"),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/17"),
|
|
("PATCH", "/api/v1/repos/stackchain/api/issues/17"),
|
|
]
|
|
assert requests[2].content == b'{"assignees":["alex"]}'
|
|
assert result["assignees"] == ["alex"]
|
|
assert result["available"] is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_handoff_candidates_exclude_current_and_malformed_users():
|
|
async def handler(request):
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
assert request.url.path == "/api/v1/repos/stackchain/api/assignees"
|
|
return httpx.Response(200, json=[
|
|
{"login": "timmy", "full_name": "Timmy"},
|
|
{"login": "alex", "full_name": "Alexander"},
|
|
{"login": "casey", "full_name": ""},
|
|
{"login": ""},
|
|
{"full_name": "Missing login"},
|
|
"malformed",
|
|
])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.issue_handoff_candidates("stackchain/api")
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result == [
|
|
{"login": "alex", "name": "Alexander"},
|
|
{"login": "casey", "name": "casey"},
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_mention_candidates_match_login_or_name_and_are_bounded():
|
|
async def handler(request):
|
|
assert request.url.path == "/api/v1/repos/stackchain/api/assignees"
|
|
return httpx.Response(200, json=[
|
|
{"login": "alexa", "full_name": "Alexa Dev"},
|
|
{"login": "buildbot", "full_name": "Alex Builder"},
|
|
{"login": "alexb", "full_name": "Alex Backup"},
|
|
{"login": "casey", "full_name": "Casey"},
|
|
{"login": "AL", "full_name": "Al"},
|
|
{"login": "bad space", "full_name": "Malformed"},
|
|
{"full_name": "Missing login"},
|
|
])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.mention_candidates("stackchain/api", "alex", limit=2)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result == [
|
|
{"login": "alexa", "name": "Alexa Dev"},
|
|
{"login": "alexb", "name": "Alex Backup"},
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_handoff_replaces_operator_and_preserves_coassignees():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.url.path.endswith("/assignees"):
|
|
return httpx.Response(200, json=[
|
|
{"login": "alex", "full_name": "Alexander"},
|
|
{"login": "casey", "full_name": "Casey"},
|
|
])
|
|
if request.method == "GET":
|
|
return httpx.Response(200, json={
|
|
"number": 17, "state": "open", "pull_request": None,
|
|
"assignees": [{"login": "timmy"}, {"login": "casey"}],
|
|
})
|
|
assert request.content == b'{"assignees":["casey","alex"]}'
|
|
return httpx.Response(200, json={
|
|
"number": 17, "state": "open",
|
|
"assignees": [{"login": "casey"}, {"login": "alex"}],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.handoff_assigned_issue("stackchain/api", 17, "alex")
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result == {
|
|
"repository": "stackchain/api", "number": 17, "state": "open",
|
|
"assignees": ["casey", "alex"], "recipient": "alex",
|
|
}
|
|
assert [(request.method, request.url.path) for request in requests] == [
|
|
("GET", "/api/v1/user"),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/17"),
|
|
("GET", "/api/v1/user"),
|
|
("GET", "/api/v1/repos/stackchain/api/assignees"),
|
|
("PATCH", "/api/v1/repos/stackchain/api/issues/17"),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_release_issue_endpoint_invalidates_available_work_and_returns_confirmation(monkeypatch):
|
|
calls = []
|
|
|
|
async def release(repository, number):
|
|
calls.append((repository, number))
|
|
return {
|
|
"number": number, "repository": repository, "state": "open",
|
|
"assignees": [], "available": True,
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "release_assigned_issue", release, raising=False)
|
|
monkeypatch.setattr(main, "_available_issue_snapshot_value", [{"number": 99}])
|
|
monkeypatch.setattr(main, "_available_issue_snapshot_created_at", 123.0)
|
|
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/issues/17/release")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["available"] is True
|
|
assert calls == [("stackchain/api", 17)]
|
|
assert main._available_issue_snapshot_value is None
|
|
assert main._available_issue_snapshot_created_at is None
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_handoff_endpoints_list_candidates_and_confirm_transfer(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, "state": "open",
|
|
"assignees": [recipient], "recipient": recipient,
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "issue_handoff_candidates", candidates)
|
|
monkeypatch.setattr(main.gitea_proxy, "handoff_assigned_issue", handoff)
|
|
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/issues/17/handoff-candidates"
|
|
)
|
|
transferred = await client.patch(
|
|
"/api/v1/repos/stackchain/api/issues/17/handoff",
|
|
json={"recipient": "alex"},
|
|
)
|
|
|
|
assert listed.status_code == 200
|
|
assert listed.json() == [{"login": "alex", "name": "Alexander"}]
|
|
assert transferred.status_code == 200
|
|
assert transferred.json()["recipient"] == "alex"
|
|
assert listed.headers["cache-control"] == "no-store"
|
|
assert transferred.headers["cache-control"] == "no-store"
|
|
assert calls == [
|
|
("assigned", "stackchain/api", 17),
|
|
("candidates", "stackchain/api"),
|
|
("handoff", "stackchain/api", 17, "alex"),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_mention_candidate_endpoint_bounds_query_and_disables_caching(monkeypatch):
|
|
calls = []
|
|
|
|
async def candidates(repository, query, *, limit):
|
|
calls.append((repository, query, limit))
|
|
return [{"login": "alex", "name": "Alexander"}]
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "mention_candidates", candidates, raising=False)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
too_short = await client.get(
|
|
"/api/v1/repos/stackchain/api/mention-candidates?q=a"
|
|
)
|
|
response = await client.get(
|
|
"/api/v1/repos/stackchain/api/mention-candidates?q=AlEx"
|
|
)
|
|
|
|
assert too_short.status_code == 422
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == [{"login": "alex", "name": "Alexander"}]
|
|
assert calls == [("stackchain/api", "alex", 8)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_handoff_returns_conflict_when_assignment_or_recipient_changed(monkeypatch):
|
|
async def handoff(_repository, _number, _recipient):
|
|
raise gitea_proxy.IssueNotAvailableError("stale")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "handoff_assigned_issue", handoff)
|
|
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/issues/17/handoff",
|
|
json={"recipient": "alex"},
|
|
)
|
|
|
|
assert response.status_code == 409
|
|
assert response.json() == {
|
|
"error": "The issue or recipient changed. Reload before handing off."
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_handoff_candidates_reject_an_issue_not_assigned_to_operator(monkeypatch):
|
|
called = False
|
|
|
|
async def assigned(_repository, _number):
|
|
return False
|
|
|
|
async def candidates(_repository):
|
|
nonlocal called
|
|
called = True
|
|
return []
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "issue_handoff_candidates", candidates)
|
|
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/issues/17/handoff-candidates"
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_detail_endpoint_returns_assigned_issue_with_no_store(monkeypatch):
|
|
async def assigned(repository, number):
|
|
return (repository, number) == ("stackchain/api", 7)
|
|
|
|
async def detail(repository, number):
|
|
return {"repository": repository, "number": number, "title": "Fix mobile flow"}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned, raising=False)
|
|
monkeypatch.setattr(main.gitea_proxy, "issue_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/issues/7/detail")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {
|
|
"repository": "stackchain/api", "number": 7, "title": "Fix mobile flow"
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_assigned_issue_conversation_endpoint_is_authorized_bounded_and_no_store(monkeypatch):
|
|
calls = []
|
|
|
|
async def assigned(repository, number):
|
|
calls.append(("assigned", repository, number))
|
|
return repository == "stackchain/api"
|
|
|
|
async def conversation(repository, number, page, limit):
|
|
calls.append(("conversation", repository, number, page, limit))
|
|
return {"comments": [{"id": 21}], "page": 2, "older_page": 1, "total": 21}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", 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/issues/7/comments?page=2&limit=20"
|
|
)
|
|
missing = await client.get(
|
|
"/api/v1/repos/private/secret/issues/7/comments?page=2&limit=20"
|
|
)
|
|
invalid = await client.get(
|
|
"/api/v1/repos/stackchain/api/issues/7/comments?page=999&limit=999"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["older_page"] == 1
|
|
assert missing.status_code == 404
|
|
assert invalid.status_code == 422
|
|
assert calls == [
|
|
("assigned", "stackchain/api", 7),
|
|
("conversation", "stackchain/api", 7, 2, 20),
|
|
("assigned", "private/secret", 7),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_detail_deadline_is_retryable_sanitized_and_cancels_work(monkeypatch):
|
|
cancelled = asyncio.Event()
|
|
|
|
async def assigned(repository, number):
|
|
try:
|
|
await asyncio.Event().wait()
|
|
finally:
|
|
cancelled.set()
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main, "ISSUE_ACTION_TIMEOUT_SECONDS", 0.01)
|
|
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/issues/7/detail")
|
|
|
|
assert response.status_code == 503
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.headers["retry-after"] == "1"
|
|
assert response.json() == {"error": "Loading the issue timed out. Please retry."}
|
|
assert cancelled.is_set()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize(
|
|
("method", "path", "json_body"),
|
|
[
|
|
("GET", "/api/v1/repos/stackchain/api/issues/7/detail", None),
|
|
("POST", "/api/v1/repos/stackchain/api/issues/7/comments", {"body": "Hello"}),
|
|
("PATCH", "/api/v1/repos/stackchain/api/issues/7/close", None),
|
|
],
|
|
)
|
|
async def test_issue_actions_reject_items_not_assigned_to_service_user(
|
|
monkeypatch, method, path, json_body
|
|
):
|
|
async def assigned(repository, number):
|
|
return False
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.request(method, path, json=json_body)
|
|
|
|
assert response.status_code == 404
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("body", [" ", "x" * 10_001])
|
|
async def test_issue_comment_rejects_blank_or_oversized_body_before_upstream(
|
|
monkeypatch, body
|
|
):
|
|
called = False
|
|
|
|
async def assigned(repository, number):
|
|
nonlocal called
|
|
called = True
|
|
return True
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
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/issues/7/comments", json={"body": body}
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_comment_endpoint_posts_only_to_assigned_issue(monkeypatch):
|
|
calls = []
|
|
|
|
async def assigned(repository, number):
|
|
return True
|
|
|
|
async def comment(repository, number, body):
|
|
calls.append((repository, number, body))
|
|
return {"id": 82, "author": "timmy", "body": body}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", 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/issues/7/comments",
|
|
json={"body": " Ready to ship "},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {"id": 82, "author": "timmy", "body": "Ready to ship"}
|
|
assert calls == [("stackchain/api", 7, "Ready to ship")]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_close_endpoint_mutates_assigned_issue_only_after_confirmation(monkeypatch):
|
|
calls = []
|
|
|
|
async def assigned(repository, number):
|
|
return True
|
|
|
|
async def close(repository, number):
|
|
calls.append((repository, number))
|
|
return {"number": number, "state": "closed", "closed_at": "now"}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "close_issue", close)
|
|
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/issues/7/close")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["state"] == "closed"
|
|
assert calls == [("stackchain/api", 7)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_close_issue_patches_state_and_confirms_closed_response():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
return httpx.Response(
|
|
200,
|
|
json={"number": 7, "state": "closed", "closed_at": "2026-08-07T10:20:00Z"},
|
|
)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.close_issue("stackchain/api", 7)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(requests) == 1
|
|
assert requests[0].method == "PATCH"
|
|
assert requests[0].url.path == "/api/v1/repos/stackchain/api/issues/7"
|
|
assert requests[0].content == b'{"state":"closed"}'
|
|
assert result == {"number": 7, "state": "closed", "closed_at": "2026-08-07T10:20:00Z"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_issue_comment_posts_body_and_returns_safe_identity():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
return httpx.Response(
|
|
201,
|
|
json={
|
|
"id": 82,
|
|
"body": "Ready to ship",
|
|
"created_at": "2026-08-07T10:10:00Z",
|
|
"html_url": "https://forge.example/stackchain/api/issues/7#issuecomment-82",
|
|
"user": {"login": "timmy"},
|
|
},
|
|
)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.comment_on_issue("stackchain/api", 7, "Ready to ship")
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(requests) == 1
|
|
assert requests[0].method == "POST"
|
|
assert requests[0].url.path == "/api/v1/repos/stackchain/api/issues/7/comments"
|
|
assert requests[0].content == b'{"body":"Ready to ship"}'
|
|
assert result == {
|
|
"id": 82,
|
|
"author": "timmy",
|
|
"body": "Ready to ship",
|
|
"created_at": "2026-08-07T10:10:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/7#issuecomment-82",
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_issue_detail_returns_normalized_context_and_recent_comments():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append((request.method, request.url.path, request.url.query.decode()))
|
|
if request.url.path.endswith("/dependencies"):
|
|
return httpx.Response(
|
|
200,
|
|
json=[
|
|
{
|
|
"number": 3,
|
|
"title": "Restore signing service",
|
|
"state": "open",
|
|
"html_url": "https://forge.example/stackchain/platform/issues/3",
|
|
"repository": {"full_name": "stackchain/platform"},
|
|
},
|
|
{
|
|
"number": 2,
|
|
"title": "Completed prerequisite",
|
|
"state": "closed",
|
|
"repository": {"full_name": "stackchain/platform"},
|
|
},
|
|
],
|
|
)
|
|
if request.url.path.endswith("/comments"):
|
|
return httpx.Response(
|
|
200,
|
|
json=[
|
|
{
|
|
"id": 81,
|
|
"body": "Latest update",
|
|
"created_at": "2026-08-07T10:00:00Z",
|
|
"html_url": "https://forge.example/stackchain/api/issues/7#issuecomment-81",
|
|
"user": {"login": "sam"},
|
|
}
|
|
],
|
|
)
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"number": 7,
|
|
"title": "Fix mobile flow",
|
|
"state": "open",
|
|
"body": "Full issue context",
|
|
"updated_at": "2026-08-07T09:59:00Z",
|
|
"milestone": {"id": 9, "title": "August RC", "description": "not exposed"},
|
|
"html_url": "https://forge.example/stackchain/api/issues/7",
|
|
"labels": [{"name": "P1"}],
|
|
"assignees": [{"login": "timmy"}],
|
|
},
|
|
)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.issue_detail("stackchain/api", 7)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests == [
|
|
("GET", "/api/v1/repos/stackchain/api/issues/7", ""),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/7/comments", "limit=20&page=1"),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/7/dependencies", "limit=20"),
|
|
]
|
|
assert result == {
|
|
"repository": "stackchain/api",
|
|
"number": 7,
|
|
"title": "Fix mobile flow",
|
|
"state": "open",
|
|
"body": "Full issue context",
|
|
"updated_at": "2026-08-07T09:59:00Z",
|
|
"due_date": None,
|
|
"milestone": {"id": 9, "title": "August RC"},
|
|
"url": "https://forge.example/stackchain/api/issues/7",
|
|
"labels": ["P1"],
|
|
"assignees": ["timmy"],
|
|
"dependencies_available": True,
|
|
"dependencies": [
|
|
{
|
|
"repository": "stackchain/platform",
|
|
"number": 3,
|
|
"title": "Restore signing service",
|
|
"state": "open",
|
|
"url": "https://forge.example/stackchain/platform/issues/3",
|
|
}
|
|
],
|
|
"comments": [
|
|
{
|
|
"id": 81,
|
|
"author": "sam",
|
|
"body": "Latest update",
|
|
"created_at": "2026-08-07T10:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/7#issuecomment-81",
|
|
}
|
|
],
|
|
"conversation": {
|
|
"comments": [
|
|
{
|
|
"id": 81,
|
|
"author": "sam",
|
|
"body": "Latest update",
|
|
"created_at": "2026-08-07T10:00:00Z",
|
|
"url": "https://forge.example/stackchain/api/issues/7#issuecomment-81",
|
|
}
|
|
],
|
|
"page": 1,
|
|
"older_page": None,
|
|
"total": 1,
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_issue_detail_keeps_planning_usable_when_dependencies_are_unavailable():
|
|
async def handler(request):
|
|
if request.url.path.endswith("/dependencies"):
|
|
return httpx.Response(503, json={"message": "upstream timeout"})
|
|
if request.url.path.endswith("/comments"):
|
|
return httpx.Response(200, json=[])
|
|
return httpx.Response(200, json={
|
|
"number": 7,
|
|
"title": "Fix mobile flow",
|
|
"state": "open",
|
|
"assignees": [{"login": "timmy"}],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.issue_detail("stackchain/api", 7)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result["title"] == "Fix mobile flow"
|
|
assert result["dependencies_available"] is False
|
|
assert result["dependencies"] == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_conversation_page_opens_newest_page_and_reports_older_history():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append((request.url.path, request.url.query.decode()))
|
|
page = request.url.params.get("page")
|
|
comments = {
|
|
"1": [{"id": value, "body": f"Comment {value}", "user": {"login": "sam"}}
|
|
for value in range(1, 21)],
|
|
"3": [{"id": value, "body": f"Comment {value}", "user": {"login": "sam"}}
|
|
for value in range(41, 48)],
|
|
}[page]
|
|
return httpx.Response(200, json=comments, headers={"X-Total-Count": "47"})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.issue_conversation_page("stackchain/api", 7)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests == [
|
|
("/api/v1/repos/stackchain/api/issues/7/comments", "limit=20&page=1"),
|
|
("/api/v1/repos/stackchain/api/issues/7/comments", "limit=20&page=3"),
|
|
]
|
|
assert [comment["id"] for comment in result["comments"]] == list(range(41, 48))
|
|
assert result["page"] == 3
|
|
assert result["older_page"] == 2
|
|
assert result["total"] == 47
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_conversation_page_locally_bounds_unpaginated_gitea_comments():
|
|
requests = []
|
|
all_comments = [
|
|
{"id": value, "body": f"Comment {value}", "user": {"login": "sam"}}
|
|
for value in range(1, 48)
|
|
]
|
|
|
|
async def handler(request):
|
|
requests.append(request.url.query.decode())
|
|
return httpx.Response(200, json=all_comments, headers={"X-Total-Count": "47"})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.issue_conversation_page("stackchain/api", 7)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests == ["limit=20&page=1"]
|
|
assert [comment["id"] for comment in result["comments"]] == list(range(41, 48))
|
|
assert result["page"] == 3
|
|
assert result["older_page"] == 2
|
|
assert result["total"] == 47
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_update_assigned_issue_labels_validates_and_returns_confirmed_labels(monkeypatch):
|
|
calls = []
|
|
|
|
async def assigned(repository, number):
|
|
return (repository, number) == ("stackchain/api", 7)
|
|
|
|
async def labels(repository):
|
|
assert repository == "stackchain/api"
|
|
return [
|
|
{"id": 3, "name": "P0", "color": "d73a4a"},
|
|
{"id": 8, "name": "frontend", "color": "1d76db"},
|
|
]
|
|
|
|
async def update(repository, number, label_ids):
|
|
calls.append((repository, number, label_ids))
|
|
return {"number": number, "labels": ["P0", "frontend"]}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "repo_labels", labels)
|
|
monkeypatch.setattr(main.gitea_proxy, "update_issue_labels", 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/issues/7/labels",
|
|
json={"label_ids": [3, 8]},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {"number": 7, "labels": ["P0", "frontend"]}
|
|
assert calls == [("stackchain/api", 7, [3, 8])]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_assigned_issue_label_options_do_not_depend_on_repository_listing(monkeypatch):
|
|
calls = []
|
|
|
|
async def assigned(repository, number):
|
|
calls.append(("assigned", repository, number))
|
|
return True
|
|
|
|
async def labels(repository):
|
|
calls.append(("labels", repository))
|
|
return [{"id": 3, "name": "P0", "color": "d73a4a", "description": "Urgent"}]
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
|
|
monkeypatch.setattr(main.gitea_proxy, "repo_labels", labels)
|
|
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/issues/7/labels")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()[0]["name"] == "P0"
|
|
assert calls == [
|
|
("assigned", "stackchain/api", 7),
|
|
("labels", "stackchain/api"),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_update_issue_labels_patches_ids_and_normalizes_confirmation():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
return httpx.Response(200, json={
|
|
"number": 7,
|
|
"labels": [
|
|
{"id": 3, "name": "P0"},
|
|
{"id": 8, "name": "frontend"},
|
|
],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.update_issue_labels("stackchain/api", 7, [3, 8])
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(requests) == 1
|
|
assert requests[0].method == "PATCH"
|
|
assert requests[0].url.path == "/api/v1/repos/stackchain/api/issues/7"
|
|
assert requests[0].content == b'{"labels":[3,8]}'
|
|
assert result == {"number": 7, "labels": ["P0", "frontend"]}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_update_issue_labels_rejects_unconfirmed_label_set():
|
|
async def handler(_request):
|
|
return httpx.Response(200, json={
|
|
"number": 7,
|
|
"labels": [{"id": 8, "name": "frontend"}],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
with pytest.raises(ValueError, match="confirm"):
|
|
await gitea_proxy.update_issue_labels("stackchain/api", 7, [3])
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_assigned_issue_blocker_route_returns_canonical_dependencies(monkeypatch):
|
|
calls = []
|
|
|
|
async def mutate(repository, number, blocker_repository, blocker_number, remove=False):
|
|
calls.append((repository, number, blocker_repository, blocker_number, remove))
|
|
return {
|
|
"repository": repository, "number": number, "dependencies_available": True,
|
|
"dependencies": [] if remove else [{
|
|
"repository": blocker_repository, "number": blocker_number,
|
|
"title": "Restore API", "state": "open",
|
|
}],
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "mutate_assigned_issue_dependency", mutate, raising=False)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
added = await client.post(
|
|
"/api/v1/repos/stackchain/dashboard/issues/17/blockers",
|
|
json={"repository": "stackchain/api", "number": 9},
|
|
)
|
|
removed = await client.request(
|
|
"DELETE", "/api/v1/repos/stackchain/dashboard/issues/17/blockers",
|
|
json={"repository": "stackchain/api", "number": 9},
|
|
)
|
|
|
|
assert [added.status_code, removed.status_code] == [200, 200]
|
|
assert added.headers["cache-control"] == "no-store"
|
|
assert added.json()["dependencies"][0]["number"] == 9
|
|
assert removed.json()["dependencies"] == []
|
|
assert calls == [
|
|
("stackchain/dashboard", 17, "stackchain/api", 9, False),
|
|
("stackchain/dashboard", 17, "stackchain/api", 9, True),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_assigned_issue_blocker_patch_converges_to_desired_state(monkeypatch):
|
|
calls = []
|
|
|
|
async def mutate(repository, number, blocker_repository, blocker_number, remove=False):
|
|
calls.append((repository, number, blocker_repository, blocker_number, remove))
|
|
return {
|
|
"repository": repository, "number": number, "dependencies_available": True,
|
|
"dependencies": [] if remove else [{"repository": blocker_repository, "number": blocker_number}],
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "mutate_assigned_issue_dependency", mutate, 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/dashboard/issues/17/blockers",
|
|
headers={"Idempotency-Key": "offline-blocker-1"},
|
|
json={"repository": "stackchain/api", "number": 9, "present": False},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["dependencies"] == []
|
|
assert calls == [("stackchain/dashboard", 17, "stackchain/api", 9, True)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_assigned_issue_blocker_route_rejects_invalid_candidate(monkeypatch):
|
|
async def mutate(*_args, **_kwargs):
|
|
raise gitea_proxy.IssueDependencyInvalidError("blocker must be an open issue")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "mutate_assigned_issue_dependency", mutate, 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/dashboard/issues/17/blockers",
|
|
json={"repository": "stackchain/api", "number": 9},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
assert response.json() == {"error": "blocker must be an open issue"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_add_dependency_validates_assignment_candidate_and_confirmation():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(request)
|
|
path = request.url.path
|
|
if path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if path == "/api/v1/repos/stackchain/dashboard/issues/17":
|
|
return httpx.Response(200, json={
|
|
"number": 17, "state": "open", "assignees": [{"login": "timmy"}],
|
|
})
|
|
if path == "/api/v1/repos/stackchain/api/issues/9":
|
|
return httpx.Response(200, json={"number": 9, "state": "open"})
|
|
if request.method == "GET" and path.endswith("/dependencies"):
|
|
mutated = any(r.method == "POST" for r in requests)
|
|
return httpx.Response(200, json=[{
|
|
"number": 9, "state": "open", "title": "Restore API",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"html_url": "https://forge.example/stackchain/api/issues/9",
|
|
}] if mutated else [])
|
|
if request.method == "POST" and path.endswith("/dependencies"):
|
|
return httpx.Response(201, json={})
|
|
raise AssertionError((request.method, path))
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.mutate_assigned_issue_dependency(
|
|
"stackchain/dashboard", 17, "stackchain/api", 9
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
mutation = next(request for request in requests if request.method == "POST")
|
|
assert mutation.url.path == "/api/v1/repos/stackchain/dashboard/issues/17/dependencies"
|
|
assert json.loads(mutation.content) == {"owner": "stackchain", "repo": "api", "index": 9}
|
|
assert result["dependencies"] == [{
|
|
"repository": "stackchain/api", "number": 9, "title": "Restore API",
|
|
"state": "open", "url": "https://forge.example/stackchain/api/issues/9",
|
|
}]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gitea_dependency_retry_returns_canonical_state_without_duplicate_mutation():
|
|
mutations = []
|
|
dependency = {
|
|
"number": 9, "state": "open", "title": "Restore API",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"html_url": "https://forge.example/stackchain/api/issues/9",
|
|
}
|
|
|
|
async def handler(request):
|
|
path = request.url.path
|
|
if path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if path == "/api/v1/repos/stackchain/dashboard/issues/17":
|
|
return httpx.Response(200, json={
|
|
"number": 17, "state": "open", "assignees": [{"login": "timmy"}],
|
|
})
|
|
if path == "/api/v1/repos/stackchain/api/issues/9":
|
|
return httpx.Response(200, json={"number": 9, "state": "open"})
|
|
if request.method == "GET" and path.endswith("/dependencies"):
|
|
return httpx.Response(200, json=[dependency])
|
|
if path.endswith("/dependencies"):
|
|
mutations.append(request.method)
|
|
return httpx.Response(201, json={})
|
|
raise AssertionError((request.method, path))
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.mutate_assigned_issue_dependency(
|
|
"stackchain/dashboard", 17, "stackchain/api", 9
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert mutations == []
|
|
assert result["dependencies"][0]["number"] == 9
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("candidate,blocker_repository,blocker_number,existing", [
|
|
({"number": 17, "state": "open"}, "stackchain/dashboard", 17, []),
|
|
({"number": 9, "state": "closed"}, "stackchain/api", 9, []),
|
|
({"number": 9, "state": "open", "pull_request": {}}, "stackchain/api", 9, []),
|
|
])
|
|
async def test_gitea_dependency_rejects_invalid_candidates(
|
|
candidate, blocker_repository, blocker_number, existing
|
|
):
|
|
async def handler(request):
|
|
if request.url.path == "/api/v1/user":
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.url.path == "/api/v1/repos/stackchain/dashboard/issues/17":
|
|
return httpx.Response(200, json={
|
|
"number": 17, "state": "open", "assignees": [{"login": "timmy"}],
|
|
})
|
|
if request.url.path.endswith("/dependencies"):
|
|
return httpx.Response(200, json=existing)
|
|
return httpx.Response(200, json=candidate)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
with pytest.raises(gitea_proxy.IssueDependencyInvalidError):
|
|
await gitea_proxy.mutate_assigned_issue_dependency(
|
|
"stackchain/dashboard", 17, blocker_repository, blocker_number
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|