tests/test_ops_client.py

import json
from pathlib import Path

import httpx
import pytest
import respx

import ops as ops_module
from ops import OPSNotConfigured, _OPSTokenManager, _ops_get, ops_scrap_patent, ops_search

FIXTURES = Path(__file__).parent / "fixtures" / "ops"
TOKEN_URL = "https://ops.epo.org/3.2/auth/accesstoken"
OPS_BASE = "https://ops.epo.org/3.2/rest-services"


def load_json(name: str):
    return json.loads((FIXTURES / name).read_text())


def mock_token_endpoint():
    return respx.post(TOKEN_URL).mock(
        return_value=httpx.Response(200, json=load_json("token_response.json"))
    )


@pytest.fixture
def fresh_token_manager(monkeypatch):
    """A token manager with test credentials, swapped in for the module-level
    singleton so `_ops_get`/`ops_search`/etc. (which reach for the module
    global `token_manager` directly) pick it up.
    """
    tm = _OPSTokenManager()
    tm._key = "test-key"
    tm._secret = "test-secret"
    monkeypatch.setattr(ops_module, "token_manager", tm)
    return tm


class TestTokenManager:
    async def test_not_configured_raises(self):
        tm = _OPSTokenManager()
        tm._key = None
        tm._secret = None
        async with httpx.AsyncClient() as client:
            with pytest.raises(OPSNotConfigured):
                await tm.get_token(client)

    async def test_fetches_and_caches_the_token(self):
        tm = _OPSTokenManager()
        tm._key = "k"
        tm._secret = "s"
        with respx.mock:
            route = mock_token_endpoint()
            async with httpx.AsyncClient() as client:
                token1 = await tm.get_token(client)
                token2 = await tm.get_token(client)

        assert token1 == "test-access-token-123"
        assert token2 == "test-access-token-123"
        assert route.call_count == 1

    async def test_force_refresh_refetches_even_if_cached(self):
        tm = _OPSTokenManager()
        tm._key = "k"
        tm._secret = "s"
        with respx.mock:
            route = mock_token_endpoint()
            async with httpx.AsyncClient() as client:
                await tm.get_token(client)
                await tm.get_token(client, force=True)

        assert route.call_count == 2


class TestOpsGet:
    async def test_retries_once_when_the_token_is_stale(self, fresh_token_manager):
        with respx.mock:
            mock_token_endpoint()
            get_route = respx.get(f"{OPS_BASE}/some/path").mock(
                side_effect=[
                    httpx.Response(401, text="Invalid access token."),
                    httpx.Response(200, json={"ok": True}),
                ]
            )
            async with httpx.AsyncClient() as client:
                data = await _ops_get(client, "/some/path")

        assert data == {"ok": True}
        assert get_route.call_count == 2

    async def test_propagates_non_token_errors(self, fresh_token_manager):
        with respx.mock:
            mock_token_endpoint()
            respx.get(f"{OPS_BASE}/some/path").mock(return_value=httpx.Response(500, text="server error"))
            async with httpx.AsyncClient() as client:
                with pytest.raises(httpx.HTTPStatusError):
                    await _ops_get(client, "/some/path")


class TestOpsSearch:
    async def test_maps_ops_search_results_to_serp_dicts(self, fresh_token_manager):
        with respx.mock:
            mock_token_endpoint()
            respx.get(f"{OPS_BASE}/published-data/search/biblio").mock(
                return_value=httpx.Response(200, json=load_json("search_response.json"))
            )
            async with httpx.AsyncClient() as client:
                results = await ops_search(client, "widget", 10)

        assert results == [
            {
                "title": "Widget apparatus",
                "body": "An apparatus comprising a widget.",
                "href": "https://worldwide.espacenet.com/patent/search?q=US11930446",
                "id": "US11930446",
            },
            {
                "title": "Gadget system",
                "body": "A system comprising a gadget.",
                "href": "https://worldwide.espacenet.com/patent/search?q=EP4760514",
                "id": "EP4760514",
            },
        ]

    async def test_returns_empty_list_on_404_not_found(self, fresh_token_manager):
        with respx.mock:
            mock_token_endpoint()
            respx.get(f"{OPS_BASE}/published-data/search/biblio").mock(
                return_value=httpx.Response(404, json=load_json("search_not_found_response.json"))
            )
            async with httpx.AsyncClient() as client:
                results = await ops_search(client, "nonexistent widget xyz", 10)

        assert results == []


class TestOpsScrapPatent:
    async def test_assembles_a_full_patent_from_three_endpoints(self, fresh_token_manager):
        number = "US11930446"
        with respx.mock:
            mock_token_endpoint()
            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/biblio").mock(
                return_value=httpx.Response(200, json=load_json("biblio_response.json"))
            )
            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/claims").mock(
                return_value=httpx.Response(200, json=load_json("claims_response.json"))
            )
            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/description").mock(
                return_value=httpx.Response(200, json=load_json("description_response.json"))
            )
            async with httpx.AsyncClient() as client:
                # Kind-code suffix ("B2") should be normalized away before
                # any of the three requests go out.
                result = await ops_scrap_patent(client, "US11930446B2")

        assert result.title == "Widget apparatus"
        assert result.abstract == "An apparatus comprising a widget."
        assert result.claims == "1. A widget comprising a gadget.\n2. The widget of claim 1, further comprising a mechanism."
        assert result.description == "This invention relates to widgets.\nThe widget comprises a gadget and a mechanism."
        assert {c.code for c in result.classifications} == {"G06F17/30", "H04L9/00"}

    async def test_missing_fulltext_degrades_to_none_instead_of_raising(self, fresh_token_manager):
        number = "US11930446"
        with respx.mock:
            mock_token_endpoint()
            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/biblio").mock(
                return_value=httpx.Response(200, json=load_json("biblio_response.json"))
            )
            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/claims").mock(
                return_value=httpx.Response(404)
            )
            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/description").mock(
                return_value=httpx.Response(404)
            )
            async with httpx.AsyncClient() as client:
                result = await ops_scrap_patent(client, number)

        assert result.title == "Widget apparatus"
        assert result.claims is None
        assert result.description is None