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

test: pin suite to reserved llm_tester tenant with guard and teardown

Browse Source 
Rewrite tests/conftest.py for the shared-services testing model where
tenancy is the only isolation wall:

- Remove the hardcoded production host default (192.168.86.149):
  TEST_HOST env with a safe localhost default; LIBRARY_DESK_URL selects
  the local wakeup server (8778), never the production container (8089).
- Pin the suite to the reserved test tenant llm_tester (TEST_TENANT may
  only choose a tenant inside the reserved llm_tester* namespace).
- Session guard (autouse) hard-aborts the whole run if the effective
  tenant is jpmschweitzer or outside the reserved namespace.
- Integration-marked tests only run with RUN_INTEGRATION_TESTS=1 and a
  passing guard; they are skipped otherwise.
- Session-scoped teardown deletes ALL llm_tester artifacts created
  during the run: Qdrant *_llm_tester collections, Neo4j
  User_Llm_Tester* nodes, wiki subtree users/llm_tester (and hyphen
  variant), llm_tester Redis keys on the service DB - with hard
  assert_safe_test_tenant() checks before every delete. Uses a sync
  fixture + asyncio.run to avoid the session loop-scope mismatch.
- Legacy tests/test_integration.py marked integration and pinned to the
  test tenant (taxonomy/list reads no longer touch the production
  namespace; Qdrant tests use the tenant-scoped collection name).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2026-07-14 11:21:23 +02:00
