jpmschweitzerandClaude Fable 5
a66d801abd
Mock-based tests (no live services) covering:
- ollama_llm_model resolution under the OLLAMA_MODEL env collision
- per-leg retrieval failure -> source_status/degraded signaling
- Phase 0/Phase 4 LLM timeout fallbacks
- /stats wiki page count using the users/{user} path prefix
- Wiki.js list_all_pages limit-growth pagination and
list_pages limit-after-filter behavior
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
100 lines
2.7 KiB
Python
100 lines
2.7 KiB
Python
"""
|
|
Unit tests for the /stats endpoint (offline, all clients mocked).
|
|
|
|
Covers the wiki page count fix: the endpoint must count pages under the
|
|
user namespace ("users/{user}"), not pass the bare user name as prefix
|
|
(which matched nothing and always reported 0 pages).
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from src.main import stats
|
|
|
|
|
|
@pytest.fixture
|
|
def neo4j_client():
|
|
neo4j = MagicMock()
|
|
neo4j.execute_query = AsyncMock(return_value=[{"count": 5}])
|
|
return neo4j
|
|
|
|
|
|
@pytest.fixture
|
|
def qdrant_client():
|
|
qdrant = MagicMock()
|
|
qdrant.list_collections = AsyncMock(return_value=[
|
|
{"name": "user_jp", "vectors_count": 42}
|
|
])
|
|
return qdrant
|
|
|
|
|
|
@pytest.fixture
|
|
def wikijs_client():
|
|
wikijs = MagicMock()
|
|
wikijs.list_all_pages = AsyncMock(return_value=[
|
|
{"id": i, "path": f"users/jpmschweitzer/p{i}"} for i in range(138)
|
|
])
|
|
return wikijs
|
|
|
|
|
|
@pytest.fixture
|
|
def paperless_client():
|
|
paperless = MagicMock()
|
|
paperless.list_documents = AsyncMock(return_value={"count": 3})
|
|
paperless.list_tags = AsyncMock(return_value=[])
|
|
paperless.list_correspondents = AsyncMock(return_value=[])
|
|
paperless.list_document_types = AsyncMock(return_value=[])
|
|
return paperless
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestStatsWikiPageCount:
|
|
"""/stats must scope the wiki page count to the user namespace."""
|
|
|
|
async def test_uses_user_namespace_path_prefix(
|
|
self, neo4j_client, qdrant_client, wikijs_client, paperless_client
|
|
):
|
|
await stats(
|
|
user="jpmschweitzer",
|
|
neo4j=neo4j_client,
|
|
qdrant=qdrant_client,
|
|
wikijs=wikijs_client,
|
|
paperless=paperless_client,
|
|
api_key="test-key",
|
|
)
|
|
|
|
wikijs_client.list_all_pages.assert_awaited_once_with(
|
|
path_prefix="users/jpmschweitzer"
|
|
)
|
|
|
|
async def test_reports_page_count(
|
|
self, neo4j_client, qdrant_client, wikijs_client, paperless_client
|
|
):
|
|
response = await stats(
|
|
user="jpmschweitzer",
|
|
neo4j=neo4j_client,
|
|
qdrant=qdrant_client,
|
|
wikijs=wikijs_client,
|
|
paperless=paperless_client,
|
|
api_key="test-key",
|
|
)
|
|
|
|
assert response.wiki_pages == 138
|
|
|
|
async def test_wiki_failure_degrades_to_zero(
|
|
self, neo4j_client, qdrant_client, wikijs_client, paperless_client
|
|
):
|
|
wikijs_client.list_all_pages = AsyncMock(side_effect=RuntimeError("wiki down"))
|
|
|
|
response = await stats(
|
|
user="jpmschweitzer",
|
|
neo4j=neo4j_client,
|
|
qdrant=qdrant_client,
|
|
wikijs=wikijs_client,
|
|
paperless=paperless_client,
|
|
api_key="test-key",
|
|
)
|
|
|
|
assert response.wiki_pages == 0
|