936 lines
34 KiB
Python
936 lines
34 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
import httpx
|
|
|
|
from src import gitea_proxy
|
|
from src import main
|
|
from src.available_issue_snapshot_store import AvailableIssueSnapshotStore
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_available_issue_snapshot(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(
|
|
main,
|
|
"_available_issue_snapshot_store",
|
|
AvailableIssueSnapshotStore(tmp_path / "available-issues.sqlite3"),
|
|
)
|
|
main._available_issue_snapshot_task = None
|
|
main._available_issue_snapshot_lock = asyncio.Lock()
|
|
main._available_issue_snapshot_value = None
|
|
main._available_issue_snapshot_created_at = None
|
|
main._available_issue_snapshot_retry_at = None
|
|
yield
|
|
task = main._available_issue_snapshot_task
|
|
if task is not None and not task.done():
|
|
task.cancel()
|
|
main._available_issue_snapshot_task = None
|
|
main._available_issue_snapshot_lock = asyncio.Lock()
|
|
main._available_issue_snapshot_value = None
|
|
main._available_issue_snapshot_created_at = None
|
|
main._available_issue_snapshot_retry_at = None
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_work_page_preserves_total_and_reason_without_loading_other_pages():
|
|
requests = []
|
|
|
|
def upstream(request):
|
|
requests.append(str(request.url))
|
|
return httpx.Response(
|
|
200,
|
|
headers={"X-Total-Count": "84"},
|
|
json=[{
|
|
"id": 51,
|
|
"number": 51,
|
|
"title": "Older review",
|
|
"state": "open",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"html_url": "https://forge.example/stackchain/api/pulls/51",
|
|
}],
|
|
)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.work_page("review", page=2)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests == [
|
|
"http://127.0.0.1:3000/api/v1/repos/issues/search?state=open&review_requested=true&type=pulls&limit=50&page=2"
|
|
]
|
|
assert result["page"] == 2
|
|
assert result["total"] == 84
|
|
assert result["has_more"] is False
|
|
assert result["items"][0]["work_reasons"] == ["review_requested"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_page_filters_assigned_and_pull_items_then_ranks_priority():
|
|
requests = []
|
|
|
|
def upstream(request):
|
|
requests.append(str(request.url))
|
|
return httpx.Response(
|
|
200,
|
|
headers={"X-Total-Count": "4"},
|
|
json=[
|
|
{"id": 1, "number": 1, "title": "Ordinary", "state": "open",
|
|
"updated_at": "2026-08-07T12:00:00Z", "assignees": None,
|
|
"pull_request": None,
|
|
"labels": [], "repository": {"full_name": "stackchain/api"}},
|
|
{"id": 2, "number": 2, "title": "Claimed", "state": "open",
|
|
"assignees": [{"login": "alex"}], "repository": {"full_name": "stackchain/api"}},
|
|
{"id": 3, "number": 3, "title": "A pull", "state": "open",
|
|
"assignees": [], "pull_request": {"merged": False},
|
|
"repository": {"full_name": "stackchain/api"}},
|
|
{"id": 4, "number": 4, "title": "Critical", "state": "open",
|
|
"updated_at": "2026-08-07T10:00:00Z", "assignees": [],
|
|
"pull_request": None,
|
|
"labels": [{"name": "critical"}],
|
|
"repository": {"full_name": "stackchain/web"}},
|
|
],
|
|
)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.available_issue_page(page=1)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests == [
|
|
"http://127.0.0.1:3000/api/v1/repos/issues/search?state=open&type=issues&limit=50&page=1"
|
|
]
|
|
assert [item["title"] for item in result["items"]] == ["Critical", "Ordinary"]
|
|
assert result == {
|
|
"items": result["items"], "page": 1, "total": 2, "has_more": False,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_page_ranks_all_upstream_pages_before_logical_pagination():
|
|
requests = []
|
|
|
|
def issue(number, *, assigned=True, critical=False, updated_at="2026-08-07T12:00:00Z"):
|
|
return {
|
|
"id": number, "number": number, "title": f"Issue {number}", "state": "open",
|
|
"updated_at": updated_at, "assignees": [{"login": "alex"}] if assigned else None,
|
|
"pull_request": None,
|
|
"labels": [{"name": "critical"}] if critical else [],
|
|
"repository": {"full_name": "stackchain/api"},
|
|
}
|
|
|
|
first_page = [issue(number) for number in range(1, 51)]
|
|
first_page[0] = issue(1, assigned=False, updated_at="2026-08-07T13:00:00Z")
|
|
|
|
def upstream(request):
|
|
requests.append(str(request.url))
|
|
page = int(request.url.params["page"])
|
|
payload = first_page if page == 1 else [
|
|
issue(51, assigned=False, critical=True, updated_at="2026-08-07T10:00:00Z"),
|
|
issue(52, assigned=False, updated_at="2026-08-07T11:00:00Z"),
|
|
]
|
|
return httpx.Response(200, headers={"X-Total-Count": "52"}, json=payload)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.available_issue_page(page=1, limit=2)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(requests) == 2
|
|
assert [item["number"] for item in result["items"]] == [51, 1]
|
|
assert result == {
|
|
"items": result["items"], "page": 1, "total": 3, "has_more": True,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_snapshot_loads_known_remaining_pages_concurrently():
|
|
active = 0
|
|
peak_active = 0
|
|
remaining_started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def upstream(request):
|
|
nonlocal active, peak_active
|
|
page = int(request.url.params["page"])
|
|
if page == 1:
|
|
return httpx.Response(
|
|
200, headers={"X-Total-Count": "200"},
|
|
json=[{
|
|
"id": number, "number": number, "title": f"Issue {number}",
|
|
"state": "open", "assignees": [], "pull_request": None,
|
|
"labels": [], "repository": {"full_name": "stackchain/api"},
|
|
} for number in range(1, 51)],
|
|
)
|
|
active += 1
|
|
peak_active = max(peak_active, active)
|
|
if peak_active == 3:
|
|
remaining_started.set()
|
|
await release.wait()
|
|
active -= 1
|
|
start = (page - 1) * 50 + 1
|
|
return httpx.Response(200, headers={"X-Total-Count": "200"}, json=[{
|
|
"id": number, "number": number, "title": f"Issue {number}",
|
|
"state": "open", "assignees": [], "pull_request": None,
|
|
"labels": [], "repository": {"full_name": "stackchain/api"},
|
|
} for number in range(start, start + 50)])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
task = asyncio.create_task(gitea_proxy.available_issue_snapshot())
|
|
try:
|
|
await asyncio.wait_for(remaining_started.wait(), timeout=1)
|
|
assert peak_active == 3
|
|
release.set()
|
|
result = await asyncio.wait_for(task, timeout=1)
|
|
finally:
|
|
release.set()
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(result) == 200
|
|
assert peak_active == 3
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_endpoint_is_bounded_retryable_and_no_store(monkeypatch):
|
|
calls = []
|
|
|
|
async def available():
|
|
calls.append(True)
|
|
return [{"number": number} for number in range(1, 52)]
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", available)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/available-issues?page=1")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {
|
|
"items": [{"number": number} for number in range(1, 51)],
|
|
"page": 1, "total": 51, "has_more": True,
|
|
}
|
|
assert calls == [True]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_search_filters_full_catalog_before_pagination(monkeypatch):
|
|
async def available():
|
|
return [
|
|
{
|
|
"repository": "stackchain/api",
|
|
"number": number,
|
|
"title": "Routine API maintenance",
|
|
}
|
|
for number in range(1, 52)
|
|
] + [{
|
|
"repository": "stackchain/dashboard",
|
|
"number": 673,
|
|
"title": "Search the full Find Work catalog",
|
|
}]
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", available)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
by_repo = await client.get("/api/v1/available-issues?page=1&q=DASHBOARD")
|
|
by_number = await client.get("/api/v1/available-issues?page=1&q=%23673")
|
|
by_title = await client.get("/api/v1/available-issues?page=1&q=full%20find")
|
|
|
|
expected = [{
|
|
"repository": "stackchain/dashboard",
|
|
"number": 673,
|
|
"title": "Search the full Find Work catalog",
|
|
}]
|
|
assert by_repo.json() == {"items": expected, "page": 1, "total": 1, "has_more": False}
|
|
assert by_number.json() == by_repo.json()
|
|
assert by_title.json() == by_repo.json()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_facets_filter_before_pagination_and_describe_catalog(monkeypatch):
|
|
async def available():
|
|
return [
|
|
{"repository": "stackchain/api", "number": 1, "title": "Critical API", "labels": ["critical", "backend"]},
|
|
{"repository": "stackchain/api", "number": 2, "title": "API docs", "labels": ["docs"]},
|
|
{"repository": "stackchain/dashboard", "number": 3, "title": "Critical UI", "labels": ["critical", "frontend"]},
|
|
]
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", available)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/available-issues",
|
|
params=[("repository", "stackchain/api"), ("label", "critical")],
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"items": [{"repository": "stackchain/api", "number": 1, "title": "Critical API", "labels": ["critical", "backend"]}],
|
|
"page": 1,
|
|
"total": 1,
|
|
"has_more": False,
|
|
"facets": {
|
|
"repositories": ["stackchain/api", "stackchain/dashboard"],
|
|
"labels": ["backend", "critical", "docs", "frontend"],
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_facets_or_within_each_facet_and_compose_with_search(monkeypatch):
|
|
async def available():
|
|
return [
|
|
{"repository": "stackchain/api", "number": 1, "title": "Repair worker", "labels": ["backend"]},
|
|
{"repository": "stackchain/web", "number": 2, "title": "Repair mobile", "labels": ["frontend"]},
|
|
{"repository": "stackchain/docs", "number": 3, "title": "Repair guide", "labels": ["docs"]},
|
|
{"repository": "stackchain/api", "number": 4, "title": "Routine task", "labels": ["backend"]},
|
|
]
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", available)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/available-issues",
|
|
params=[("q", "repair"), ("repository", "stackchain/api"), ("repository", "stackchain/web"),
|
|
("label", "backend"), ("label", "frontend")],
|
|
)
|
|
|
|
assert [item["number"] for item in response.json()["items"]] == [1, 2]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_facets_reject_excess_values_before_scanning(monkeypatch):
|
|
async def must_not_scan():
|
|
raise AssertionError("invalid facets must be rejected before catalog scan")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", must_not_scan)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/available-issues",
|
|
params=[("label", f"label-{index}") for index in range(11)],
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_search_rejects_oversized_query_without_scanning(monkeypatch):
|
|
async def must_not_scan():
|
|
raise AssertionError("invalid query must be rejected before catalog scan")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", must_not_scan)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/available-issues", params={"q": "x" * 101})
|
|
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_endpoint_reuses_catalog_published_by_another_worker(
|
|
monkeypatch, tmp_path
|
|
):
|
|
store = AvailableIssueSnapshotStore(tmp_path / "available.sqlite3", clock=lambda: 100.0)
|
|
owner = store.try_acquire_refresh(lease_seconds=5)
|
|
store.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])
|
|
monkeypatch.setattr(main, "_available_issue_snapshot_store", store, raising=False)
|
|
monkeypatch.setattr(main.time, "time", lambda: 100.0)
|
|
|
|
async def must_not_scan():
|
|
raise AssertionError("fresh shared catalog should avoid an upstream scan")
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", must_not_scan)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/available-issues?page=1")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"items": [{"repository": "stackchain/api", "number": 7}],
|
|
"page": 1, "total": 1, "has_more": False,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_endpoint_waits_for_slow_cold_catalog_from_another_worker(
|
|
monkeypatch, tmp_path
|
|
):
|
|
now = 100.0
|
|
path = tmp_path / "shared-available.sqlite3"
|
|
owner_store = AvailableIssueSnapshotStore(path, clock=lambda: now)
|
|
worker_store = AvailableIssueSnapshotStore(path, clock=lambda: now)
|
|
owner = owner_store.try_acquire_refresh(lease_seconds=6)
|
|
monkeypatch.setattr(main, "_available_issue_snapshot_store", worker_store)
|
|
|
|
sleep_calls = 0
|
|
|
|
async def publish_after_existing_one_second_window(_delay):
|
|
nonlocal sleep_calls
|
|
sleep_calls += 1
|
|
if sleep_calls == 60:
|
|
owner_store.publish(
|
|
owner,
|
|
items=[{"repository": "stackchain/api", "number": 8}],
|
|
)
|
|
|
|
async def must_not_scan():
|
|
raise AssertionError("a valid cross-worker lease must prevent a duplicate scan")
|
|
|
|
monkeypatch.setattr(main.asyncio, "sleep", publish_after_existing_one_second_window)
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", must_not_scan)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/available-issues?page=1")
|
|
|
|
assert sleep_calls > 50
|
|
assert response.status_code == 200
|
|
assert response.json()["items"] == [
|
|
{"repository": "stackchain/api", "number": 8}
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_endpoint_takes_over_released_cold_refresh(
|
|
monkeypatch, tmp_path
|
|
):
|
|
now = 100.0
|
|
path = tmp_path / "shared-available.sqlite3"
|
|
owner_store = AvailableIssueSnapshotStore(path, clock=lambda: now)
|
|
worker_store = AvailableIssueSnapshotStore(path, clock=lambda: now)
|
|
owner = owner_store.try_acquire_refresh(lease_seconds=6)
|
|
monkeypatch.setattr(main, "_available_issue_snapshot_store", worker_store)
|
|
|
|
sleep_calls = 0
|
|
scan_calls = 0
|
|
|
|
async def release_failed_owner(_delay):
|
|
nonlocal sleep_calls
|
|
sleep_calls += 1
|
|
if sleep_calls == 3:
|
|
owner_store.release_refresh(owner)
|
|
|
|
async def available():
|
|
nonlocal scan_calls
|
|
scan_calls += 1
|
|
return [{"repository": "stackchain/web", "number": 9}]
|
|
|
|
monkeypatch.setattr(main.asyncio, "sleep", release_failed_owner)
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", available)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/available-issues?page=1")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["items"] == [
|
|
{"repository": "stackchain/web", "number": 9}
|
|
]
|
|
assert scan_calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_endpoint_coalesces_cold_scan_and_reuses_it_for_pages(monkeypatch):
|
|
calls = 0
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def available():
|
|
nonlocal calls
|
|
calls += 1
|
|
started.set()
|
|
await release.wait()
|
|
return [{"number": number} for number in range(1, 64)]
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", available)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first = asyncio.create_task(client.get("/api/v1/available-issues?page=1"))
|
|
second = asyncio.create_task(client.get("/api/v1/available-issues?page=2"))
|
|
await asyncio.wait_for(started.wait(), timeout=1)
|
|
release.set()
|
|
first_response, second_response = await asyncio.gather(first, second)
|
|
repeated_response = await client.get("/api/v1/available-issues?page=1")
|
|
|
|
assert calls == 1
|
|
assert len(first_response.json()["items"]) == 50
|
|
assert len(second_response.json()["items"]) == 13
|
|
assert second_response.json()["total"] == 63
|
|
assert second_response.json()["has_more"] is False
|
|
assert repeated_response.status_code == 200
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_endpoint_retains_last_snapshot_on_refresh_failure(monkeypatch):
|
|
async def unavailable():
|
|
raise httpx.ConnectError("offline")
|
|
|
|
main._available_issue_snapshot_value = [{"number": 7}]
|
|
main._available_issue_snapshot_created_at = 0.0
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", unavailable)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/available-issues?page=1")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"items": [{"number": 7}], "page": 1, "total": 1,
|
|
"has_more": False, "stale": True, "revalidating": True,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_endpoint_serves_expired_snapshot_while_one_refresh_runs(monkeypatch):
|
|
calls = 0
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
|
|
async def refresh():
|
|
nonlocal calls
|
|
calls += 1
|
|
started.set()
|
|
await release.wait()
|
|
return [{"number": 8}]
|
|
|
|
main._available_issue_snapshot_value = [{"number": 7}]
|
|
main._available_issue_snapshot_created_at = 0.0
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", refresh)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first, second = await asyncio.gather(
|
|
client.get("/api/v1/available-issues?page=1"),
|
|
client.get("/api/v1/available-issues?page=1"),
|
|
)
|
|
await asyncio.wait_for(started.wait(), timeout=1)
|
|
|
|
assert calls == 1
|
|
assert first.json() == second.json() == {
|
|
"items": [{"number": 7}], "page": 1, "total": 1,
|
|
"has_more": False, "stale": True, "revalidating": True,
|
|
}
|
|
release.set()
|
|
await asyncio.wait_for(main._available_issue_snapshot_task, timeout=1)
|
|
assert main._available_issue_snapshot_value == [{"number": 8}]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_available_issue_endpoint_backs_off_after_background_refresh_failure(monkeypatch):
|
|
calls = 0
|
|
|
|
async def unavailable():
|
|
nonlocal calls
|
|
calls += 1
|
|
raise httpx.ConnectError("offline")
|
|
|
|
main._available_issue_snapshot_value = [{"number": 7}]
|
|
main._available_issue_snapshot_created_at = 0.0
|
|
monkeypatch.setattr(main.gitea_proxy, "available_issue_snapshot", unavailable)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first = await client.get("/api/v1/available-issues?page=1")
|
|
await asyncio.gather(main._available_issue_snapshot_task, return_exceptions=True)
|
|
second = await client.get("/api/v1/available-issues?page=1")
|
|
|
|
assert calls == 1
|
|
assert first.json()["revalidating"] is True
|
|
assert second.json() == {
|
|
"items": [{"number": 7}], "page": 1, "total": 1,
|
|
"has_more": False, "stale": True, "refresh_failed": True,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_confirmed_claim_is_removed_from_retained_available_snapshot(monkeypatch):
|
|
main._available_issue_snapshot_value = [
|
|
{"number": 7, "repository": "stackchain/api"},
|
|
{"number": 8, "repository": "stackchain/web"},
|
|
]
|
|
main._available_issue_snapshot_created_at = 10**12
|
|
|
|
async def claim(repository, number):
|
|
return {"repository": repository, "number": number, "assignees": ["timmy"]}
|
|
|
|
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:
|
|
claimed = await client.patch("/api/v1/repos/stackchain/api/issues/7/claim")
|
|
available = await client.get("/api/v1/available-issues?page=1")
|
|
|
|
assert claimed.status_code == 200
|
|
assert available.json() == {
|
|
"items": [{"number": 8, "repository": "stackchain/web"}],
|
|
"page": 1, "total": 1, "has_more": False,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_initial_work_collections_expose_independent_pagination(monkeypatch):
|
|
async def fake_page(stream, page=1, limit=50):
|
|
assert page == 1
|
|
totals = {"issue": 84, "pull": 61, "review": 73}
|
|
return {
|
|
"items": [], "page": 1, "total": totals[stream],
|
|
"has_more": True, "stream": stream,
|
|
}
|
|
|
|
monkeypatch.setattr(gitea_proxy, "work_page", fake_page)
|
|
|
|
assigned_issues = await gitea_proxy.issues()
|
|
pulls = await gitea_proxy.pull_requests()
|
|
|
|
assert assigned_issues.pagination == {
|
|
"issue": {"page": 1, "total": 84, "has_more": True}
|
|
}
|
|
assert pulls.pagination == {
|
|
"pull": {"page": 1, "total": 61, "has_more": True},
|
|
"review": {"page": 1, "total": 73, "has_more": True},
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_work_collections_include_supported_review_request_search(monkeypatch):
|
|
requested_streams = []
|
|
|
|
async def fake_page(stream, page=1, limit=50):
|
|
requested_streams.append((stream, page, limit))
|
|
return {"stream": stream, "items": [], "page": page, "total": 0, "has_more": False}
|
|
|
|
monkeypatch.setattr(gitea_proxy, "work_page", fake_page)
|
|
|
|
assert await gitea_proxy.issues() == []
|
|
assert await gitea_proxy.pull_requests() == []
|
|
assert requested_streams == [
|
|
("issue", 1, 50),
|
|
("pull", 1, 50),
|
|
("review", 1, 50),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pull_requests_merge_assignment_and_review_responsibilities(monkeypatch):
|
|
assigned = {
|
|
"id": 11,
|
|
"number": 7,
|
|
"title": "Review API",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
}
|
|
review_only = {
|
|
"id": 12,
|
|
"number": 8,
|
|
"title": "Review mobile",
|
|
"repository": {"full_name": "stackchain/mobile"},
|
|
}
|
|
|
|
async def fake_page(stream, page=1, limit=50):
|
|
items = [assigned] if stream == "pull" else [assigned.copy(), review_only]
|
|
reason = "assigned_to_me" if stream == "pull" else "review_requested"
|
|
return {
|
|
"stream": stream,
|
|
"items": [{**item, "work_reasons": [reason]} for item in items],
|
|
"page": page, "total": len(items), "has_more": False,
|
|
}
|
|
|
|
monkeypatch.setattr(gitea_proxy, "work_page", fake_page)
|
|
|
|
pulls = await gitea_proxy.pull_requests()
|
|
|
|
assert [pull["id"] for pull in pulls] == [11, 12]
|
|
assert pulls[0]["work_reasons"] == ["assigned_to_me", "review_requested"]
|
|
assert pulls[1]["work_reasons"] == ["review_requested"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pull_request_searches_start_concurrently(monkeypatch):
|
|
started = [asyncio.Event(), asyncio.Event()]
|
|
release = asyncio.Event()
|
|
|
|
async def fake_page(stream, page=1, limit=50):
|
|
index = 0 if stream == "pull" else 1
|
|
started[index].set()
|
|
await release.wait()
|
|
return {"stream": stream, "items": [], "page": page, "total": 0, "has_more": False}
|
|
|
|
monkeypatch.setattr(gitea_proxy, "work_page", fake_page)
|
|
task = asyncio.create_task(gitea_proxy.pull_requests())
|
|
try:
|
|
await asyncio.wait_for(
|
|
asyncio.gather(*(event.wait() for event in started)), timeout=1
|
|
)
|
|
finally:
|
|
release.set()
|
|
|
|
assert await task == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_requested_review_guard_uses_direct_pull_and_current_user(monkeypatch):
|
|
requested = []
|
|
|
|
async def fake_fetch(path):
|
|
requested.append(path)
|
|
if path == "user":
|
|
return {"login": "timmy"}
|
|
assert path == "repos/stackchain/api/pulls/77"
|
|
return {
|
|
"state": "open",
|
|
"requested_reviewers": [{"login": "timmy"}],
|
|
}
|
|
|
|
async def reject_search(*args, **kwargs):
|
|
raise AssertionError("authorization must not scan review search pages")
|
|
|
|
monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
|
|
monkeypatch.setattr(gitea_proxy, "work_page", reject_search)
|
|
|
|
assert await gitea_proxy.is_requested_review("stackchain/api", 77) is True
|
|
assert sorted(requested) == ["repos/stackchain/api/pulls/77", "user"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_assigned_issue_guard_uses_direct_target_and_current_user(monkeypatch):
|
|
requested = []
|
|
|
|
async def fake_fetch(path):
|
|
requested.append(path)
|
|
if path == "user":
|
|
return {"login": "timmy"}
|
|
assert path == "repos/stackchain/api/issues/77"
|
|
return {
|
|
"state": "open",
|
|
"assignees": [{"login": "timmy"}],
|
|
}
|
|
|
|
async def reject_search(*args, **kwargs):
|
|
raise AssertionError("authorization must not scan work search pages")
|
|
|
|
monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
|
|
monkeypatch.setattr(gitea_proxy, "work_page", reject_search)
|
|
|
|
assert await gitea_proxy.is_assigned_issue("stackchain/api", 77) is True
|
|
assert sorted(requested) == ["repos/stackchain/api/issues/77", "user"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_assigned_pull_guard_uses_direct_target_and_current_user(monkeypatch):
|
|
requested = []
|
|
|
|
async def fake_fetch(path):
|
|
requested.append(path)
|
|
if path == "user":
|
|
return {"login": "timmy"}
|
|
assert path == "repos/stackchain/api/pulls/77"
|
|
return {
|
|
"state": "open",
|
|
"assignees": [{"login": "timmy"}],
|
|
}
|
|
|
|
async def reject_search(*args, **kwargs):
|
|
raise AssertionError("authorization must not scan work search pages")
|
|
|
|
monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
|
|
monkeypatch.setattr(gitea_proxy, "work_page", reject_search)
|
|
|
|
assert await gitea_proxy.is_assigned_pull("stackchain/api", 77) is True
|
|
assert sorted(requested) == ["repos/stackchain/api/pulls/77", "user"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_context_preserves_repository_and_update_time_for_cross_repo_work(monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
async def repositories():
|
|
return []
|
|
|
|
async def issues():
|
|
return [{
|
|
"id": 10,
|
|
"number": 7,
|
|
"title": "Ship mobile flow",
|
|
"state": "open",
|
|
"labels": [{"name": "P0"}],
|
|
"assignees": [{"login": "timmy"}],
|
|
"repository": {"full_name": "stackchain/mobile"},
|
|
"updated_at": "2026-08-06T12:00:00Z",
|
|
"due_date": "2026-08-09T23:59:59Z",
|
|
"html_url": "https://forge.example/stackchain/mobile/issues/7",
|
|
}]
|
|
|
|
async def pulls():
|
|
return [{
|
|
"id": 11,
|
|
"number": 7,
|
|
"title": "Review API",
|
|
"state": "open",
|
|
"user": {"login": "alex"},
|
|
"labels": [{"name": "priority-high"}],
|
|
"assignees": [{"login": "timmy"}],
|
|
"work_reasons": ["assigned_to_me", "review_requested"],
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"updated_at": "2026-08-06T11:00:00Z",
|
|
"html_url": "https://forge.example/stackchain/api/pulls/7",
|
|
}]
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", repositories)
|
|
monkeypatch.setattr(main, "issues", issues)
|
|
monkeypatch.setattr(main, "pull_requests", pulls)
|
|
|
|
payload = json.loads((await main.context()).body)
|
|
|
|
assert payload["issues"][0]["repository"] == "stackchain/mobile"
|
|
assert payload["issues"][0]["updated_at"] == "2026-08-06T12:00:00Z"
|
|
assert payload["issues"][0]["due_date"] == "2026-08-09T23:59:59Z"
|
|
assert payload["pull_requests"][0]["repository"] == "stackchain/api"
|
|
assert payload["pull_requests"][0]["updated_at"] == "2026-08-06T11:00:00Z"
|
|
assert payload["pull_requests"][0]["labels"] == ["priority-high"]
|
|
assert payload["pull_requests"][0]["assignees"] == ["timmy"]
|
|
assert payload["pull_requests"][0]["work_reasons"] == [
|
|
"assigned_to_me",
|
|
"review_requested",
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pull_review_detail_combines_pr_files_status_and_reviews(monkeypatch):
|
|
requested_paths = []
|
|
|
|
async def fake_fetch(path):
|
|
requested_paths.append(path)
|
|
if path.endswith("/pulls/7"):
|
|
return {
|
|
"title": "Review API",
|
|
"body": "Please check the retry flow.",
|
|
"html_url": "https://forge.example/stackchain/api/pulls/7",
|
|
"user": {"login": "alex"},
|
|
"head": {"sha": "abc123"},
|
|
}
|
|
if path.endswith("/files"):
|
|
return [
|
|
{"filename": "src/api.py", "status": "modified", "additions": 8, "deletions": 2},
|
|
None,
|
|
]
|
|
if "/commits/abc123/status" in path:
|
|
return {"state": "success"}
|
|
return [
|
|
{"user": {"login": "sam"}, "state": "APPROVED", "body": "Looks good"},
|
|
"malformed",
|
|
]
|
|
|
|
monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
|
|
|
|
async def fake_fetch_text(path, max_bytes):
|
|
assert path == "repos/stackchain/api/pulls/7.diff"
|
|
assert max_bytes == gitea_proxy.REVIEW_DIFF_MAX_BYTES
|
|
return "", False
|
|
|
|
monkeypatch.setattr(gitea_proxy, "fetch_text", fake_fetch_text)
|
|
|
|
detail = await gitea_proxy.pull_review_detail("stackchain/api", 7)
|
|
|
|
assert requested_paths == [
|
|
"repos/stackchain/api/pulls/7",
|
|
"repos/stackchain/api/pulls/7/files",
|
|
"repos/stackchain/api/commits/abc123/status",
|
|
"repos/stackchain/api/pulls/7/reviews",
|
|
]
|
|
assert detail["author"] == "alex"
|
|
assert detail["head_sha"] == "abc123"
|
|
assert detail["ci_state"] == "success"
|
|
assert detail["files"][0]["filename"] == "src/api.py"
|
|
assert detail["reviews"][0]["state"] == "APPROVED"
|
|
assert len(detail["files"]) == 1
|
|
assert len(detail["reviews"]) == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pull_review_detail_fetches_independent_head_resources_concurrently(monkeypatch):
|
|
started = {name: asyncio.Event() for name in ("files", "status", "reviews", "diff")}
|
|
release = asyncio.Event()
|
|
|
|
async def wait_for_release(name, result):
|
|
started[name].set()
|
|
await release.wait()
|
|
return result
|
|
|
|
async def fake_fetch(path):
|
|
if path.endswith("/pulls/7"):
|
|
return {"head": {"sha": "abc123"}, "user": {"login": "alex"}}
|
|
if path.endswith("/files"):
|
|
return await wait_for_release("files", [])
|
|
if path.endswith("/reviews"):
|
|
return await wait_for_release("reviews", [])
|
|
return await wait_for_release("status", {"state": "success"})
|
|
|
|
async def fake_fetch_text(path, max_bytes):
|
|
return await wait_for_release("diff", ("", False))
|
|
|
|
monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
|
|
monkeypatch.setattr(gitea_proxy, "fetch_text", fake_fetch_text)
|
|
|
|
detail_task = asyncio.create_task(
|
|
gitea_proxy.pull_review_detail("stackchain/api", 7)
|
|
)
|
|
try:
|
|
await asyncio.wait_for(
|
|
asyncio.gather(*(event.wait() for event in started.values())), timeout=1
|
|
)
|
|
finally:
|
|
release.set()
|
|
|
|
detail = await detail_task
|
|
assert detail["head_sha"] == "abc123"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_pull_review_detail_attaches_bounded_per_file_diff_previews(monkeypatch):
|
|
async def fake_fetch(path):
|
|
if path.endswith("/pulls/7"):
|
|
return {
|
|
"title": "Review API",
|
|
"head": {"sha": "abc123"},
|
|
"user": {"login": "alex"},
|
|
}
|
|
if path.endswith("/files"):
|
|
return [
|
|
{"filename": "src/api.py", "status": "modified"},
|
|
{"filename": "assets/logo.png", "status": "modified"},
|
|
]
|
|
if path.endswith("/reviews"):
|
|
return []
|
|
return {"state": "success"}
|
|
|
|
diff = """diff --git a/src/api.py b/src/api.py
|
|
index 123..456 100644
|
|
--- a/src/api.py
|
|
+++ b/src/api.py
|
|
@@ -1,2 +1,3 @@
|
|
context
|
|
-old <value>
|
|
+new <value>
|
|
diff --git a/assets/logo.png b/assets/logo.png
|
|
Binary files a/assets/logo.png and b/assets/logo.png differ
|
|
"""
|
|
|
|
async def fake_fetch_text(path, max_bytes):
|
|
assert path == "repos/stackchain/api/pulls/7.diff"
|
|
return diff, True
|
|
|
|
monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
|
|
monkeypatch.setattr(gitea_proxy, "fetch_text", fake_fetch_text)
|
|
|
|
detail = await gitea_proxy.pull_review_detail("stackchain/api", 7)
|
|
|
|
source, binary = detail["files"]
|
|
assert source["diff_lines"] == [
|
|
"@@ -1,2 +1,3 @@",
|
|
" context",
|
|
"-old <value>",
|
|
"+new <value>",
|
|
]
|
|
assert source["diff_truncated"] is True
|
|
assert source["diff_available"] is True
|
|
assert binary["diff_available"] is False
|
|
assert binary["diff_binary"] is True
|