tests/test_app.py

"""app.py: HTTP wiring only.

The orchestration these endpoints used to contain now lives in
services.py and is tested in test_search_service.py and
test_patent_fallback.py without an HTTP layer. What's left here is what
app.py is actually responsible for: request validation, dependency wiring,
the lifespan, and turning domain results into responses.
"""

import httpx
import pytest
from httpx import ASGITransport

import app as app_module
from scrap import PatentScrapBulkResponse, PatentScrapResult
from serp import SerpResults


@pytest.fixture
async def client():
    transport = ASGITransport(app=app_module.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


@pytest.fixture
def override_services():
    """Install stand-in services for the duration of one test."""
    def _install(*, search=None, patent=None):
        if search is not None:
            app_module.app.dependency_overrides[app_module.get_search_service] = lambda: search
        if patent is not None:
            app_module.app.dependency_overrides[app_module.get_patent_service] = lambda: patent

    yield _install
    app_module.app.dependency_overrides.clear()


class _RecordingSearchService:
    """Records which service method each endpoint dispatched to."""

    def __init__(self):
        self.calls = []

    def _record(self, name):
        async def _fn(params):
            self.calls.append((name, list(params.queries), params.n_results))
            return SerpResults(results=[{"title": name}], error=None)
        return _fn

    def __getattr__(self, name):
        return self._record(name)


class _StubPatentService:
    def __init__(self):
        self.calls = []

    async def scrap(self, patent_id):
        self.calls.append(("scrap", patent_id))
        return PatentScrapResult(title="Widget apparatus")

    async def scrap_bulk(self, patent_ids):
        self.calls.append(("scrap_bulk", patent_ids))
        return PatentScrapBulkResponse(patents=[], failed_ids=list(patent_ids))

    async def ops_scrap(self, patent_id):
        self.calls.append(("ops_scrap", patent_id))
        return PatentScrapResult(title="From OPS")

    async def ops_scrap_bulk(self, patent_ids):
        self.calls.append(("ops_scrap_bulk", patent_ids))
        return PatentScrapBulkResponse(patents=[], failed_ids=list(patent_ids))


# ------------------------------- endpoint dispatch -------------------------------


@pytest.mark.parametrize("path, expected_method", [
    ("/serp/search_scholar", "google_scholar"),
    ("/serp/search_arxiv", "arxiv"),
    ("/serp/search_patents", "patents"),
    ("/serp/search_brave", "brave"),
    ("/serp/search_bing", "bing"),
    ("/serp/search_duck", "duckduckgo"),
    ("/serp/search", "search"),
    ("/ops/search", "ops_keyword_search"),
])
async def test_each_search_endpoint_dispatches_to_its_service_method(
        client, override_services, path, expected_method):
    """Guards the wiring itself: every endpoint reaching the right method
    with the request's queries and result count intact."""
    service = _RecordingSearchService()
    override_services(search=service)

    resp = await client.post(path, json={"queries": ["a", "b"], "n_results": 25})

    assert resp.status_code == 200
    assert service.calls == [(expected_method, ["a", "b"], 25)]
    assert resp.json()["results"] == [{"title": expected_method}]


@pytest.mark.parametrize("method, path, expected_call", [
    ("get", "/scrap/scrap_patent/US11930446B2", ("scrap", "US11930446B2")),
    ("get", "/ops/scrap_patent/US11930446B2", ("ops_scrap", "US11930446B2")),
])
async def test_patent_endpoints_dispatch_to_their_service_method(
        client, override_services, method, path, expected_call):
    service = _StubPatentService()
    override_services(patent=service)

    resp = await getattr(client, method)(path)

    assert resp.status_code == 200
    assert service.calls == [expected_call]


@pytest.mark.parametrize("path, expected_method", [
    ("/scrap/scrap_patents_bulk", "scrap_bulk"),
    ("/ops/scrap_patents_bulk", "ops_scrap_bulk"),
])
async def test_bulk_endpoints_dispatch_to_their_service_method(
        client, override_services, path, expected_method):
    service = _StubPatentService()
    override_services(patent=service)

    resp = await client.post(path, json={"patent_ids": ["US11930446B2"]})

    assert resp.status_code == 200
    assert service.calls == [(expected_method, ["US11930446B2"])]


# ------------------------------- patent_id validation -------------------------------


@pytest.mark.parametrize("path", [
    "/scrap/scrap_patent/not-a-patent-id",
    "/ops/scrap_patent/not-a-patent-id",
])
async def test_malformed_patent_id_is_rejected(client, path):
    resp = await client.get(path)
    assert resp.status_code == 422


@pytest.mark.parametrize("path", [
    "/scrap/scrap_patents_bulk", "/ops/scrap_patents_bulk"])
async def test_malformed_patent_id_in_a_bulk_list_is_rejected(client, path):
    resp = await client.post(path, json={"patent_ids": ["US11930446B2", "garbage!!"]})
    assert resp.status_code == 422


async def test_well_formed_patent_id_is_accepted(client, override_services):
    override_services(patent=_StubPatentService())

    resp = await client.get("/scrap/scrap_patent/US11930446B2")

    assert resp.status_code == 200
    assert resp.json()["title"] == "Widget apparatus"


# ---------------------------- default dependency wiring ----------------------------


def test_the_default_search_service_gets_the_apps_collaborators():
    """The providers exist so tests can override them; this pins that the
    un-overridden ones still hand the service the real collaborators."""
    service = app_module.get_search_service()

    assert service._http is app_module.httpx_client
    assert service._breaker is app_module._backend_circuit_breaker
    assert service._ops_tokens is app_module.ops_token_manager


def test_the_search_service_reads_the_browser_lazily(monkeypatch):
    """The browser is started by the lifespan, after the service may already
    exist, and stays None when Playwright fails to start - so the service
    has to read it per call rather than capture it at construction."""
    service = app_module.get_search_service()
    monkeypatch.setattr(app_module, "pw_browser", "a-browser")

    assert service.browser == "a-browser"


def test_the_default_patent_service_gets_the_apps_collaborators():
    service = app_module.get_patent_service()

    assert service._http is app_module.httpx_client
    assert service._ops_tokens is app_module.ops_token_manager


# ------------------------------------- api_lifespan -------------------------------------


class _RecordingClient:
    def __init__(self):
        self.closed = False

    async def aclose(self):
        self.closed = True


async def test_api_lifespan_launches_chromium_without_a_sandbox(monkeypatch):
    """The container runs as a non-root user (see the Dockerfile), and
    Chromium's own internal sandbox needs privileges a non-root container
    user doesn't have - so it must be launched with --no-sandbox, or it
    fails to start at all in that environment.
    """
    launch_calls = []

    class FakeBrowser:
        async def close(self):
            pass

    class FakeChromium:
        async def launch(self, **kwargs):
            launch_calls.append(kwargs)
            return FakeBrowser()

    class FakePlaywright:
        chromium = FakeChromium()

        async def stop(self):
            pass

    class FakePlaywrightContextManager:
        async def start(self):
            return FakePlaywright()

    # Reset via monkeypatch (not a plain assignment) so its automatic
    # teardown restores whatever these module globals held before this test,
    # regardless of what api_lifespan reassigns them to while it runs.
    monkeypatch.setattr(app_module, "pw_browser", None)
    monkeypatch.setattr(app_module, "playwright", None)
    monkeypatch.setattr(app_module, "async_playwright", lambda: FakePlaywrightContextManager())
    # The lifespan also closes the shared httpx client on shutdown; swap in
    # a stand-in so this test doesn't close the real one for the rest of
    # the session.
    monkeypatch.setattr(app_module, "httpx_client", _RecordingClient())

    async with app_module.api_lifespan(app_module.app):
        pass

    assert launch_calls[0]["headless"] is True
    assert "--no-sandbox" in launch_calls[0].get("args", [])


async def test_api_lifespan_closes_the_http_client_on_shutdown(monkeypatch):
    """The client was created at import and never closed - the browser was
    torn down on shutdown but its connection pool was not.
    """
    recording_client = _RecordingClient()
    monkeypatch.setattr(app_module, "pw_browser", None)
    monkeypatch.setattr(app_module, "playwright", None)
    monkeypatch.setattr(app_module, "httpx_client", recording_client)
    # Playwright failing to start is the documented degraded mode; the
    # client must still be closed cleanly on the way out.
    monkeypatch.setattr(
        app_module, "async_playwright", lambda: (_ for _ in ()).throw(RuntimeError("no browser")))

    async with app_module.api_lifespan(app_module.app):
        pass

    assert recording_client.closed is True


async def test_api_lifespan_survives_playwright_failing_to_start(monkeypatch):
    """Documented behaviour: the app still comes up, pw_browser stays None,
    and the browser-backed endpoints fail individually rather than the whole
    deployment refusing to boot.
    """
    monkeypatch.setattr(app_module, "pw_browser", None)
    monkeypatch.setattr(app_module, "playwright", None)
    monkeypatch.setattr(app_module, "httpx_client", _RecordingClient())
    monkeypatch.setattr(
        app_module, "async_playwright", lambda: (_ for _ in ()).throw(RuntimeError("no browser")))

    async with app_module.api_lifespan(app_module.app):
        assert app_module.pw_browser is None