135 lines
4.7 KiB
Python
135 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from src.rag.qa_v1_1_73 import api as api_module
|
|
from src.rag.qa_v1_1_73 import pipeline as pipeline_module
|
|
from src.rag.qa_v1_1_73.schemas import EvidenceItem, RetrievalHypothesis
|
|
|
|
|
|
class FakePipeline:
|
|
def __init__(self, *, available: bool = True, result=None, error: Exception | None = None):
|
|
self.llm = SimpleNamespace(available=available)
|
|
self.result = result or {"success": True, "answer": "ok", "sources": []}
|
|
self.error = error
|
|
self.calls = []
|
|
|
|
def answer(self, question, **kwargs):
|
|
self.calls.append((question, kwargs))
|
|
if self.error:
|
|
raise self.error
|
|
return dict(self.result)
|
|
|
|
|
|
def source(pmid: str = "34573426") -> EvidenceItem:
|
|
return EvidenceItem(
|
|
evidence_id="S1",
|
|
kb="V16/V15",
|
|
source="AtRALF1",
|
|
relation="inhibits",
|
|
target="root growth",
|
|
evidence="AtRALF1 inhibits root growth.",
|
|
pmid=pmid,
|
|
)
|
|
|
|
|
|
def install_pipeline(monkeypatch, pipeline: FakePipeline) -> None:
|
|
monkeypatch.setattr(api_module, "get_pipeline", lambda: pipeline)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_rejects_explicit_use_llm_false(monkeypatch):
|
|
install_pipeline(monkeypatch, FakePipeline())
|
|
with pytest.raises(HTTPException) as exc:
|
|
await api_module.chat(api_module.QARequest(question="RALF有哪些功能？", use_llm=False))
|
|
assert exc.value.status_code == 422
|
|
assert exc.value.detail["answer"] == ""
|
|
assert exc.value.detail["error_code"] == "llm_required"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_llm_configuration_returns_503_without_answer(monkeypatch):
|
|
install_pipeline(monkeypatch, FakePipeline(available=False))
|
|
with pytest.raises(HTTPException) as exc:
|
|
await api_module.chat(api_module.QARequest(question="RALF有哪些功能？", use_llm=True))
|
|
assert exc.value.status_code == 503
|
|
assert exc.value.detail["answer"] == ""
|
|
assert exc.value.detail["error_code"] == "llm_unavailable"
|
|
|
|
|
|
def test_pipeline_rejects_unknown_visible_pmid():
|
|
pipeline = pipeline_module.QAV1Pipeline.__new__(pipeline_module.QAV1Pipeline)
|
|
pipeline.answer_composer = SimpleNamespace(
|
|
compose=lambda *_args, **_kwargs: "RALF参与根生长调控。[PMID: 99999999]"
|
|
)
|
|
with pytest.raises(pipeline_module.QAAnswerError) as exc:
|
|
pipeline._compose_validated_answer(
|
|
"RALF有哪些功能？", RetrievalHypothesis(), [source()], {}, [], ""
|
|
)
|
|
assert exc.value.error_code == "invalid_llm_citation"
|
|
assert exc.value.answer == ""
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("result", "error", "error_code"),
|
|
[
|
|
("", None, "llm_empty_answer"),
|
|
(None, TimeoutError("upstream timeout"), "llm_upstream_error"),
|
|
(None, KeyError("choices"), "llm_upstream_error"),
|
|
],
|
|
)
|
|
def test_pipeline_fails_closed_for_empty_or_upstream_llm_failure(result, error, error_code):
|
|
pipeline = pipeline_module.QAV1Pipeline.__new__(pipeline_module.QAV1Pipeline)
|
|
|
|
def compose(*_args, **_kwargs):
|
|
if error:
|
|
raise error
|
|
return result
|
|
|
|
pipeline.answer_composer = SimpleNamespace(compose=compose)
|
|
with pytest.raises(pipeline_module.QAAnswerError) as exc:
|
|
pipeline._compose_validated_answer(
|
|
"RALF有哪些功能？", RetrievalHypothesis(), [source()], {}, [], ""
|
|
)
|
|
assert exc.value.error_code == error_code
|
|
assert exc.value.answer == ""
|
|
|
|
|
|
def test_pipeline_accepts_only_selected_source_pmids():
|
|
pipeline = pipeline_module.QAV1Pipeline.__new__(pipeline_module.QAV1Pipeline)
|
|
pipeline.answer_composer = SimpleNamespace(
|
|
compose=lambda *_args, **_kwargs: "AtRALF1抑制根生长。[PMID: 34573426]"
|
|
)
|
|
answer, audit = pipeline._compose_validated_answer(
|
|
"RALF有哪些功能？", RetrievalHypothesis(), [source()], {}, [], ""
|
|
)
|
|
assert answer.endswith("[PMID: 34573426]")
|
|
assert audit == {
|
|
"valid": True,
|
|
"cited_pmids": ["34573426"],
|
|
"unknown_pmids": [],
|
|
"cited_pmcids": [],
|
|
"unknown_pmcids": [],
|
|
"cited_dois": [],
|
|
"unknown_dois": [],
|
|
"invalid_public_citations": [],
|
|
"malformed_citation_labels": [],
|
|
"internal_labels": [],
|
|
"rewrite_attempted": False,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_is_retrieval_only_even_without_llm(monkeypatch):
|
|
pipeline = FakePipeline(
|
|
available=False,
|
|
result={"success": True, "answer": "", "sources": [{"evidence_id": "S1"}]},
|
|
)
|
|
install_pipeline(monkeypatch, pipeline)
|
|
response = await api_module.search(api_module.QARequest(question="RALF有哪些功能？"))
|
|
assert response["answer"] == ""
|
|
assert pipeline.calls[0][1]["retrieval_only"] is True
|