503 lines
18 KiB
Python
503 lines
18 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy, main
|
|
|
|
|
|
@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_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": "https://forge.example/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": "https://forge.example/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": "https://forge.example/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": "https://forge.example/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
|
|
@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": "https://forge.example/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_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": "https://forge.example/stackchain/api/issues/42",
|
|
}, {
|
|
"id": 4, "number": 42, "title": "Repair queue", "state": "open",
|
|
"repository": {"full_name": "stackchain/api"},
|
|
"html_url": "https://forge.example/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": "https://forge.example/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 all(
|
|
request["q"] == "mobile queue"
|
|
and request["limit"] == "4"
|
|
and request["state"] == "all"
|
|
for request in requests
|
|
)
|
|
assert results == {
|
|
"items": [{
|
|
"kind": "issue", "repository": "stackchain/api", "number": 42,
|
|
"title": "Repair queue", "state": "open",
|
|
"url": "https://forge.example/stackchain/api/issues/42",
|
|
}, {
|
|
"kind": "pull", "repository": "stackchain/web", "number": 9,
|
|
"title": "Improve search", "state": "closed",
|
|
"url": "https://forge.example/stackchain/web/pulls/9",
|
|
}],
|
|
"partial": False,
|
|
"has_more": False,
|
|
"next_page": 2,
|
|
}
|
|
|
|
|
|
@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": "https://forge.example/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": "https://forge.example/stackchain/web/pulls/9",
|
|
}],
|
|
"partial": True,
|
|
"has_more": False,
|
|
"next_page": 2,
|
|
}
|
|
|
|
|
|
@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"https://forge.example/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"https://forge.example/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": "https://forge.example/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": "https://forge.example/stackchain/api/issues/42",
|
|
"claimable": True,
|
|
"reopenable": False,
|
|
"assigned_to_me": False,
|
|
"reviewable": False,
|
|
}
|
|
|
|
|
|
@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": "https://forge.example/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": "https://forge.example/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": "https://forge.example/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
|