412 lines
14 KiB
Python
412 lines
14 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
import httpx
|
|
|
|
from src import gitea_proxy
|
|
from src import main
|
|
|
|
|
|
@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_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",
|
|
"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["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
|