tests/test_patent_fallback.py

"""The Google Patents -> EPO OPS fallback chain.

The policy lives in services.PatentService, so most of this is plain unit
testing with no ASGI client involved. Only the last section goes through
the app, and only to check that the domain errors the service raises map
onto the right status codes.

The distinction these tests exist to protect: a 404 means "this patent does
not exist", and an agent is explicitly told by MCP_INSTRUCTIONS to believe
it and move on without retrying. Anything else - an upstream outage, a
timeout, a blocked scrape - must not be reported that way.
"""

import httpx
import pytest
from httpx import ASGITransport, HTTPStatusError, Request, Response

import app as app_module
import services as services_module
from scrap import PatentScrapResult
from services import (OPSUnconfigured, PatentNotFound, PatentService,
                      UpstreamUnavailable)

PATENT_ID = "US11930446B2"


class FakeOPSTokens:
    """Stand-in for the OPS token manager. Tests only care whether
    credentials are configured, and this avoids assigning to private
    attributes on the real module-level singleton.
    """

    def __init__(self, configured: bool):
        self.configured = configured


def make_service(*, ops_configured: bool) -> PatentService:
    return PatentService(http_client=None, ops_tokens=FakeOPSTokens(ops_configured))


def _http_error(status: int) -> HTTPStatusError:
    request = Request("GET", "https://patents.google.com/")
    return HTTPStatusError(
        str(status), request=request, response=Response(status, request=request))


def _raises(exc):
    async def _fn(*args, **kwargs):
        raise exc
    return _fn


def _returns(value):
    async def _fn(*args, **kwargs):
        return value
    return _fn


# ------------------------------- the happy path -------------------------------


async def test_successful_scrape_never_touches_ops(monkeypatch):
    ops_called = False

    async def fake_ops(*args, **kwargs):
        nonlocal ops_called
        ops_called = True
        return PatentScrapResult(title="from OPS")

    monkeypatch.setattr(
        services_module, "scrap_patent_async",
        _returns(PatentScrapResult(title="from Google Patents")))
    monkeypatch.setattr(services_module, "ops_scrap_patent", fake_ops)

    result = await make_service(ops_configured=True).scrap(PATENT_ID)

    assert result.title == "from Google Patents"
    assert ops_called is False


# ------------------------- genuine miss vs upstream failure -------------------------


async def test_a_real_404_from_google_patents_is_not_found(monkeypatch):
    """The one case where "not found" is the truth."""
    monkeypatch.setattr(services_module, "scrap_patent_async", _raises(_http_error(404)))

    with pytest.raises(PatentNotFound):
        await make_service(ops_configured=False).scrap(PATENT_ID)


@pytest.mark.parametrize("status", [429, 500, 502, 503])
async def test_an_upstream_failure_is_not_reported_as_not_found(monkeypatch, status):
    """MCP_INSTRUCTIONS tells the model a "not found" patent is genuinely
    absent everywhere and to move on rather than retrying. Reporting a
    transient upstream failure that way teaches an agent - with the
    server's explicit encouragement - that a real patent does not exist.
    """
    monkeypatch.setattr(services_module, "scrap_patent_async", _raises(_http_error(status)))

    with pytest.raises(UpstreamUnavailable) as exc_info:
        await make_service(ops_configured=False).scrap(PATENT_ID)

    assert exc_info.value.timeout is False


async def test_a_timeout_is_flagged_as_a_timeout(monkeypatch):
    monkeypatch.setattr(
        services_module, "scrap_patent_async", _raises(httpx.ConnectTimeout("timed out")))

    with pytest.raises(UpstreamUnavailable) as exc_info:
        await make_service(ops_configured=False).scrap(PATENT_ID)

    assert exc_info.value.timeout is True


async def test_an_unparseable_page_is_an_upstream_failure(monkeypatch):
    """parse_patent_html raises ValueError when the page isn't a patent page
    (an interstitial, or a markup change). That is our problem or theirs,
    but it is not evidence the patent doesn't exist.
    """
    monkeypatch.setattr(
        services_module, "scrap_patent_async", _raises(ValueError("no DC.title")))

    with pytest.raises(UpstreamUnavailable):
        await make_service(ops_configured=False).scrap(PATENT_ID)


# --------------------------------- the OPS fallback ---------------------------------


async def test_falls_back_to_ops_when_google_patents_fails(monkeypatch):
    monkeypatch.setattr(services_module, "scrap_patent_async", _raises(_http_error(503)))
    monkeypatch.setattr(
        services_module, "ops_scrap_patent", _returns(PatentScrapResult(title="from OPS")))

    result = await make_service(ops_configured=True).scrap(PATENT_ID)

    assert result.title == "from OPS"


async def test_missing_from_both_backends_is_not_found(monkeypatch):
    monkeypatch.setattr(services_module, "scrap_patent_async", _raises(_http_error(404)))
    monkeypatch.setattr(services_module, "ops_scrap_patent", _raises(_http_error(404)))

    with pytest.raises(PatentNotFound):
        await make_service(ops_configured=True).scrap(PATENT_ID)


async def test_ops_failing_after_a_google_patents_404_is_not_a_miss(monkeypatch):
    """Google Patents says the patent is missing, but OPS - the backend that
    covers what Google Patents doesn't - never answered. That is unknown,
    not absent.
    """
    monkeypatch.setattr(services_module, "scrap_patent_async", _raises(_http_error(404)))
    monkeypatch.setattr(services_module, "ops_scrap_patent", _raises(_http_error(500)))

    with pytest.raises(UpstreamUnavailable):
        await make_service(ops_configured=True).scrap(PATENT_ID)


# ------------------------------ the OPS-only path ------------------------------


async def test_ops_only_retrieval_requires_credentials():
    with pytest.raises(OPSUnconfigured):
        await make_service(ops_configured=False).ops_scrap(PATENT_ID)


async def test_ops_only_retrieval_reports_a_real_miss(monkeypatch):
    """Here OPS is the only backend asked, so its 404 is the whole answer."""
    monkeypatch.setattr(services_module, "ops_scrap_patent", _raises(_http_error(404)))

    with pytest.raises(PatentNotFound):
        await make_service(ops_configured=True).ops_scrap(PATENT_ID)


async def test_ops_only_retrieval_flags_a_timeout(monkeypatch):
    monkeypatch.setattr(
        services_module, "ops_scrap_patent", _raises(httpx.ConnectTimeout("timed out")))

    with pytest.raises(UpstreamUnavailable) as exc_info:
        await make_service(ops_configured=True).ops_scrap(PATENT_ID)

    assert exc_info.value.timeout is True


# ---------------------------- domain error -> status code ----------------------------


@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_patent_service():
    """Swap the app's patent service for one the test controls, then put the
    real provider back."""
    def _install(service):
        app_module.app.dependency_overrides[app_module.get_patent_service] = lambda: service

    yield _install
    app_module.app.dependency_overrides.clear()


class _StubService:
    def __init__(self, error):
        self._error = error

    async def scrap(self, patent_id):
        raise self._error

    ops_scrap = scrap


@pytest.mark.parametrize("error, expected_status", [
    (PatentNotFound("gone"), 404),
    (UpstreamUnavailable("upstream broke"), 502),
    (UpstreamUnavailable("slow", timeout=True), 504),
    (OPSUnconfigured("no credentials"), 503),
])
async def test_domain_errors_map_onto_status_codes(
        client, override_patent_service, error, expected_status):
    override_patent_service(_StubService(error))

    resp = await client.get(f"/scrap/scrap_patent/{PATENT_ID}")

    assert resp.status_code == expected_status
    assert resp.json()["detail"] == str(error)