662 lines
24 KiB
Python
662 lines
24 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy, main
|
|
|
|
|
|
@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, "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_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_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, "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_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, "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": [], "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_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. Refresh Find Work."
|
|
}
|
|
|
|
|
|
@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_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("/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",
|
|
"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"),
|
|
]
|
|
assert result == {
|
|
"repository": "stackchain/api",
|
|
"number": 7,
|
|
"title": "Fix mobile flow",
|
|
"state": "open",
|
|
"body": "Full issue context",
|
|
"url": "https://forge.example/stackchain/api/issues/7",
|
|
"labels": ["P1"],
|
|
"assignees": ["timmy"],
|
|
"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",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
@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()
|