| """Exercise real function bodies with stdlib-only mock transport, not FastAPI/Pydantic.""" |
| import ast |
| import asyncio |
| import logging |
| from pathlib import Path |
| import re |
| import sys |
| from types import SimpleNamespace, ModuleType |
| from typing import Optional |
| from urllib.parse import quote, urlparse, parse_qs |
|
|
| source = Path(__file__).resolve().parents[1] / 'src/api/search.py' |
| tree = ast.parse(source.read_text()) |
| nodes = [node for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name != 'search_endpoint'] |
| namespace = dict(asyncio=asyncio, logging=logging, re=re, Optional=Optional, quote=quote, |
| SearchRequest=SimpleNamespace, logger=logging.getLogger('search-offline'), |
| PRIMO_REQUEST_TIMEOUT=0.005, PUBMED_REQUEST_TIMEOUT=0.005) |
| exec(compile(ast.Module(body=nodes, type_ignores=[]), str(source), 'exec'), namespace) |
|
|
| calls = [] |
| responses = [] |
| class ProviderError(Exception): |
| pass |
| class Client: |
| def __init__(self, **kwargs): |
| pass |
| async def __aenter__(self): |
| return self |
| async def __aexit__(self, *args): |
| pass |
| async def get(self, url, **kwargs): |
| calls.append((url, kwargs)) |
| item = responses.pop(0) |
| if isinstance(item, Exception): |
| raise item |
| if item == 'timeout': |
| await asyncio.sleep(0.05) |
| status, payload = item |
| return SimpleNamespace(status_code=status, json=lambda: payload) |
| httpx = ModuleType('httpx') |
| httpx.AsyncClient = Client |
| httpx.HTTPError = ProviderError |
| sys.modules['httpx'] = httpx |
|
|
| def req(**kwargs): |
| fields = dict(query='topic', boolean_query=None, peer_reviewed=False, open_access=False, |
| year_from=None, year_to=None, resource_type='articles', source='primo', limit=5) |
| fields.update(kwargs) |
| return SimpleNamespace(**fields) |
|
|
| def run(name, request, *args): |
| calls.clear() |
| return asyncio.run(namespace[name](request, *args)) |
|
|
| responses[:] = [(200, {'info': {'total': 0}, 'docs': []})] |
| run('search_primo', req(query='"solar energy" AND wind', peer_reviewed=True, open_access=True, |
| year_from='2019', year_to='2025'), 'SECRET_KEY') |
| qs = parse_qs(urlparse(calls[0][0]).query) |
| assert qs['q'] == ['any,contains,"solar energy" AND wind'] |
| assert qs['qInclude'] == ['facet_tlevel,exact,peer_reviewed|,|facet_tlevel,exact,open_access|,|facet_searchcreationdate,exact,[2019 TO 2025]|,|facet_rtype,exact,articles'] |
| assert 'SECRET_KEY' not in calls[0][0] |
| assert calls[0][1]['headers']['Authorization'] == 'apikey SECRET_KEY' |
|
|
| responses[:] = [(200, {'info': {'total': 1}, 'docs': [{'pnx': { |
| 'display': {'title': 'Example', 'creator': 'A. Author'}, |
| 'control': {'recordid': ['record/123']}, |
| 'links': {'linktorsrc': ['$$Uhttps://example.org/paper$$DRead article']}, |
| }}]})] |
| result = run('search_primo', req(), 'key') |
| assert result['results'][0]['link'] == 'https://example.org/paper' |
| assert result['results'][0]['creator'] == 'A. Author' |
| assert namespace['_result_link']('javascript:alert(1)', 'https://safe.example') == 'https://safe.example' |
|
|
| responses[:] = [ProviderError('private-secret-url')] * 3 |
| result = run('search_primo', req(), 'key') |
| assert len(calls) == 3 |
| assert result['error'] == 'PRIMO API unavailable' |
| assert 'private-secret-url' not in str(result) |
|
|
| responses[:] = [(429, {})] |
| result = run('search_primo', req(), 'key') |
| assert len(calls) == 1 |
| assert 'limit reached' in result['error'] |
|
|
| responses[:] = ['timeout'] * 3 |
| result = run('search_primo', req(), 'key') |
| assert len(calls) == 3 |
| assert result['error'] == 'PRIMO API unavailable' |
|
|
| responses[:] = [(200, {'esearchresult': {'count': '1', 'idlist': ['123'], 'querytranslation': 'mapped'}}), |
| (200, {'result': {'123': {'title': 'Review title', 'pubtype': ['Review'], |
| 'articleids': [{'idtype': 'doi', 'value': '10.1000/example'}]}}})] |
| result = run('search_pubmed', req(query='natural question', boolean_query='"heart failure" AND therapy', |
| open_access=True, peer_reviewed=True, year_from='2020', year_to='2025', source='pubmed')) |
| term = calls[0][1]['params']['term'] |
| assert 'natural question' not in term |
| assert '"heart failure" AND therapy' in term |
| assert 'free full text[sb]' in term |
| assert '2020:2025[dp]' in term |
| assert 'NOT "pubmed books"[sb]' in term |
| assert len(result['warnings']) == 2 |
| assert result['results'][0]['doi'] == '10.1000/example' |
| assert result['results'][0]['type'] == 'Review' |
| assert result['results'][0]['free_full_text'] is True |
| assert 'open_access' not in result['results'][0] |
| assert namespace['_pubmed_query'](req(query='anatomy', resource_type='books', year_to='1800')) == '((anatomy) AND (0001:1800[dp])) AND "pubmed books"[sb]' |
|
|
| responses[:] = [(200, {'esearchresult': {'errorlist': {'fieldsnotfound': ['badfield']}}})] |
| assert run('search_pubmed', req(source='pubmed'))['error'] |
| responses[:] = [(200, {})] |
| assert run('search_pubmed', req(source='pubmed'))['error'] |
| responses[:] = [ProviderError('private-diagnostic')] |
| assert 'private-diagnostic' not in str(run('search_pubmed', req(source='pubmed'))) |
| print('PASS: 9 offline function/transport scenarios. FastAPI/Pydantic validation and real httpx integration NOT exercised.') |
|
|
