135 lines
6.5 KiB
Python
135 lines
6.5 KiB
Python
"""Contract tests use real SQLite retrieval and API boundaries, not an LLM."""
|
|
import json
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.main as main
|
|
import app.routes as routes
|
|
import tools.rag.build_sqlite_index as indexer
|
|
import tools.rag.search_core as core
|
|
import tools.rag.search_utils as search
|
|
from tools.rag.rag_utils import build_rag_context
|
|
from tools.retrieval_contract import RetrievalResponse, error_result
|
|
|
|
|
|
@pytest.fixture
|
|
def index(tmp_path, monkeypatch):
|
|
documents = []; chunks = []
|
|
for number in range(3):
|
|
path = f'pages/topics/item{number}/README.md'
|
|
doc = {'path': path, 'title': f'Contract {number}', 'published': True}
|
|
documents.append(doc)
|
|
for part in range(2):
|
|
chunks.append({'chunk_id': path+f'::{part}', 'document_path': path,
|
|
'title': doc['title'], 'published': True, 'chunk_index': part,
|
|
'text': f'spoločný výraz {number} úsek {part}', 'heading_paths': [],
|
|
'tags': [], 'categories': []})
|
|
for name, rows in [('documents.json', documents), ('chunks.json', chunks)]:
|
|
(tmp_path/name).write_text(json.dumps(rows))
|
|
monkeypatch.setattr(indexer, 'embed_passages', lambda texts: np.ones((len(texts),4), dtype=np.float32))
|
|
monkeypatch.setattr(core, 'embed_query', lambda *args, **kwargs: np.ones(4, dtype=np.float32))
|
|
db = tmp_path/'zp_index.sqlite'
|
|
indexer.build_database(documents_file=tmp_path/'documents.json', chunks_file=tmp_path/'chunks.json', db_file=db)
|
|
monkeypatch.setattr(routes, 'DB_FILE', db)
|
|
return db
|
|
|
|
|
|
def test_rag_limits_are_reported_and_rich_sources_preserved(index):
|
|
raw = search.search_database(index, 'spoločný výraz', limit=2, max_per_document=1)
|
|
assert raw['status'] == 'ok' and raw['truncated'] and raw['complete'] is False
|
|
assert set(raw['truncation_reasons']) == {'result_limit','per_document_limit'}
|
|
assert len(raw['sources']) == len(raw['results']) == 2
|
|
rag = build_rag_context(index, 'spoločný výraz', limit=2)
|
|
RetrievalResponse.model_validate(rag)
|
|
assert rag['truncated'] and rag['complete'] is False and rag['reason'] is None
|
|
assert rag['index_version'] is None
|
|
assert len(rag['sources']) == rag['source_count']
|
|
assert 'retrieval' in rag['sources'][0] and 'text' in rag['sources'][0]
|
|
|
|
|
|
def test_candidate_cap_detected_by_extra_row(index, monkeypatch):
|
|
monkeypatch.setattr(search, 'MIN_CANDIDATES', 1)
|
|
monkeypatch.setattr(search, 'DEFAULT_CANDIDATE_MULTIPLIER', 1)
|
|
result = search.search_database(index, 'spoločný výraz', limit=1)
|
|
assert result['truncation_reasons'] == ['candidate_limit']
|
|
assert result['complete'] is False and len(result['results']) == 1
|
|
|
|
|
|
def test_ranked_rag_never_promises_exhaustive_answer(index):
|
|
result = build_rag_context(index, 'spoločný výraz', limit=20, max_per_document=0)
|
|
assert result['status']=='ok' and not result['truncated']
|
|
assert result['complete'] is None
|
|
|
|
|
|
def test_empty_rag_result_has_reason_and_null_version(index):
|
|
result = build_rag_context(index, ' ')
|
|
assert result['status']=='no_answer' and result['sources']==[]
|
|
assert result['reason']=='empty_query' and result['index_version'] is None
|
|
|
|
|
|
@pytest.mark.parametrize('path', ['/rag','/search'])
|
|
def test_api_success_contains_common_fields(index, path):
|
|
with TestClient(main.app) as client:
|
|
response = client.post(path, json={'query':'spoločný výraz','limit':2},
|
|
headers={'X-API-Key':'a'*64})
|
|
assert response.status_code==200
|
|
body = response.json()
|
|
assert set(RetrievalResponse.model_fields) <= body.keys()
|
|
RetrievalResponse.model_validate(body)
|
|
assert body['status']=='ok' and body['sources'] and body['complete'] is False
|
|
|
|
|
|
@pytest.mark.parametrize('path', ['/rag','/search'])
|
|
@pytest.mark.parametrize('code,reason,error', [
|
|
(400,'invalid_query',ValueError('PRIVATE_ERROR')),
|
|
(503,'retrieval_unavailable',RuntimeError('PRIVATE_ERROR')),
|
|
(500,'internal_error',TypeError('PRIVATE_ERROR')),
|
|
])
|
|
def test_api_errors_have_contract_and_preserve_http_status(index, monkeypatch, path, code, reason, error):
|
|
def fail(*args, **kwargs):
|
|
raise error
|
|
monkeypatch.setattr(routes, 'build_rag_context' if path=='/rag' else 'search_database', fail)
|
|
with TestClient(main.app) as client:
|
|
response = client.post(path, json={'query':'test'}, headers={'X-API-Key':'a'*64})
|
|
assert response.status_code==code
|
|
body = response.json()
|
|
assert body['status']=='error' and body['reason']==reason and body['sources']==[]
|
|
assert body['complete'] is False and body['index_version'] is None
|
|
assert 'PRIVATE_ERROR' not in response.text
|
|
assert 'detail' in body
|
|
|
|
|
|
@pytest.mark.parametrize('path', ['/rag','/search'])
|
|
def test_auth_and_validation_errors_use_same_envelope(index, path):
|
|
with TestClient(main.app) as client:
|
|
unauthorized = client.post(path, json={'query':'test'})
|
|
invalid = client.post(path, json={'query':'test','unknown':'PRIVATE_INPUT'},
|
|
headers={'X-API-Key':'a'*64})
|
|
assert unauthorized.status_code==401 and unauthorized.json()['reason']=='unauthorized'
|
|
assert invalid.status_code==422 and invalid.json()['reason']=='invalid_request'
|
|
assert 'PRIVATE_INPUT' not in invalid.text
|
|
for response in [unauthorized,invalid]:
|
|
assert set(RetrievalResponse.model_fields) <= response.json().keys()
|
|
|
|
|
|
def test_error_factory_does_not_invent_sources_or_active_version():
|
|
result = error_result(RuntimeError('password=PRIVATE'))
|
|
assert result['status']=='error' and result['sources']==[] and result['index_version'] is None
|
|
assert not result['truncated'] and result['complete'] is False
|
|
assert 'PRIVATE' not in json.dumps(result)
|
|
|
|
|
|
@pytest.mark.parametrize('path', ['/rag','/search'])
|
|
def test_malformed_backend_output_still_has_safe_error_contract(index, monkeypatch, path):
|
|
monkeypatch.setattr(routes, 'build_rag_context' if path=='/rag' else 'search_database',
|
|
lambda *args, **kwargs: {'unexpected':'PRIVATE_BACKEND_OUTPUT'})
|
|
with TestClient(main.app) as client:
|
|
response = client.post(path, json={'query':'test'}, headers={'X-API-Key':'a'*64})
|
|
assert response.status_code==500
|
|
assert set(RetrievalResponse.model_fields) <= response.json().keys()
|
|
assert response.json()['reason']=='internal_error' and response.json()['status']=='error'
|
|
assert 'PRIVATE_BACKEND_OUTPUT' not in response.text
|
|
assert response.headers['Cache-Control']=='no-store'
|