jpmschweitzerandClaude Fable 5
b47c5b9281
Session-scoped autouse guard in tests/conftest.py refuses to run any test (pytest.exit, returncode 1) when the effective tenant resolves to the production tenant jpmschweitzer - the same guard library-desk applies on its side. _initialize_app now depends on the guard so the refusal happens before any initialization. Suite-level assertions pin that the live session runs under the llm_tester namespaces: Qdrant memories_llm_tester collection and Redis session:llm_tester:* keys. The biographer/memory unit tests already run fully mocked (no shared-service writes); the e2e isolation tests already used llm_tester - their constants now derive from the shared TEST_TENANT/PRODUCTION_TENANT config constants so a drift fails loudly instead of silently splitting. Verified: ENVIRONMENT=production pytest run exits 1 with the TENANT GUARD message and zero tests executed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
"""
|
|
Shared test fixtures for all tests.
|
|
Following FastAPI testing best practices.
|
|
"""
|
|
import asyncio
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from src.main import app
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def _tenant_guard():
|
|
"""
|
|
Hard-fail the whole suite if the effective tenant resolves to the
|
|
production tenant ("jpmschweitzer").
|
|
|
|
Isolation is tenant-based: tests that touch shared services
|
|
(Qdrant memories collections, Wiki.js via library-desk, Neo4j,
|
|
Redis) must run under the reserved test tenant "llm_tester" (or a
|
|
test_-prefixed namespace). This mirrors the guard library-desk
|
|
applies on its side.
|
|
"""
|
|
from src.core.config import PRODUCTION_TENANT, config
|
|
from src.core.context import get_default_user
|
|
from src.core.multi_tenancy import get_memory_collection_name
|
|
|
|
effective = get_default_user()
|
|
if (
|
|
effective == PRODUCTION_TENANT
|
|
or config.effective_default_user == PRODUCTION_TENANT
|
|
):
|
|
pytest.exit(
|
|
f"TENANT GUARD: refusing to run the test suite - the effective "
|
|
f"tenant resolves to the production tenant '{PRODUCTION_TENANT}' "
|
|
f"(ENVIRONMENT={config.ENVIRONMENT.value}, "
|
|
f"DEFAULT_USER={config.DEFAULT_USER}). Tests must run under "
|
|
f"'llm_tester' or a test_-prefixed tenant.",
|
|
returncode=1,
|
|
)
|
|
|
|
# The Qdrant memories namespace derived from the effective tenant
|
|
# must never be the production collection.
|
|
assert get_memory_collection_name(effective) != get_memory_collection_name(
|
|
PRODUCTION_TENANT
|
|
), "test suite would target the production memories collection"
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def _initialize_app(_tenant_guard):
|
|
"""
|
|
Run application lifespan (Claude health check, household registration, etc.)
|
|
once per test session. ASGITransport doesn't trigger lifespan events,
|
|
so we call it explicitly.
|
|
|
|
Depends on _tenant_guard so the suite refuses to start under the
|
|
production tenant before any initialization happens.
|
|
"""
|
|
from src.core.startup import initialize_application
|
|
asyncio.run(initialize_application())
|
|
|
|
|
|
@pytest.fixture
|
|
def client() -> TestClient:
|
|
"""
|
|
Synchronous test client for FastAPI.
|
|
|
|
Use for simple tests that don't require async.
|
|
"""
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture
|
|
async def async_client() -> AsyncClient:
|
|
"""
|
|
Async test client for FastAPI.
|
|
|
|
Use for testing async endpoints and streaming.
|
|
"""
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app),
|
|
base_url="http://test"
|
|
) as client:
|
|
yield client
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_chat_request() -> dict:
|
|
"""Standard chat completion request fixture."""
|
|
return {
|
|
"model": "lorem-tester",
|
|
"messages": [
|
|
{"role": "user", "content": "Hello, world!"}
|
|
],
|
|
"temperature": 0.7,
|
|
"stream": False,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_streaming_chat_request(mock_chat_request) -> dict:
|
|
"""Streaming chat completion request fixture."""
|
|
return {**mock_chat_request, "stream": True}
|