co-authored by Claude Fable 5
parent 33e62db6a2
commit b2399de5f9
3 changed files with 326 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+1
View File
@@ -26,6 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- **Degradation signaling** - `HybridRAGResponse` now includes `source_status` (per-leg `'ok'`/`'failed'`/`'disabled'` for vector, graph, web, volatile, documents) and `degraded` (true when any enabled leg failed). Retrieval legs report errors instead of silently swallowing them; failed legs are logged at WARNING. Both fields are additive and optional, so clients that ignore them are unaffected.
- **Hard-isolated test-tenant lifecycle for the test suite** - `tests/conftest.py` rewritten: the production host default (`192.168.86.149`) is gone (`TEST_HOST` env, safe `localhost` default; the API under test is the local wakeup server via `LIBRARY_DESK_URL`, never the production container on 8089). The suite is pinned to the reserved test tenant `llm_tester`; a session guard aborts the entire run if the effective tenant is `jpmschweitzer` or outside the reserved `llm_tester*` namespace. Integration tests are marked and only run with `RUN_INTEGRATION_TESTS=1` (plus a passing guard). A session-scoped teardown deletes ALL `llm_tester` artifacts created during the run — Qdrant `*_llm_tester` collections, Neo4j `User_Llm_Tester*`-labelled nodes, the `users/llm_tester` wiki subtree, and `llm_tester` Redis keys on the service DB — with hard tenant assertions before every delete. Legacy integration tests were pinned to the test tenant (no more production-namespace reads).
- **Offline unit tests** - New mock-based tests (no live services) for model-name resolution under the env collision, per-leg failure signaling, the `/stats` page-count prefix, LLM-call timeouts, and Wiki.js listing pagination.
### Fixed
tests/conftest.py
+302 -13
View File
@@ -1,21 +1,308 @@
"""Pytest configuration and shared fixtures for Library Desk tests."""
"""
Pytest configuration and shared fixtures for Library Desk tests.
TENANT SAFETY MODEL
===================
There is no separate test infrastructure: integration tests run against the
SHARED production services (Qdrant / Neo4j / Wiki.js / Redis / Wiki.js).
Tenancy is the ONLY isolation wall, therefore:
- The suite is pinned to the reserved test tenant ``llm_tester`` (env
``TEST_TENANT`` may only select a tenant inside the reserved
``llm_tester*`` namespace — anything else aborts the whole session).
- The production tenant ``jpmschweitzer`` is NEVER written to. A session
guard hard-fails immediately if the effective tenant is the production
tenant or outside the reserved namespace.
- Integration tests (marked ``integration``) only run when
``RUN_INTEGRATION_TESTS=1`` is set; otherwise they are skipped. Offline
unit tests never contact the shared services.
- A session-scoped teardown deletes ALL ``llm_tester`` artifacts created
during the run (Qdrant ``*_llm_tester`` collections, Neo4j nodes under
the ``User_Llm_Tester*`` labels, the ``users/llm_tester`` wiki subtree,
and ``llm_tester``-prefixed Redis keys on the service DB), with a hard
assertion on the tenant string before any delete.
HOSTS
=====
``TEST_HOST`` selects where the shared services live. It defaults to
``localhost`` (safe: nothing listens there unless you forwarded the
services yourself). For live runs against the shared stack set it
explicitly, e.g. ``TEST_HOST=192.168.86.149``.
The API under test is the LOCAL wakeup server (``./wakeup.sh``, port 8778),
selected via ``LIBRARY_DESK_URL`` (default ``http://localhost:8778``).
NEVER point tests at the production container (port 8089).
"""
import asyncio
import logging
import os
import pytest
import pytest_asyncio
from typing import AsyncGenerator
# Test configuration
import pytest
from src.core.multi_tenancy import sanitize_user_id
logger = logging.getLogger(__name__)
pytest_plugins = ("pytest_asyncio",)
# Use real host for tests (services available at this IP)
TEST_HOST = os.environ.get("TEST_HOST", "192.168.86.149")
# ---------------------------------------------------------------------------
# Tenancy constants
# ---------------------------------------------------------------------------
#: The production tenant. No test may ever write under it.
PRODUCTION_TENANT = "jpmschweitzer"
#: The reserved test tenant namespace. The effective tenant must be
#: exactly this or a sub-tenant of it (llm_tester_*).
RESERVED_TEST_TENANT = "llm_tester"
#: Effective tenant for the whole suite (guard-checked below).
TEST_TENANT = os.environ.get("TEST_TENANT", RESERVED_TEST_TENANT)
# ---------------------------------------------------------------------------
# Hosts / URLs
# ---------------------------------------------------------------------------
#: Shared-services host. Default localhost — NOT the production host.
TEST_HOST = os.environ.get("TEST_HOST", "localhost")
#: Base URL of the local dev server under test (./wakeup.sh, port 8778).
LIBRARY_DESK_URL = os.environ.get("LIBRARY_DESK_URL", "http://localhost:8778")
RUN_INTEGRATION = os.environ.get("RUN_INTEGRATION_TESTS") == "1"
def is_reserved_test_tenant(tenant: str) -> bool:
"""True if tenant is inside the reserved llm_tester namespace."""
sanitized = sanitize_user_id(tenant)
return sanitized == RESERVED_TEST_TENANT or sanitized.startswith(
RESERVED_TEST_TENANT + "_"
)
def assert_safe_test_tenant(tenant: str) -> str:
"""
Hard assertion used before ANY destructive operation.
Raises AssertionError unless the tenant is inside the reserved test
namespace and is not the production tenant.
"""
sanitized = sanitize_user_id(tenant)
assert sanitized != sanitize_user_id(PRODUCTION_TENANT), (
f"TENANT GUARD: refusing to touch production tenant {tenant!r}"
)
assert is_reserved_test_tenant(tenant), (
f"TENANT GUARD: {tenant!r} is not in the reserved test namespace "
f"({RESERVED_TEST_TENANT}*)"
)
return sanitized
# ---------------------------------------------------------------------------
# Session guard + integration gating
# ---------------------------------------------------------------------------
def pytest_collection_modifyitems(config, items):
"""Skip integration-marked tests unless explicitly enabled AND safe."""
if RUN_INTEGRATION and is_reserved_test_tenant(TEST_TENANT):
return
reason = (
"integration tests disabled (set RUN_INTEGRATION_TESTS=1, TEST_HOST "
"and TEST_TENANT inside the reserved llm_tester namespace to run "
"against the shared services)"
)
skip_marker = pytest.mark.skip(reason=reason)
for item in items:
if "integration" in item.keywords:
item.add_marker(skip_marker)
@pytest.fixture(scope="session", autouse=True)
def tenant_guard():
"""
Session guard: hard-fail the entire run if the effective tenant is the
production tenant or outside the reserved test namespace.
"""
if sanitize_user_id(TEST_TENANT) == sanitize_user_id(PRODUCTION_TENANT):
pytest.exit(
f"TENANT GUARD: effective test tenant is the PRODUCTION tenant "
f"({TEST_TENANT!r}) - aborting the whole session.",
returncode=3,
)
if not is_reserved_test_tenant(TEST_TENANT):
pytest.exit(
f"TENANT GUARD: effective test tenant {TEST_TENANT!r} is not in "
f"the reserved test namespace ({RESERVED_TEST_TENANT}*) - "
f"aborting the whole session.",
returncode=3,
)
yield
# ---------------------------------------------------------------------------
# Session teardown: purge ALL llm_tester artifacts created during the run
# ---------------------------------------------------------------------------
async def _purge_qdrant_test_artifacts() -> None:
"""Delete Qdrant collections belonging to the reserved test tenant."""
from qdrant_client import QdrantClient
tenant = assert_safe_test_tenant(TEST_TENANT)
client = QdrantClient(url=f"http://{TEST_HOST}:6333", timeout=10)
try:
for coll in client.get_collections().collections:
name = coll.name
# Only *_<tenant> style collections (library_desk_llm_tester,
# volatile_llm_tester, memories_llm_tester, ...).
if not (name.endswith(f"_{tenant}") or f"_{tenant}_" in name):
continue
assert PRODUCTION_TENANT not in name # hard guard
assert tenant in name
client.delete_collection(name)
logger.info(f"[teardown] deleted Qdrant collection {name}")
finally:
client.close()
async def _purge_neo4j_test_artifacts() -> None:
"""Delete Neo4j nodes under the reserved test tenant's labels."""
from src.clients.neo4j_client import Neo4jClient
from src.config import get_settings
from src.core.multi_tenancy import get_neo4j_user_base_label
tenant = assert_safe_test_tenant(TEST_TENANT)
label_prefix = get_neo4j_user_base_label(tenant) # e.g. User_Llm_Tester
assert "Jpmschweitzer" not in label_prefix # hard guard
assert "Llm_Tester" in label_prefix
settings = get_settings()
client = Neo4jClient(
uri=f"bolt://{TEST_HOST}:7687",
user=settings.neo4j_user,
password=settings.neo4j_password,
)
try:
await client.connect()
result = await client.execute_write(
"""
MATCH (n)
WHERE any(l IN labels(n) WHERE l STARTS WITH $prefix)
DETACH DELETE n
RETURN count(n) AS deleted
""",
{"prefix": label_prefix},
)
deleted = result[0]["deleted"] if result else 0
if deleted:
logger.info(f"[teardown] deleted {deleted} Neo4j {label_prefix}* nodes")
finally:
await client.close()
async def _purge_wiki_test_artifacts() -> None:
"""Delete the reserved test tenant's wiki subtree (users/llm_tester)."""
from src.clients.wikijs_client import WikiJSClient
from src.config import get_settings
tenant = assert_safe_test_tenant(TEST_TENANT)
settings = get_settings()
client = WikiJSClient(
base_url=settings.wikijs_url,
api_token=settings.wiki_graphql_api,
)
try:
for prefix in (f"users/{tenant}", f"users/{tenant.replace('_', '-')}"):
assert PRODUCTION_TENANT not in prefix # hard guard
pages = await client.list_all_pages(path_prefix=prefix)
for page in pages:
path = page.get("path", "")
assert PRODUCTION_TENANT not in path # hard guard
assert path.lstrip("/").startswith("users/llm")
await client.delete_page(page["id"])
logger.info(f"[teardown] deleted wiki page {path} (id={page['id']})")
finally:
await client.close()
async def _purge_redis_test_artifacts() -> None:
"""Delete llm_tester-prefixed keys on the Redis DB the service uses."""
import redis.asyncio as aioredis
from src.config import get_settings
tenant = assert_safe_test_tenant(TEST_TENANT)
settings = get_settings()
client = aioredis.from_url(
f"redis://{TEST_HOST}:6379/{settings.redis_db}",
encoding="utf-8",
decode_responses=True,
)
try:
deleted = 0
for pattern in (f"*{tenant}*", f"*{tenant.replace('_', '-')}*"):
async for key in client.scan_iter(match=pattern, count=200):
assert PRODUCTION_TENANT not in key # hard guard
assert "llm" in key
await client.delete(key)
deleted += 1
if deleted:
logger.info(f"[teardown] deleted {deleted} Redis keys for {tenant}")
finally:
await client.aclose()
@pytest.fixture(scope="session", autouse=True)
def purge_test_tenant_artifacts(tenant_guard):
"""
Session teardown: after the run, delete ALL artifacts under the reserved
test tenant on the shared services. Only active for integration runs
(offline unit-test runs never touch the shared services and must not
try to connect to them).
"""
yield
if not RUN_INTEGRATION:
return
# Hard assertion before ANY delete.
assert_safe_test_tenant(TEST_TENANT)
async def _teardown():
for name, purge in (
("qdrant", _purge_qdrant_test_artifacts),
("neo4j", _purge_neo4j_test_artifacts),
("wiki", _purge_wiki_test_artifacts),
("redis", _purge_redis_test_artifacts),
):
try:
await purge()
except AssertionError:
raise # tenant-guard violations must never be swallowed
except Exception as e:
logger.warning(f"[teardown] {name} purge failed: {e}")
# Sync fixture + asyncio.run avoids the session/function loop-scope
# mismatch (see CLAUDE.md).
asyncio.run(_teardown())
# ---------------------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def test_user() -> str:
"""Default test user."""
return "test_user"
"""The reserved test tenant. Pinned for the whole suite."""
return TEST_TENANT
@pytest.fixture
def library_desk_url() -> str:
"""Base URL of the LOCAL dev server under test (never production :8089)."""
return LIBRARY_DESK_URL
@pytest.fixture
@@ -40,11 +327,11 @@ def qdrant_test_url() -> str:
@pytest.fixture
def wikijs_test_config() -> dict:
"""Test Wiki.js configuration."""
"""Test Wiki.js configuration (URL from settings, host env-overridable)."""
from src.config import get_settings
settings = get_settings()
return {
"base_url": f"http://{TEST_HOST}:3000",
"base_url": settings.wikijs_url,
"api_token": settings.wiki_graphql_api
}
@@ -69,7 +356,9 @@ def ollama_test_config() -> dict:
@pytest.fixture
def redis_test_url() -> str:
"""Test Redis URL."""
return f"redis://{TEST_HOST}:6379/4"
from src.config import get_settings
settings = get_settings()
return f"redis://{TEST_HOST}:6379/{settings.redis_db}"
@pytest.fixture
@@ -81,7 +370,7 @@ def sample_document() -> dict:
"content": "This is a test document for unit testing.",
"metadata": {
"source": "test",
"author": "test_user"
"author": TEST_TENANT
}
}
tests/test_integration.py
+23 -6
View File
@@ -7,13 +7,22 @@ These tests require actual service connectivity:
- SearXNG running at http://searxng:8080
- Ollama running at http://ollama:11434
Run with: pytest tests/test_integration.py -v
These tests run against the SHARED production services under the reserved
test tenant only. They are gated behind the session tenant guard and the
RUN_INTEGRATION_TESTS=1 environment flag (see tests/conftest.py) and are
skipped otherwise.
Run with:
RUN_INTEGRATION_TESTS=1 TEST_HOST=<shared-host> \
.venv/bin/python -m pytest tests/test_integration.py -v
"""
import pytest
import pytest_asyncio
from typing import AsyncGenerator
from tests.conftest import TEST_TENANT, assert_safe_test_tenant
from src.clients.neo4j_client import Neo4jClient
from src.clients.qdrant_client import QdrantClientWrapper
from src.clients.wikijs_client import WikiJSClient
@@ -22,6 +31,9 @@ from src.clients.ollama_client import OllamaClient
from src.jobs.job_manager import JobManager, JobType, JobStatus
from src.config import get_settings
# Only run when the tenant guard passes and integration mode is enabled.
pytestmark = pytest.mark.integration
@pytest.fixture
def settings():
@@ -100,6 +112,7 @@ class TestNeo4jIntegration:
@pytest.mark.asyncio
async def test_create_and_get_document(self, neo4j_client, test_user):
"""Test creating and retrieving a document node."""
assert_safe_test_tenant(test_user)
doc_id = "test_doc_integration"
# Create document
@@ -131,9 +144,10 @@ class TestQdrantIntegration:
@pytest.mark.asyncio
async def test_collection_creation(self, qdrant_client, test_user):
"""Test creating a collection."""
await qdrant_client.ensure_collection(test_user)
assert_safe_test_tenant(test_user)
collection_name = qdrant_client.get_collection_name(test_user)
await qdrant_client.ensure_collection(collection_name)
collections = qdrant_client.client.get_collections()
collection_names = [c.name for c in collections.collections]
@@ -142,7 +156,10 @@ class TestQdrantIntegration:
@pytest.mark.asyncio
async def test_upsert_and_search(self, qdrant_client, test_user):
"""Test upserting and searching chunks."""
await qdrant_client.ensure_collection(test_user)
assert_safe_test_tenant(test_user)
await qdrant_client.ensure_collection(
qdrant_client.get_collection_name(test_user)
)
# Create test chunks with 768-dimensional embeddings
chunks = [
@@ -190,7 +207,7 @@ class TestWikiJSIntegration:
@pytest.mark.asyncio
async def test_list_all_pages(self, wikijs_client):
"""Test listing all pages with pagination support."""
pages = await wikijs_client.list_all_pages(path_prefix="users/")
pages = await wikijs_client.list_all_pages(path_prefix=f"users/{TEST_TENANT}")
assert isinstance(pages, list)
# Verify each page has expected fields
for page in pages[:5]: # Check first 5
@@ -201,7 +218,7 @@ class TestWikiJSIntegration:
@pytest.mark.asyncio
async def test_get_taxonomy_structure(self, wikijs_client):
"""Test getting taxonomy structure for a user."""
taxonomy = await wikijs_client.get_taxonomy_structure("users/jpmschweitzer")
taxonomy = await wikijs_client.get_taxonomy_structure(f"users/{TEST_TENANT}")
assert isinstance(taxonomy, dict)
# Each key should be a category, value should be list of subcategories
for category, subcategories in taxonomy.items():