318 lines
11 KiB
Python
318 lines
11 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy, main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_global_search_endpoint_returns_bounded_normalized_results(monkeypatch):
|
|
requested = []
|
|
|
|
async def search(query, limit, page):
|
|
requested.append((query, limit, page))
|
|
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)]
|
|
assert response.json() == {"query": "mobile", "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
|
|
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" 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,
|
|
}
|
|
|
|
|
|
@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_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
|