280 lines
9.3 KiB
Python
280 lines
9.3 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from src import gitea_proxy, main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_work_route_endpoint_resolves_one_authorized_item(monkeypatch):
|
|
requested = []
|
|
|
|
async def resolve(kind, repository, number, notification_id):
|
|
requested.append((kind, repository, number, notification_id))
|
|
return {
|
|
"kind": "review",
|
|
"repository": "stackchain/dashboard",
|
|
"number": 87,
|
|
"title": "Review the mobile resolver",
|
|
"is_review": True,
|
|
"url": "https://forge.example/stackchain/dashboard/pulls/87",
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "resolve_work_route", resolve, 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/work-route?kind=review&repository=stackchain/dashboard&number=87"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert requested == [("review", "stackchain/dashboard", 87, None)]
|
|
assert response.json() == {
|
|
"kind": "review",
|
|
"repository": "stackchain/dashboard",
|
|
"number": 87,
|
|
"title": "Review the mobile resolver",
|
|
"is_review": True,
|
|
"url": "https://forge.example/stackchain/dashboard/pulls/87",
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_work_route_endpoint_accepts_filed_issue_identity(monkeypatch):
|
|
requested = []
|
|
|
|
async def resolve(kind, repository, number, notification_id):
|
|
requested.append((kind, repository, number, notification_id))
|
|
return {
|
|
"kind": "filed",
|
|
"repository": repository,
|
|
"number": number,
|
|
"title": "Delegated filing",
|
|
"is_filed": True,
|
|
"is_assigned": False,
|
|
}
|
|
|
|
monkeypatch.setattr(main.gitea_proxy, "resolve_work_route", resolve)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/work-route?kind=filed&repository=stackchain/dashboard&number=88"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert requested == [("filed", "stackchain/dashboard", 88, None)]
|
|
assert response.json()["kind"] == "filed"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_resolve_work_route_returns_only_a_requested_review():
|
|
requests = []
|
|
|
|
async def upstream(request):
|
|
requests.append(request.url.path)
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"id": 700,
|
|
"number": 87,
|
|
"title": "Review the resolver",
|
|
"state": "open",
|
|
"html_url": gitea_proxy.GITEA_URL + "/stackchain/dashboard/pulls/87",
|
|
"repository": {"full_name": "stackchain/dashboard"},
|
|
"requested_reviewers": [{"login": "timmy"}],
|
|
"assignees": [],
|
|
"user": {"login": "alex"},
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.resolve_work_route(
|
|
"review", "stackchain/dashboard", 87, None
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(requests) == 2
|
|
assert result == {
|
|
"kind": "review",
|
|
"repository": "stackchain/dashboard",
|
|
"number": 87,
|
|
"title": "Review the resolver",
|
|
"state": "open",
|
|
"url": gitea_proxy.GITEA_URL + "/stackchain/dashboard/pulls/87",
|
|
"is_review": True,
|
|
"work_reasons": ["review_requested"],
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_resolve_work_route_returns_an_open_authored_unassigned_pull():
|
|
async def upstream(request):
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"id": 703,
|
|
"number": 90,
|
|
"title": "Ship authored flow",
|
|
"state": "open",
|
|
"html_url": gitea_proxy.GITEA_URL + "/stackchain/dashboard/pulls/90",
|
|
"assignees": [],
|
|
"requested_reviewers": [],
|
|
"user": {"login": "timmy"},
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.resolve_work_route(
|
|
"pull", "stackchain/dashboard", 90, None
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result["kind"] == "pull"
|
|
assert result["number"] == 90
|
|
assert result["is_assigned"] is False
|
|
assert result["work_reasons"] == ["authored_by_me"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_resolve_work_route_returns_an_open_issue_filed_by_the_current_user():
|
|
requests = []
|
|
|
|
async def upstream(request):
|
|
requests.append(request.url.path)
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"id": 701,
|
|
"number": 88,
|
|
"title": "Delegated filing",
|
|
"state": "open",
|
|
"html_url": gitea_proxy.GITEA_URL + "/stackchain/dashboard/issues/88",
|
|
"assignees": [{"login": "alex"}],
|
|
"user": {"login": "timmy"},
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.resolve_work_route(
|
|
"filed", "stackchain/dashboard", 88, None
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert len(requests) == 2
|
|
assert result == {
|
|
"kind": "filed",
|
|
"repository": "stackchain/dashboard",
|
|
"number": 88,
|
|
"title": "Delegated filing",
|
|
"state": "open",
|
|
"url": gitea_proxy.GITEA_URL + "/stackchain/dashboard/issues/88",
|
|
"is_filed": True,
|
|
"is_assigned": False,
|
|
"work_reasons": ["created_by_me"],
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_resolve_work_route_returns_a_completed_issue_filed_by_the_current_user():
|
|
async def upstream(request):
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"id": 702,
|
|
"number": 89,
|
|
"title": "Completed delegation",
|
|
"state": "closed",
|
|
"html_url": gitea_proxy.GITEA_URL + "/stackchain/dashboard/issues/89",
|
|
"assignees": [{"login": "alex"}],
|
|
"user": {"login": "timmy"},
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.resolve_work_route(
|
|
"filed", "stackchain/dashboard", 89, None
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result == {
|
|
"kind": "filed",
|
|
"repository": "stackchain/dashboard",
|
|
"number": 89,
|
|
"title": "Completed delegation",
|
|
"state": "closed",
|
|
"url": gitea_proxy.GITEA_URL + "/stackchain/dashboard/issues/89",
|
|
"is_filed": True,
|
|
"is_assigned": False,
|
|
"work_reasons": ["created_by_me"],
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_completed_filed_issue_remains_authorized_for_detail_and_reply():
|
|
async def upstream(request):
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"state": "closed",
|
|
"user": {"login": "timmy"},
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
authorized = await gitea_proxy.is_authored_issue("stackchain/dashboard", 89)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert authorized is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize(
|
|
("state", "author", "authorized"),
|
|
[("open", "timmy", True), ("closed", "timmy", False), ("open", "alex", False)],
|
|
)
|
|
async def test_only_open_authored_issues_are_authorized_for_withdrawal(
|
|
state, author, authorized
|
|
):
|
|
async def upstream(request):
|
|
if request.url.path.endswith("/user"):
|
|
return httpx.Response(200, json={"login": "timmy"})
|
|
return httpx.Response(200, json={
|
|
"state": state,
|
|
"user": {"login": author},
|
|
})
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
result = await gitea_proxy.is_open_authored_issue(
|
|
"stackchain/dashboard", 89
|
|
)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|
|
|
|
assert result is authorized
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_resolve_work_route_rejects_a_notification_that_is_already_read():
|
|
async def upstream(request):
|
|
if request.url.path.endswith("/notifications/threads/913"):
|
|
return httpx.Response(200, json={
|
|
"id": 913,
|
|
"unread": False,
|
|
"repository": {"full_name": "stackchain/dashboard"},
|
|
"subject": {"title": "Already handled", "type": "Issue"},
|
|
})
|
|
return httpx.Response(404)
|
|
|
|
gitea_proxy.start_client(transport=httpx.MockTransport(upstream))
|
|
try:
|
|
with pytest.raises(gitea_proxy.WorkRouteUnavailableError):
|
|
await gitea_proxy.resolve_work_route("update", None, None, 913)
|
|
finally:
|
|
await gitea_proxy.stop_client()
|