stackchain-dashboard/tests/test_global_search.py

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):
        requested.append((query, limit))
        return [{
            "kind": "issue",
            "repository": "stackchain/api",
            "number": 42,
            "title": "Repair mobile queue",
            "state": "open",
            "url": "https://forge.example/stackchain/api/issues/42",
        }]

    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")

    assert response.status_code == 200
    assert response.headers["cache-control"] == "no-store"
    assert requested == [("mobile", 7)]
    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",
    }]}


@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"] == "7" for request in requests)
    assert results == [{
        "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",
    }]


@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,
        "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