jpmschweitzerandClaude Fable 5
8cf3609948
Phase A review minors: - Coverage note: source_status (when present) is now used exclusively; the source_counts-absence fallback only considers the optional legs the request explicitly enabled (web/documents/volatile). library-desk computes source_counts from the final top-N fused results only, so absence of the always-on vector/graph legs is normal ranking behavior - the old heuristic warned on virtually every healthy search - update_wiki_page: the empty-list tags sentinel (leave unchanged) made clearing all tags impossible; pass exactly ["__CLEAR__"] to send an empty tag list, documented in the docstring for the local model - Text-delegation parallel fallback: zip(..., strict=True) with an explicit count-mismatch guard so results can never be silently attributed to the wrong agent Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L
345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""
|
|
Contract tests for HybridRAG parsing against a recorded live response.
|
|
|
|
The fixture in fixtures/hybrid_query_recorded.json is a real (recorded)
|
|
response from library-desk's POST /query/hybrid. These tests pin the
|
|
field mapping (source_type/sources, rrf_score, context, per-item
|
|
related_dossiers, keywords dict with nested synonyms) so a drift in
|
|
either side shows up as a test failure instead of every result
|
|
rendering as "unknown (score: 0.00)".
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src.agents.librarian.client import HybridRAGResponse, LibraryDeskClient
|
|
from src.agents.librarian.tools import SOURCE_ICONS, _coverage_note, hybrid_search
|
|
|
|
FIXTURE_PATH = Path(__file__).parent / "fixtures" / "hybrid_query_recorded.json"
|
|
|
|
|
|
@pytest.fixture
|
|
def recorded_response() -> dict:
|
|
"""Load the recorded /query/hybrid response."""
|
|
return json.loads(FIXTURE_PATH.read_text())
|
|
|
|
|
|
@pytest.fixture
|
|
def client_with_recorded_response(recorded_response):
|
|
"""LibraryDeskClient whose httpx client replays the recorded response."""
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = recorded_response
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_httpx = AsyncMock(spec=httpx.AsyncClient)
|
|
mock_httpx.post.return_value = mock_response
|
|
|
|
client = LibraryDeskClient(base_url="http://test:8089", api_key="test-key")
|
|
client._client = mock_httpx
|
|
return client
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestHybridRAGContract:
|
|
"""Contract tests for parsing the live /query/hybrid response shape."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sources_are_not_unknown(self, client_with_recorded_response):
|
|
"""Every result maps source_type - nothing falls back to 'unknown'."""
|
|
response = await client_with_recorded_response.hybrid_search(
|
|
"home server infrastructure", user="testuser"
|
|
)
|
|
|
|
assert isinstance(response, HybridRAGResponse)
|
|
assert response.results, "recorded fixture must contain results"
|
|
for result in response.results:
|
|
assert result.source != "unknown"
|
|
assert result.source in {"wiki", "web", "volatile", "document"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scores_are_non_zero(self, client_with_recorded_response):
|
|
"""rrf_score maps to score - no silent 0.00 fallback."""
|
|
response = await client_with_recorded_response.hybrid_search(
|
|
"home server infrastructure", user="testuser"
|
|
)
|
|
|
|
for result in response.results:
|
|
assert result.score > 0.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sources_list_and_icons(self, client_with_recorded_response):
|
|
"""Per-item sources list is parsed and every value has an icon."""
|
|
response = await client_with_recorded_response.hybrid_search(
|
|
"home server infrastructure", user="testuser"
|
|
)
|
|
|
|
for result in response.results:
|
|
assert result.sources, f"result '{result.title}' has empty sources"
|
|
for source in result.sources:
|
|
assert source in SOURCE_ICONS, f"no icon for source '{source}'"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_context_maps_to_formatted_context(
|
|
self, client_with_recorded_response
|
|
):
|
|
"""Top-level 'context' field maps to formatted_context."""
|
|
response = await client_with_recorded_response.hybrid_search(
|
|
"home server infrastructure", user="testuser"
|
|
)
|
|
|
|
assert response.formatted_context != ""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_keywords_and_synonyms_from_dict(
|
|
self, client_with_recorded_response
|
|
):
|
|
"""keywords is a dict: core_keywords + nested synonyms map."""
|
|
response = await client_with_recorded_response.hybrid_search(
|
|
"home server infrastructure", user="testuser"
|
|
)
|
|
|
|
assert response.keywords, "core_keywords should be extracted"
|
|
assert all(isinstance(k, str) for k in response.keywords)
|
|
# synonyms map in the fixture is empty, but must parse to a list
|
|
assert isinstance(response.synonyms, list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_per_item_related_dossiers(self, client_with_recorded_response):
|
|
"""related_dossiers live per result and aggregate to unique titles."""
|
|
response = await client_with_recorded_response.hybrid_search(
|
|
"home server infrastructure", user="testuser"
|
|
)
|
|
|
|
per_item = [d for r in response.results for d in r.related_dossiers]
|
|
assert per_item, "recorded fixture contains per-item related_dossiers"
|
|
for dossier in per_item:
|
|
assert "title" in dossier
|
|
assert "tag" in dossier
|
|
|
|
assert response.related_dossiers, "top-level titles are aggregated"
|
|
assert len(response.related_dossiers) == len(set(response.related_dossiers))
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_payload_never_sends_zero_limits(
|
|
self, client_with_recorded_response
|
|
):
|
|
"""The live service 422s on limits < 1; disabled legs use enable_* flags."""
|
|
await client_with_recorded_response.hybrid_search(
|
|
"home server infrastructure",
|
|
user="testuser",
|
|
web_limit=0,
|
|
document_limit=0,
|
|
volatile_limit=0,
|
|
)
|
|
|
|
payload = client_with_recorded_response._client.post.call_args.kwargs["json"]
|
|
config = payload["config"]
|
|
for key in (
|
|
"vector_limit",
|
|
"graph_limit",
|
|
"web_limit",
|
|
"document_limit",
|
|
"volatile_limit",
|
|
):
|
|
assert config[key] >= 1
|
|
assert config["enable_web"] is False
|
|
assert config["enable_documents"] is False
|
|
assert config["enable_volatile"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_always_sent_as_query_param(
|
|
self, client_with_recorded_response
|
|
):
|
|
"""The tenant is always sent explicitly - library-desk is removing
|
|
its server-side default, so a missing user would 422."""
|
|
await client_with_recorded_response.hybrid_search(
|
|
"home server infrastructure", user="testuser"
|
|
)
|
|
|
|
params = client_with_recorded_response._client.post.call_args.kwargs[
|
|
"params"
|
|
]
|
|
assert params["user"] == "testuser"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_counts_and_timing_parsed(
|
|
self, client_with_recorded_response
|
|
):
|
|
"""source_counts and timing map into the response model."""
|
|
response = await client_with_recorded_response.hybrid_search(
|
|
"home server infrastructure", user="testuser"
|
|
)
|
|
|
|
assert response.source_counts == {"graph": 3, "web": 2}
|
|
assert response.timing.get("total_ms", 0) > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_status_absent_is_tolerated(
|
|
self, client_with_recorded_response
|
|
):
|
|
"""Recorded response predates source_status/degraded - defaults apply."""
|
|
response = await client_with_recorded_response.hybrid_search(
|
|
"home server infrastructure", user="testuser"
|
|
)
|
|
|
|
assert response.source_status == {}
|
|
assert response.degraded is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_status_parsed_when_present(self, recorded_response):
|
|
"""Additive source_status/degraded fields parse when the service sends them."""
|
|
enriched = dict(recorded_response)
|
|
enriched["source_status"] = {
|
|
"vector": "ok",
|
|
"graph": "ok",
|
|
"web": "failed",
|
|
"volatile": "disabled",
|
|
"documents": "ok",
|
|
}
|
|
enriched["degraded"] = True
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = enriched
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_httpx = AsyncMock(spec=httpx.AsyncClient)
|
|
mock_httpx.post.return_value = mock_response
|
|
|
|
client = LibraryDeskClient(base_url="http://test:8089", api_key="test-key")
|
|
client._client = mock_httpx
|
|
|
|
response = await client.hybrid_search("home server infrastructure", user="u")
|
|
|
|
assert response.degraded is True
|
|
assert response.source_status["web"] == "failed"
|
|
assert response.source_status["volatile"] == "disabled"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_renders_no_unknown_results(self, client_with_recorded_response, monkeypatch):
|
|
"""The hybrid_search tool renders real sources and non-zero scores."""
|
|
|
|
class _Factory:
|
|
def __call__(self):
|
|
return self
|
|
|
|
async def __aenter__(self):
|
|
return client_with_recorded_response
|
|
|
|
async def __aexit__(self, *args):
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
"src.agents.librarian.tools.LibraryDeskClient", _Factory()
|
|
)
|
|
|
|
output = await hybrid_search("home server infrastructure")
|
|
|
|
assert "unknown" not in output
|
|
assert "score: 0.00" not in output
|
|
assert "•" not in output, "every source value should map to an icon"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCoverageNote:
|
|
"""Coverage note makes degraded searches visible to model and user."""
|
|
|
|
def _response(self, **kwargs) -> HybridRAGResponse:
|
|
return HybridRAGResponse(**kwargs)
|
|
|
|
def test_no_note_when_all_legs_report(self):
|
|
response = self._response(
|
|
source_counts={
|
|
"vector": 2,
|
|
"graph": 1,
|
|
"web": 2,
|
|
"documents": 1,
|
|
"volatile": 1,
|
|
},
|
|
)
|
|
note = _coverage_note(
|
|
response, include_web=True, include_documents=True, include_volatile=True
|
|
)
|
|
assert note == ""
|
|
|
|
def test_note_when_enabled_leg_missing_from_counts(self):
|
|
response = self._response(source_counts={"graph": 3, "web": 2})
|
|
note = _coverage_note(
|
|
response, include_web=True, include_documents=True, include_volatile=True
|
|
)
|
|
assert "Coverage note" in note
|
|
assert "documents" in note
|
|
assert "volatile" in note
|
|
# Always-on wiki legs are never inferred from count absence
|
|
assert "vector" not in note
|
|
assert "graph" not in note
|
|
|
|
def test_wiki_leg_absence_is_not_degradation(self):
|
|
"""vector/graph missing from top-N counts is healthy ranking, not outage."""
|
|
response = self._response(
|
|
source_counts={"web": 2, "documents": 1, "volatile": 1}
|
|
)
|
|
note = _coverage_note(
|
|
response, include_web=True, include_documents=True, include_volatile=True
|
|
)
|
|
assert note == ""
|
|
|
|
def test_disabled_legs_are_not_reported_missing(self):
|
|
response = self._response(source_counts={"vector": 2, "graph": 1})
|
|
note = _coverage_note(
|
|
response,
|
|
include_web=False,
|
|
include_documents=False,
|
|
include_volatile=False,
|
|
)
|
|
assert note == ""
|
|
|
|
def test_note_prefers_source_status_failures(self):
|
|
response = self._response(
|
|
source_counts={"vector": 2, "graph": 1},
|
|
source_status={
|
|
"vector": "ok",
|
|
"graph": "ok",
|
|
"web": "failed",
|
|
"volatile": "disabled",
|
|
"documents": "ok",
|
|
},
|
|
degraded=True,
|
|
)
|
|
note = _coverage_note(
|
|
response, include_web=True, include_documents=True, include_volatile=True
|
|
)
|
|
assert "failed" in note
|
|
assert "web" in note
|
|
# disabled legs are not reported as failures
|
|
assert "volatile" not in note
|
|
|
|
def test_no_note_when_status_all_ok(self):
|
|
response = self._response(
|
|
source_counts={"graph": 1},
|
|
source_status={
|
|
"vector": "ok",
|
|
"graph": "ok",
|
|
"web": "ok",
|
|
"volatile": "ok",
|
|
"documents": "ok",
|
|
},
|
|
degraded=False,
|
|
)
|
|
note = _coverage_note(
|
|
response, include_web=True, include_documents=True, include_volatile=True
|
|
)
|
|
assert note == ""
|
|
|
|
def test_degraded_without_named_failures(self):
|
|
response = self._response(
|
|
source_status={"vector": "ok", "graph": "ok"},
|
|
degraded=True,
|
|
)
|
|
note = _coverage_note(
|
|
response, include_web=True, include_documents=True, include_volatile=True
|
|
)
|
|
assert "partial" in note
|