jpmschweitzer/library-desk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

feat: add weekly quality-report endpoint writing dated wiki report

Browse Source 
POST /maintenance/quality-report {user}:
- runs the tenant-scoped duplicate scan (cosine >= threshold, default 0.9)
- flags stale pages: not updated in stale_days AND <= max_search_hits
  SearchQuery FOUND hits from the tenant's graph data
- lists pages missing tags/description (report subtree exempt)
- folds in the latest integrity-check results (Redis cache from
  /maintenance/integrity-check, or run inline when absent)
- writes the dated report to users/{user}/system/quality-reports/YYYY-MM-DD
  via the existing wiki write path; same-day reruns update the same page
  (page id remembered in Redis because the Wiki.js listing lags creation)
- response returns the full markdown report + page path + counts +
  duration_ms

Also fixes WikiJSClient.update_page: Wiki.js 2.x requires tags on the
update mutation (server maps over it unconditionally); calls without tags
failed with "Cannot read properties of undefined (reading 'map')" -
which also silently broke the consolidation page-update path. Current
tags are now preserved when the caller supplies none.

Verified live end-to-end on the local dev server as llm_tester
(tests/test_quality_report_live.py, integration-marked): probe page
flagged for missing metadata, report page written and fetched back,
same-day rerun updates in place, teardown leaves zero llm_tester pages.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L
This commit is contained in:
jpmschweitzer
2026-07-14 12:22:19 +02:00
co-authored by Claude Fable 5
parent 77dc5b00a1
commit 191a8be6c5
5 changed files with 768 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_quality_report.py
+236
View File
@@ -0,0 +1,236 @@
"""
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
tests/test_quality_report_live.py
+161
View File
@@ -0,0 +1,161 @@
"""
Live end-to-end test for the weekly quality report (integration, guard-gated).
Runs against the LOCAL wakeup server (./wakeup.sh, port 8778 — never the
production container on 8089) with the shared backing services, entirely
under the reserved llm_tester tenant namespace:
1. Create + ingest a wiki page as ``llm_tester`` (deliberately without tags
so the report has something to flag).
2. POST /maintenance/quality-report {user: llm_tester}: asserts the report
is generated, mentions the created page in the missing-metadata section,
folds in integrity results, and writes the dated report page under
users/llm_tester/system/quality-reports/YYYY-MM-DD.
3. Fetches the written report page back from the wiki.
4. Teardown deletes the created pages; the session-scoped teardown in
conftest purges every remaining llm_tester artifact.
Run with:
RUN_INTEGRATION_TESTS=1 TEST_HOST=<shared-host> \\
LIBRARY_DESK_URL=http://localhost:8778 \\
.venv/bin/python -m pytest tests/test_quality_report_live.py -v
"""
import time
import uuid
from datetime import datetime, timezone
import httpx
import pytest
from tests.conftest import (
LIBRARY_DESK_URL,
PRODUCTION_TENANT,
TEST_TENANT,
assert_safe_test_tenant,
)
pytestmark = pytest.mark.integration
@pytest.fixture(scope="module")
def api():
"""HTTP client for the local dev server, with bearer auth."""
from src.config import get_settings
assert ":8089" not in LIBRARY_DESK_URL, (
"Refusing to run the live quality-report test against the "
"production container (port 8089); point LIBRARY_DESK_URL at ./wakeup.sh"
)
settings = get_settings()
client = httpx.Client(
base_url=LIBRARY_DESK_URL,
headers={"Authorization": f"Bearer {settings.library_api_key}"},
timeout=httpx.Timeout(300.0, connect=10.0),
)
yield client
client.close()
@pytest.fixture(scope="module")
def ingested_page(api):
"""Create + ingest a metadata-poor wiki page as the test tenant."""
assert_safe_test_tenant(TEST_TENANT)
slug = f"quality-probe-{uuid.uuid4().hex[:8]}"
create_resp = api.post(
"/wiki/pages",
json={
"title": f"Quality Probe {slug}",
"path": f"/quality-tests/{slug}",
"content": (
"# Quality Probe\n\n"
"Ephemeral page used to verify the weekly quality report. "
"It intentionally has no tags so the report flags it."
),
"description": "",
"tags": [],
"user": TEST_TENANT,
},
)
assert create_resp.status_code == 201, create_resp.text
page = create_resp.json()
page_id = page["id"]
assert page["path"].lstrip("/").startswith(f"users/{TEST_TENANT}")
ingest_resp = api.post(
"/ingest/page",
json={"page_id": page_id, "user": TEST_TENANT, "force_refresh": True},
)
assert ingest_resp.status_code == 200, ingest_resp.text
# Wiki.js updates its page-listing index asynchronously after creation;
# the quality report relies on that listing, so wait until the new page
# is visible (up to ~30s) before running the report.
deadline = time.time() + 30
while time.time() < deadline:
listing = api.get("/wiki/pages", params={"user": TEST_TENANT})
assert listing.status_code == 200, listing.text
if any(p["id"] == page_id for p in listing.json().get("pages", [])):
break
time.sleep(2)
else:
pytest.fail(f"Page {page_id} never appeared in the Wiki.js listing")
yield {"page_id": page_id, "path": page["path"]}
delete_resp = api.delete(f"/wiki/pages/{page_id}", params={"user": TEST_TENANT})
assert delete_resp.status_code == 200, delete_resp.text
class TestLiveQualityReport:
def test_quality_report_end_to_end(self, api, ingested_page):
assert_safe_test_tenant(TEST_TENANT)
resp = api.post(
"/maintenance/quality-report",
json={"user": TEST_TENANT, "stale_days": 30},
)
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["success"] is True
assert data["user"] == TEST_TENANT
assert data["duration_ms"] > 0
# The dated report page was written inside the tenant's namespace
today = datetime.now(timezone.utc).date().isoformat()
expected_path = f"users/{TEST_TENANT}/system/quality-reports/{today}"
assert data["page_path"] == expected_path
assert data["page_id"]
assert PRODUCTION_TENANT not in data["page_path"]
# The metadata-poor probe page is flagged
flagged_ids = {p["page_id"] for p in data["pages_missing_metadata"]}
assert ingested_page["page_id"] in flagged_ids
# Integrity results are folded in (cached or inline)
assert data["integrity"] is not None
assert data["integrity"]["source"] in ("cached", "inline")
assert "counts" in data["integrity"]
# Report content includes the summary + the probe page path
assert "## Summary" in data["report"]
assert ingested_page["path"].lstrip("/") in data["report"]
# The written page is retrievable from the wiki via the API
page_resp = api.get(
f"/wiki/pages/{data['page_id']}", params={"user": TEST_TENANT}
)
assert page_resp.status_code == 200, page_resp.text
page = page_resp.json()
assert page["path"].lstrip("/") == expected_path
assert "## Summary" in page["content"]
# Same-day rerun must update the same page, not create a duplicate
rerun = api.post(
"/maintenance/quality-report",
json={"user": TEST_TENANT, "stale_days": 30},
)
assert rerun.status_code == 200, rerun.text
assert rerun.json()["page_id"] == data["page_id"]