895 lines
33 KiB
Python
895 lines
33 KiB
Python
import asyncio
|
|
import io
|
|
|
|
import httpx
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
from src import gitea_proxy, main
|
|
|
|
|
|
def png_bytes(color="red"):
|
|
output = io.BytesIO()
|
|
Image.new("RGB", (2, 2), color).save(output, format="PNG")
|
|
return output.getvalue()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_endpoint_scopes_every_page_to_an_exact_repository(monkeypatch):
|
|
requested = []
|
|
|
|
async def search(query, limit, page, kind, state, repository):
|
|
requested.append((query, limit, page, kind, state, repository))
|
|
return {"items": [], "partial": False, "has_more": False, "next_page": 4}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "global_search", search, raising=False)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/search?q=mobile&limit=7&page=3&kind=pull&state=open&repository=stackchain%2Fapi"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert requested == [("mobile", 7, 3, "pull", "open", "stackchain/api")]
|
|
assert response.json()["scope"] == {
|
|
"kind": "pull", "state": "open", "repository": "stackchain/api"
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_endpoint_forwards_independent_stream_continuation(monkeypatch):
|
|
requested = []
|
|
|
|
async def search(query, limit, page, kind, state, repository, continuation):
|
|
requested.append((query, limit, page, kind, state, repository, continuation))
|
|
return {
|
|
"items": [], "partial": False, "has_more": False, "next_page": 3,
|
|
"continuation": {"issues": None, "pulls": None}, "failed_streams": [],
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "global_search", search, raising=False)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/search?q=mobile&limit=7&page=2&issues_page=1&pulls_page=3"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert requested == [
|
|
("mobile", 7, 2, "all", "all", None, {"issues": 1, "pulls": 3})
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_rejects_malformed_repository_scope_before_upstream(monkeypatch):
|
|
called = False
|
|
|
|
async def search(*args):
|
|
nonlocal called
|
|
called = True
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "global_search", search, raising=False)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/search?q=mobile&repository=stackchain%2Fapi%2Fsecrets"
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_repository_scope_reaches_issue_and_pull_streams():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(dict(request.url.params))
|
|
return httpx.Response(200, json=[])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
await gitea_proxy.global_search(
|
|
"mobile", limit=6, page=2, repository="stackchain/api"
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(requests) == 2
|
|
assert all(request["owner"] == "stackchain" and request["repo"] == "api" for request in requests)
|
|
assert all(request["page"] == "2" for request in requests)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_endpoint_forwards_valid_type_and_status_scope(monkeypatch):
|
|
requested = []
|
|
|
|
async def search(query, limit, page, kind, state):
|
|
requested.append((query, limit, page, kind, state))
|
|
return {"items": [], "partial": False, "has_more": False, "next_page": 2}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "global_search", search, raising=False)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/search?q=mobile&limit=7&page=1&kind=pull&state=open"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert requested == [("mobile", 7, 1, "pull", "open")]
|
|
assert response.json()["scope"] == {"kind": "pull", "state": "open"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_specific_type_uses_one_full_width_scoped_stream():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(dict(request.url.params))
|
|
return httpx.Response(200, json=[{
|
|
"number": 9,
|
|
"title": "Review mobile search",
|
|
"state": "open",
|
|
"repository": {"full_name": "stackchain/web"},
|
|
"html_url": "http://127.0.0.1:3000/stackchain/web/pulls/9",
|
|
}])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.global_search(
|
|
"mobile", limit=7, page=2, kind="pull", state="open"
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert requests == [{
|
|
"q": "mobile", "type": "pulls", "state": "open", "limit": "7", "page": "2"
|
|
}]
|
|
assert [item["kind"] for item in result["items"]] == ["pull"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_endpoint_returns_bounded_normalized_results(monkeypatch):
|
|
requested = []
|
|
|
|
async def search(query, limit, page, kind, state):
|
|
requested.append((query, limit, page, kind, state))
|
|
return {
|
|
"items": [{
|
|
"kind": "issue",
|
|
"repository": "stackchain/api",
|
|
"number": 42,
|
|
"title": "Repair mobile queue",
|
|
"state": "open",
|
|
"url": "http://127.0.0.1:3000/stackchain/api/issues/42",
|
|
}],
|
|
"partial": False,
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "global_search", search, raising=False)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/search?q=mobile&limit=7&page=2")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert requested == [("mobile", 7, 2, "all", "all")]
|
|
assert response.json() == {"query": "mobile", "scope": {"kind": "all", "state": "all"}, "items": [{
|
|
"kind": "issue",
|
|
"repository": "stackchain/api",
|
|
"number": 42,
|
|
"title": "Repair mobile queue",
|
|
"state": "open",
|
|
"url": "http://127.0.0.1:3000/stackchain/api/issues/42",
|
|
}], "partial": False}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_preview_returns_normalized_action_context(monkeypatch):
|
|
requested = []
|
|
|
|
async def preview(repository, kind, number):
|
|
requested.append((repository, kind, number))
|
|
return {
|
|
"kind": "issue",
|
|
"repository": "stackchain/api",
|
|
"number": 42,
|
|
"title": "Repair mobile queue",
|
|
"body": "Restore the queue before release.",
|
|
"state": "open",
|
|
"author": "alex",
|
|
"labels": ["P1"],
|
|
"assignees": [],
|
|
"url": "http://127.0.0.1:3000/stackchain/api/issues/42",
|
|
"claimable": True,
|
|
"assigned_to_me": False,
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "work_preview", preview, raising=False)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/repos/stackchain/api/issues/42/preview?kind=issue"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert requested == [("stackchain/api", "issue", 42)]
|
|
assert response.json()["claimable"] is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_preview_subscription_reads_server_truth_without_caching(monkeypatch):
|
|
calls = []
|
|
|
|
async def preview(repository, kind, number):
|
|
calls.append(("preview", repository, kind, number))
|
|
return {"repository": repository, "kind": kind, "number": number,
|
|
"state": "open", "claimable": True}
|
|
|
|
async def subscription(repository, number):
|
|
calls.append(("subscription", repository, number))
|
|
return {"watching": True}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "work_preview", preview)
|
|
monkeypatch.setattr(main.gitea_proxy, "issue_subscription", subscription, raising=False)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/repos/stackchain/api/issues/42/preview/subscription?kind=issue"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {"watching": True}
|
|
assert calls == [
|
|
("preview", "stackchain/api", "issue", 42),
|
|
("subscription", "stackchain/api", 42),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize(("method", "watching"), [("PUT", True), ("DELETE", False)])
|
|
async def test_search_preview_subscription_mutation_revalidates_target_and_confirms_truth(
|
|
monkeypatch, tmp_path, method, watching
|
|
):
|
|
calls = []
|
|
store = main.FollowingStore(tmp_path / "following.sqlite3", encryption_key=b"g" * 32)
|
|
|
|
async def preview(repository, kind, number):
|
|
calls.append(("preview", repository, kind, number))
|
|
return {"repository": repository, "kind": kind, "number": number,
|
|
"title": "Quiet issue", "state": "open", "claimable": True,
|
|
"updated_at": "2026-08-23T03:00:00Z", "url": "https://forge.example/issue/42"}
|
|
|
|
async def set_subscription(repository, number, desired):
|
|
calls.append(("set", repository, number, desired))
|
|
return {"watching": desired}
|
|
|
|
async def user():
|
|
return {"login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "_following_store", lambda: store)
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main.gitea_proxy, "work_preview", preview)
|
|
monkeypatch.setattr(main.gitea_proxy, "set_issue_subscription", set_subscription, raising=False)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.request(
|
|
method,
|
|
"/api/v1/repos/stackchain/api/issues/42/preview/subscription?kind=issue",
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
expected_count = 1 if watching else 0
|
|
assert response.json() == {
|
|
"watching": watching,
|
|
"following_revision": expected_count,
|
|
"following_count": expected_count,
|
|
}
|
|
assert calls == [
|
|
("preview", "stackchain/api", "issue", 42),
|
|
("set", "stackchain/api", 42, watching),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_issue_subscription_mutation_uses_current_login_and_requires_confirmation():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append((request.method, request.url.path))
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.url.path.endswith("/subscriptions/check"):
|
|
return httpx.Response(200, json={"subscribed": True, "ignored": False})
|
|
return httpx.Response(201, json={})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.set_issue_subscription("stackchain/api", 42, True)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result == {"watching": True}
|
|
assert requests == [
|
|
("GET", "/api/v1/user"),
|
|
("PUT", "/api/v1/repos/stackchain/api/issues/42/subscriptions/timmy"),
|
|
("GET", "/api/v1/repos/stackchain/api/issues/42/subscriptions/check"),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("kind", ["issue", "pull"])
|
|
async def test_global_search_preview_conversation_returns_a_bounded_authorized_page(
|
|
monkeypatch, kind
|
|
):
|
|
requested = []
|
|
|
|
async def conversation(repository, number, page, limit):
|
|
requested.append((repository, number, page, limit))
|
|
return {
|
|
"comments": [{
|
|
"id": 7,
|
|
"author": "alexander",
|
|
"body": "Current **decision**",
|
|
"created_at": "2026-08-14T12:00:00Z",
|
|
"url": "http://127.0.0.1:3000/stackchain/api/issues/42#issuecomment-7",
|
|
}],
|
|
"page": 3,
|
|
"older_page": 2,
|
|
"total": 41,
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
main.gitea_proxy, "issue_conversation_page", conversation, raising=False
|
|
)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
f"/api/v1/repos/stackchain/api/issues/42/preview/conversation?kind={kind}&limit=20"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json()["comments"][0]["body"] == "Current **decision**"
|
|
assert requested == [("stackchain/api", 42, None, 20)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_preview_reply_is_idempotent_for_an_authorized_pull(monkeypatch):
|
|
main._idempotency_ledger.clear()
|
|
previews = []
|
|
comments = []
|
|
|
|
async def preview(repository, kind, number):
|
|
previews.append((repository, kind, number))
|
|
return {
|
|
"repository": repository,
|
|
"kind": "pull",
|
|
"number": number,
|
|
"title": "Review mobile reply",
|
|
"state": "open",
|
|
}
|
|
|
|
async def comment(repository, number, body):
|
|
comments.append((repository, number, body))
|
|
return {
|
|
"id": 91,
|
|
"author": "timmy",
|
|
"body": body,
|
|
"created_at": "2026-08-14T12:30:00Z",
|
|
"url": "http://127.0.0.1:3000/stackchain/api/pulls/42#issuecomment-91",
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "work_preview", preview)
|
|
monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "search-reply-pull-42"}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/42/preview/comments?kind=pull",
|
|
json={"body": "Ready to merge."},
|
|
headers=headers,
|
|
)
|
|
replay = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/42/preview/comments?kind=pull",
|
|
json={"body": "Ready to merge."},
|
|
headers=headers,
|
|
)
|
|
|
|
assert first.status_code == replay.status_code == 201
|
|
assert replay.json() == first.json()
|
|
assert previews == [("stackchain/api", "pull", 42)]
|
|
assert comments == [("stackchain/api", 42, "Ready to merge.")]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_preview_reply_rejects_a_mismatched_target_before_commenting(monkeypatch):
|
|
main._idempotency_ledger.clear()
|
|
comments = []
|
|
|
|
async def preview(repository, kind, number):
|
|
return {
|
|
"repository": repository,
|
|
"kind": "issue",
|
|
"number": number,
|
|
"title": "Different target",
|
|
}
|
|
|
|
async def comment(*args):
|
|
comments.append(args)
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "work_preview", preview)
|
|
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/42/preview/comments?kind=pull",
|
|
json={"body": "Do not misroute this."},
|
|
headers={"Idempotency-Key": "search-reply-mismatch-42"},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
assert comments == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_preview_attachment_is_idempotent_for_exact_visible_target(monkeypatch):
|
|
main._idempotency_ledger.clear()
|
|
previews = []
|
|
uploads = []
|
|
|
|
async def preview(repository, kind, number):
|
|
previews.append((repository, kind, number))
|
|
return {"repository": repository, "kind": kind, "number": number, "commentable": True}
|
|
|
|
async def upload(repository, number, filename, content_type, content):
|
|
uploads.append((repository, number, filename, content_type, content))
|
|
return {"name": filename, "url": "http://127.0.0.1:3000/evidence/photo.png", "size": len(content)}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "work_preview", preview)
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_preview_attachment", upload)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
headers = {"Idempotency-Key": "search-photo-pull-42"}
|
|
image = png_bytes()
|
|
files = {"file": ("photo.png", image, "image/png")}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
first = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/42/preview/attachments?kind=pull",
|
|
files=files,
|
|
headers=headers,
|
|
)
|
|
replay = await client.post(
|
|
"/api/v1/repos/stackchain/api/issues/42/preview/attachments?kind=pull",
|
|
files=files,
|
|
headers=headers,
|
|
)
|
|
|
|
assert first.status_code == replay.status_code == 201
|
|
assert replay.json() == first.json()
|
|
assert first.json()["markdown"] == ""
|
|
assert previews == [("stackchain/api", "pull", 42)]
|
|
assert uploads == [("stackchain/api", 42, "photo.png", "image/png", image)]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_preview_attachment_rejects_kind_mismatch_before_upload(monkeypatch):
|
|
main._idempotency_ledger.clear()
|
|
uploads = []
|
|
|
|
async def preview(repository, kind, number):
|
|
return {"repository": repository, "kind": "issue", "number": number}
|
|
|
|
async def upload(*args):
|
|
uploads.append(args)
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "work_preview", preview)
|
|
monkeypatch.setattr(main.gitea_proxy, "upload_preview_attachment", upload)
|
|
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/42/preview/attachments?kind=pull",
|
|
files={"file": ("photo.png", png_bytes(), "image/png")},
|
|
headers={"Idempotency-Key": "search-photo-mismatch-42"},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
assert uploads == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_queries_issues_and_pulls_and_skips_unsafe_results():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
requests.append(dict(request.url.params))
|
|
kind = request.url.params["type"]
|
|
if kind == "issues":
|
|
return httpx.Response(200, json=[{
|
|
"id": 4, "number": 42, "title": "Repair queue", "state": "open",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"html_url": "http://127.0.0.1:3000/stackchain/api/issues/42",
|
|
}, {
|
|
"id": 4, "number": 42, "title": "Repair queue", "state": "open",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"html_url": "http://127.0.0.1:3000/stackchain/api/issues/42",
|
|
}, {
|
|
"id": 5, "number": 43, "title": "Unsafe", "state": "open",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"html_url": "javascript:alert(1)",
|
|
}])
|
|
return httpx.Response(200, json=[{
|
|
"id": 8, "number": 9, "title": "Improve search", "state": "closed",
|
|
"repository": {"full_name": "stackchain/web"},
|
|
"html_url": "http://127.0.0.1:3000/stackchain/web/pulls/9",
|
|
}])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
results = await gitea_proxy.global_search("mobile queue", 7)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert {request["type"] for request in requests} == {"issues", "pulls"}
|
|
assert {request["type"]: request["limit"] for request in requests} == {
|
|
"issues": "4", "pulls": "3"
|
|
}
|
|
assert all(request["q"] == "mobile queue" and request["state"] == "all"
|
|
for request in requests)
|
|
assert results == {
|
|
"items": [{
|
|
"kind": "issue", "repository": "stackchain/api", "number": 42,
|
|
"title": "Repair queue", "state": "open",
|
|
"url": "http://127.0.0.1:3000/stackchain/api/issues/42",
|
|
}, {
|
|
"kind": "pull", "repository": "stackchain/web", "number": 9,
|
|
"title": "Improve search", "state": "closed",
|
|
"url": "http://127.0.0.1:3000/stackchain/web/pulls/9",
|
|
}],
|
|
"partial": False,
|
|
"has_more": False,
|
|
"next_page": 2,
|
|
"continuation": {"issues": None, "pulls": None},
|
|
"failed_streams": [],
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_keeps_results_on_the_configured_forge_origin(monkeypatch):
|
|
monkeypatch.setattr(gitea_proxy, "GITEA_URL", "https://forge.example/git")
|
|
|
|
async def handler(request):
|
|
if request.url.params["type"] == "pulls":
|
|
return httpx.Response(200, json=[])
|
|
return httpx.Response(200, json=[{
|
|
"id": 1, "number": 41, "title": "Foreign", "state": "open",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"html_url": "https://attacker.example/stackchain/api/issues/41",
|
|
}, {
|
|
"id": 3, "number": 43, "title": "Downgraded", "state": "open",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"html_url": "http://forge.example/git/stackchain/api/issues/43",
|
|
}, {
|
|
"id": 2, "number": 42, "title": "Relative", "state": "open",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"html_url": "/git/stackchain/api/issues/42#issuecomment-7",
|
|
}])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.global_search("queue", 10)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert [(item["number"], item["url"]) for item in result["items"]] == [
|
|
(42, "https://forge.example/git/stackchain/api/issues/42#issuecomment-7")
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_returns_healthy_stream_when_other_stream_fails():
|
|
async def handler(request):
|
|
if request.url.params["type"] == "issues":
|
|
return httpx.Response(503, json={"message": "upstream details must stay private"})
|
|
return httpx.Response(200, json=[{
|
|
"number": 9,
|
|
"title": "Improve search",
|
|
"state": "open",
|
|
"repository": {"full_name": "stackchain/web"},
|
|
"html_url": "http://127.0.0.1:3000/stackchain/web/pulls/9",
|
|
}])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.global_search("search", 10)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result == {
|
|
"items": [{
|
|
"kind": "pull",
|
|
"repository": "stackchain/web",
|
|
"number": 9,
|
|
"title": "Improve search",
|
|
"state": "open",
|
|
"url": "http://127.0.0.1:3000/stackchain/web/pulls/9",
|
|
}],
|
|
"partial": True,
|
|
"has_more": True,
|
|
"next_page": 2,
|
|
"continuation": {"issues": 1, "pulls": None},
|
|
"failed_streams": ["issue"],
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_retries_a_failed_stream_without_advancing_its_page():
|
|
requests = []
|
|
issue_attempts = 0
|
|
|
|
async def handler(request):
|
|
nonlocal issue_attempts
|
|
params = dict(request.url.params)
|
|
requests.append((params["type"], params["page"], params["limit"]))
|
|
if params["type"] == "issues":
|
|
issue_attempts += 1
|
|
if issue_attempts == 1:
|
|
return httpx.Response(503)
|
|
number = int(params["page"])
|
|
return httpx.Response(200, json=[{
|
|
"number": number,
|
|
"title": f'{params["type"]} {number}',
|
|
"state": "open",
|
|
"repository": {"full_name": "stackchain/web"},
|
|
"html_url": f'http://127.0.0.1:3000/stackchain/web/{params["type"]}/{number}',
|
|
}])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
partial = await gitea_proxy.global_search("search", limit=4)
|
|
recovered = await gitea_proxy.global_search(
|
|
"search", limit=4, continuation=partial["continuation"]
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert partial["partial"] is True
|
|
assert partial["has_more"] is True
|
|
assert partial["continuation"] == {"issues": 1, "pulls": None}
|
|
assert recovered["partial"] is False
|
|
assert [item["kind"] for item in recovered["items"]] == ["issue"]
|
|
assert requests == [
|
|
("issues", "1", "2"), ("pulls", "1", "2"), ("issues", "1", "2")
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_odd_limit_keeps_both_streams_contiguous_across_pages():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
params = dict(request.url.params)
|
|
stream_limit = int(params["limit"])
|
|
page = int(params["page"])
|
|
requests.append((params["type"], page, stream_limit))
|
|
start = (page - 1) * stream_limit + 1
|
|
return httpx.Response(200, json=[{
|
|
"number": number,
|
|
"title": f'{params["type"]} {number}',
|
|
"state": "open",
|
|
"repository": {"full_name": "stackchain/web"},
|
|
"html_url": f'http://127.0.0.1:3000/stackchain/web/{params["type"]}/{number}',
|
|
} for number in range(start, start + stream_limit)])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
first = await gitea_proxy.global_search("search", limit=7)
|
|
second = await gitea_proxy.global_search(
|
|
"search", limit=7, continuation=first["continuation"]
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
pulls = [item["number"] for result in (first, second)
|
|
for item in result["items"] if item["kind"] == "pull"]
|
|
issues = [item["number"] for result in (first, second)
|
|
for item in result["items"] if item["kind"] == "issue"]
|
|
assert pulls == [1, 2, 3, 4, 5, 6]
|
|
assert issues == [1, 2, 3, 4, 5, 6, 7, 8]
|
|
assert requests == [
|
|
("issues", 1, 4), ("pulls", 1, 3),
|
|
("issues", 2, 4), ("pulls", 2, 3),
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_caps_combined_results_to_requested_limit():
|
|
async def handler(request):
|
|
kind = request.url.params["type"]
|
|
items = [{
|
|
"number": number,
|
|
"title": f"{kind} {number}",
|
|
"state": "open",
|
|
"repository": {"full_name": "stackchain/web"},
|
|
"html_url": f"http://127.0.0.1:3000/stackchain/web/{kind}/{number}",
|
|
} for number in (1, 2)]
|
|
return httpx.Response(200, json=items)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.global_search("search", 2)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(result["items"]) == 2
|
|
assert result["partial"] is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_pages_each_stream_and_balances_results():
|
|
requests = []
|
|
|
|
async def handler(request):
|
|
params = dict(request.url.params)
|
|
requests.append(params)
|
|
kind = params["type"]
|
|
page = int(params["page"])
|
|
items = [{
|
|
"number": page * 10 + number,
|
|
"title": f"{kind} page {page} result {number}",
|
|
"state": "open",
|
|
"repository": {"full_name": "stackchain/web"},
|
|
"html_url": f"http://127.0.0.1:3000/stackchain/web/{kind}/{page * 10 + number}",
|
|
} for number in (1, 2, 3, 4)]
|
|
return httpx.Response(200, json=items)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
result = await gitea_proxy.global_search("search", limit=4, page=2)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert {(request["type"], request["page"]) for request in requests} == {
|
|
("issues", "2"), ("pulls", "2")
|
|
}
|
|
assert [item["kind"] for item in result["items"]] == ["issue", "pull", "issue", "pull"]
|
|
assert result["next_page"] == 3
|
|
assert result["has_more"] is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_propagates_cancellation_to_stop_obsolete_work():
|
|
async def handler(request):
|
|
if request.url.params["type"] == "issues":
|
|
raise asyncio.CancelledError()
|
|
return httpx.Response(200, json=[])
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await gitea_proxy.global_search("obsolete", 10)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_work_preview_normalizes_details_and_only_allows_unassigned_open_issues():
|
|
async def handler(request):
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"number": 42,
|
|
"title": "Repair queue",
|
|
"body": "Keep mobile operators moving.",
|
|
"state": "open",
|
|
"html_url": "http://127.0.0.1:3000/stackchain/api/issues/42",
|
|
"user": {"login": "alex"},
|
|
"labels": [{"name": "P1"}, None],
|
|
"assignees": [],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
preview = await gitea_proxy.work_preview("stackchain/api", "issue", 42)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert preview == {
|
|
"kind": "issue",
|
|
"repository": "stackchain/api",
|
|
"number": 42,
|
|
"title": "Repair queue",
|
|
"body": "Keep mobile operators moving.",
|
|
"state": "open",
|
|
"author": "alex",
|
|
"labels": ["P1"],
|
|
"assignees": [],
|
|
"url": "http://127.0.0.1:3000/stackchain/api/issues/42",
|
|
"claimable": True,
|
|
"reopenable": False,
|
|
"assigned_to_me": False,
|
|
"reviewable": False,
|
|
"commentable": True,
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_work_preview_derives_pull_kind_and_never_offers_issue_claim():
|
|
async def handler(request):
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"number": 9,
|
|
"title": "Improve search",
|
|
"state": "open",
|
|
"html_url": "http://127.0.0.1:3000/stackchain/web/pulls/9",
|
|
"pull_request": {"merged": False},
|
|
"assignees": [],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
preview = await gitea_proxy.work_preview("stackchain/web", "issue", 9)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert preview["kind"] == "pull"
|
|
assert preview["claimable"] is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_work_preview_marks_authoritative_requested_pull_review_actionable():
|
|
requested_paths = []
|
|
|
|
async def handler(request):
|
|
requested_paths.append(request.url.path)
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
if request.url.path.endswith("/pulls/9"):
|
|
return httpx.Response(200, json={
|
|
"state": "open",
|
|
"requested_reviewers": [{"login": "timmy"}, None],
|
|
})
|
|
return httpx.Response(200, json={
|
|
"number": 9,
|
|
"title": "Improve search",
|
|
"state": "open",
|
|
"html_url": "http://127.0.0.1:3000/stackchain/web/pulls/9",
|
|
"pull_request": {"merged": False},
|
|
"assignees": [],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
preview = await gitea_proxy.work_preview("stackchain/web", "issue", 9)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert preview["reviewable"] is True
|
|
assert "/api/v1/repos/stackchain/web/pulls/9" in requested_paths
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_work_preview_offers_reopen_only_for_closed_issues():
|
|
async def handler(request):
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"number": 42,
|
|
"title": "Resume work",
|
|
"state": "closed",
|
|
"html_url": "http://127.0.0.1:3000/stackchain/api/issues/42",
|
|
"assignees": [],
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(handler))
|
|
try:
|
|
preview = await gitea_proxy.work_preview("stackchain/api", "issue", 42)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert preview["reopenable"] is True
|
|
assert preview["claimable"] is False
|