jpmschweitzerandClaude
78066fab1b
Mechanical only, and separated from the judgment calls that follow so the reviewable changes are not buried in a 98-file whitespace diff. 227 automatic fixes: 60 blank lines carrying whitespace, 60 unsorted import blocks, 34 Optional[X] to X | None, 28 unused imports, 16 deprecated typing imports, 12 datetime.timezone.utc to datetime.UTC, and assorted smaller modernisations. Then `ruff format` over src and tests: 98 files reformatted, 35 already conforming. No file among the unused-import findings defines __all__ or is an __init__.py, so nothing here removes a re-export. `make test`: 658 passed, unchanged from HEAD. Two things observed while verifying, neither addressed here: `pytest tests/` cannot collect — tests/e2e/test_orchestration_e2e.py uses an `e2e` marker that is not registered, and the config is strict about markers. This fails identically at HEAD, so it predates this change; `make test` passes because it ignores tests/e2e, tests/integration and tests/contracts. test_tatlock_tool_call_logging_calculator is flaky. It failed once in a full run with these changes and passed on the next, passes in isolation with them, and fails in isolation at HEAD. It is order- or timing-dependent, not a regression from this commit — established by running the full suite both ways rather than by reasoning about which change could have caused it. Co-Authored-By: Claude <noreply@anthropic.com>
100 lines
3.1 KiB
Python
100 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 ASGITransport, AsyncClient
|
|
|
|
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}
|