File size: 5,252 Bytes
b4fd35c | 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 | """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.')
|