""" Offline unit tests for the weekly quality-report endpoint (Phase C item 3). All external clients are mocked - no shared services are contacted. Live verification against the local dev server runs in tests/test_quality_report_live.py (integration-marked). """ import json from datetime import datetime, timedelta, timezone from unittest.mock import AsyncMock import pytest from src.routers.maintenance import ( QualityReportRequest, _parse_wiki_timestamp, quality_report, ) TEST_USER = "llm_tester" TENANT_PREFIX = f"users/{TEST_USER}" def _iso(days_ago: int) -> str: return (datetime.now(timezone.utc) - timedelta(days=days_ago)).isoformat() def _mock_stack( pages, hit_rows, duplicate_scan=None, cached_integrity=None, ): vector_service = AsyncMock() vector_service.find_duplicate_pairs = AsyncMock( return_value=duplicate_scan or {"chunks_scanned": 0, "duplicate_groups": []} ) # Used only when integrity runs inline vector_service.get_all_chunk_references = AsyncMock(return_value=[]) graph_service = AsyncMock() graph_service.neo4j = AsyncMock() graph_service.neo4j.execute_query = AsyncMock(return_value=hit_rows) graph_service.get_all_document_references = AsyncMock(return_value=[]) wiki_client = AsyncMock() # First call: tenant listing; later calls: report-path existence check wiki_client.list_all_pages = AsyncMock(side_effect=[pages, []]) wiki_client.create_page = AsyncMock(return_value={"id": 777}) qdrant = AsyncMock() qdrant.list_collections = AsyncMock(return_value=[]) redis = AsyncMock() async def _redis_get(key): if key.startswith("library:integrity:latest:") and cached_integrity: return json.dumps(cached_integrity) return None redis.get = AsyncMock(side_effect=_redis_get) return vector_service, graph_service, wiki_client, qdrant, redis class TestQualityReport: @pytest.mark.asyncio async def test_full_report_flags_and_writes_page(self): pages = [ # stale: 60 days old, 0 hits, missing tags+description {"id": 1, "path": f"{TENANT_PREFIX}/old-page", "title": "Old", "tags": [], "description": "", "updatedAt": _iso(60)}, # old but frequently found -> NOT stale; has metadata {"id": 2, "path": f"{TENANT_PREFIX}/popular", "title": "Popular", "tags": ["x"], "description": "d", "updatedAt": _iso(60)}, # fresh page missing description only {"id": 3, "path": f"{TENANT_PREFIX}/fresh", "title": "Fresh", "tags": ["y"], "description": "", "updatedAt": _iso(1)}, # system report page is exempt from all checks {"id": 4, "path": f"{TENANT_PREFIX}/system/quality-reports/2026-07-07", "title": "Old report", "tags": [], "description": "", "updatedAt": _iso(7)}, ] hit_rows = [ {"page_id": 1, "hits": 0}, {"page_id": 2, "hits": 9}, {"page_id": 3, "hits": 0}, ] duplicate_scan = { "chunks_scanned": 10, "duplicate_groups": [{ "pages": [ {"page_id": 1, "path": f"{TENANT_PREFIX}/old-page", "title": "Old"}, {"page_id": 2, "path": f"{TENANT_PREFIX}/popular", "title": "Popular"}, ], "max_similarity": 0.93, "matching_chunk_pairs": 2, }], } cached_integrity = { "generated_at": _iso(0), "counts": {"pages_without_vectors": 1, "orphaned_vector_chunks": 0, "documents_without_wiki": 0, "unexpected_collections": 2}, "unexpected_collections": [ {"name": "library_desk_ghost", "category": "unknown_tenant"}, ], } vector_service, graph_service, wiki_client, qdrant, redis = _mock_stack( pages, hit_rows, duplicate_scan, cached_integrity ) result = await quality_report( request=QualityReportRequest(user=TEST_USER, stale_days=30), vector_service=vector_service, graph_service=graph_service, wiki_client=wiki_client, qdrant=qdrant, redis=redis, api_key="", ) assert result.success is True # Stale: only page 1 (page 2 old but popular, page 3 fresh, page 4 exempt) assert [p["page_id"] for p in result.stale_pages] == [1] # Missing metadata: page 1 (tags+description), page 3 (description) missing = {p["page_id"]: p["missing"] for p in result.pages_missing_metadata} assert missing == {1: ["tags", "description"], 3: ["description"]} # Dedup folded in assert result.counts["duplicate_groups"] == 1 # Cached integrity used (no inline re-run needed) assert result.integrity["source"] == "cached" vector_service.get_all_chunk_references.assert_not_awaited() # Page written under the dated report path today = datetime.now(timezone.utc).date().isoformat() assert result.page_path == f"{TENANT_PREFIX}/system/quality-reports/{today}" assert result.page_id == 777 wiki_client.create_page.assert_awaited_once() create_kwargs = wiki_client.create_page.await_args.kwargs assert create_kwargs["path"] == result.page_path assert "auto-generated" in create_kwargs["tags"] # Report content mentions the key findings assert "old-page" in result.report assert "0.930" in result.report assert "library_desk_ghost" in result.report @pytest.mark.asyncio async def test_runs_integrity_inline_when_no_cache(self): vector_service, graph_service, wiki_client, qdrant, redis = _mock_stack( pages=[], hit_rows=[], cached_integrity=None ) # Inline integrity re-lists all pages: give the side_effect one more value wiki_client.list_all_pages = AsyncMock(side_effect=[[], [], []]) result = await quality_report( request=QualityReportRequest(user=TEST_USER), vector_service=vector_service, graph_service=graph_service, wiki_client=wiki_client, qdrant=qdrant, redis=redis, api_key="", ) assert result.integrity["source"] == "inline" vector_service.get_all_chunk_references.assert_awaited_once() @pytest.mark.asyncio async def test_write_page_false_skips_wiki_write(self): vector_service, graph_service, wiki_client, qdrant, redis = _mock_stack( pages=[], hit_rows=[], cached_integrity={"generated_at": _iso(0), "counts": {}, "unexpected_collections": []}, ) result = await quality_report( request=QualityReportRequest(user=TEST_USER, write_page=False), vector_service=vector_service, graph_service=graph_service, wiki_client=wiki_client, qdrant=qdrant, redis=redis, api_key="", ) assert result.page_path is None assert result.page_id is None wiki_client.create_page.assert_not_awaited() wiki_client.update_page.assert_not_awaited() assert result.report # content still returned @pytest.mark.asyncio async def test_same_day_rerun_updates_existing_page(self): today = datetime.now(timezone.utc).date().isoformat() report_path = f"{TENANT_PREFIX}/system/quality-reports/{today}" vector_service, graph_service, wiki_client, qdrant, redis = _mock_stack( pages=[], hit_rows=[], cached_integrity={"generated_at": _iso(0), "counts": {}, "unexpected_collections": []}, ) wiki_client.list_all_pages = AsyncMock(side_effect=[ [], # tenant listing [{"id": 555, "path": report_path, "title": f"Quality Report {today}"}], ]) result = await quality_report( request=QualityReportRequest(user=TEST_USER), vector_service=vector_service, graph_service=graph_service, wiki_client=wiki_client, qdrant=qdrant, redis=redis, api_key="", ) assert result.page_id == 555 wiki_client.update_page.assert_awaited_once() wiki_client.create_page.assert_not_awaited() def test_requires_user(self): with pytest.raises(Exception): QualityReportRequest(user="") def test_parse_wiki_timestamp(): parsed = _parse_wiki_timestamp("2026-07-14T09:39:45.244Z") assert parsed is not None and parsed.tzinfo is not None assert _parse_wiki_timestamp(None) is None assert _parse_wiki_timestamp("not-a-date") is None