Spaces:
Sleeping
Sleeping
File size: 7,195 Bytes
5796881 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | 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
|