"""Pytest configuration and fixtures.""" from __future__ import annotations from pathlib import Path import logging import os import typing from collections.abc import AsyncGenerator, Generator if not os.environ.get("MUSE_ENV"): os.environ["MUSE_ENV"] = "test" import pytest import pytest_asyncio from httpx import AsyncClient, ASGITransport from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from sqlalchemy.pool import NullPool from musehub.db import database from musehub.db.database import Base, get_db from musehub.db.musehub_models import MusehubIdentity # Force all ORM models into Base.metadata before any create_all/drop_all. # muse_cli_models is only imported inside init_db() in production; without # this explicit import, Base.metadata is non-deterministic in tests (depends # on import order), causing drop_all to miss tables that create_all later # tries to create — resulting in duplicate-key errors on pg_type. import musehub.db.muse_cli_models as _muse_cli_models # noqa: F401 from musehub.auth.request_signing import MSignContext, optional_signed_request, require_signed_request from musehub.main import app from musehub.rate_limits import limiter type _JobPayload = dict[str, str | int | bool | None] import musehub.auth.failure_limiter as _failure_limiter @pytest.fixture(autouse=True) def _stub_push_background_tasks(monkeypatch: pytest.MonkeyPatch) -> None: """Replace enqueue_push_intel with a no-op spy during tests. The push endpoint enqueues intel jobs into the DB. During tests we don't want a live worker processing those jobs concurrently. This fixture replaces enqueue_push_intel with a no-op that records calls in a module-level list so integration tests can assert on what was enqueued without touching the DB. """ import musehub.services.musehub_jobs as _jobs _jobs._test_enqueued_calls.clear() async def _spy_enqueue( session: AsyncSession, repo_id: str, head: str, domain_id: str | None = None ) -> None: _jobs._test_enqueued_calls.append((repo_id, "enqueue_push_intel", {"head": head, "domain_id": domain_id})) monkeypatch.setattr(_jobs, "enqueue_push_intel", _spy_enqueue) @pytest.fixture(autouse=True) def _tmp_objects_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """Redirect object storage to a per-test temp directory. Prevents tests from writing to the real storage path and isolates object state between tests. autouse=True so every test gets a fresh, empty object store without needing to request the fixture. """ import musehub.storage.backends as _backends import musehub.services.musehub_wire as _wire_svc import musehub.api.routes.wire as _wire_route from musehub.config import settings obj_dir = str(tmp_path) + "/objects" os.makedirs(obj_dir, exist_ok=True) test_backend = _backends.LocalBackend(objects_dir=obj_dir) monkeypatch.setattr(_wire_svc, "get_backend", lambda: test_backend) monkeypatch.setattr(_wire_route, "get_backend", lambda: test_backend) # Keep settings.musehub_objects_dir in sync so disk_path containment # checks in get_object_content / get_blob_meta see the same root. monkeypatch.setattr(settings, "musehub_objects_dir", obj_dir) # Redirect the /releases StaticFiles mount to a temp dir so tests that # hit /releases/* don't fail because /data/releases doesn't exist locally. releases_dir = str(tmp_path) + "/releases" os.makedirs(releases_dir, exist_ok=True) from musehub.main import app as _app for _route in _app.routes: if getattr(_route, "name", None) == "releases": _static = _route.app # type: ignore[attr-defined] _static.directory = releases_dir _static.config_checked = False # force re-check with new dir break def pytest_configure(config: pytest.Config) -> None: """Ensure asyncio_mode is auto so async fixtures work (e.g. in Docker when pyproject not in cwd).""" if hasattr(config.option, "asyncio_mode") and config.option.asyncio_mode is None: config.option.asyncio_mode = "auto" # Suppress verbose library loggers that flood the test output with DEBUG lines. for name in ("httpcore", "httpx", "sqlalchemy", "asyncio", "faker"): logging.getLogger(name).setLevel(logging.WARNING) @pytest.fixture(autouse=True) def reset_rate_limiter() -> Generator[None, None, None]: """Reset in-memory rate-limit counters before every test. Without this, the shared MemoryStorage accumulates hits across all tests in a session. Auth endpoints cap at 20/minute; running 30+ auth tests back-to-back exhausts that budget and causes 429s for legitimate calls. """ limiter.reset() _failure_limiter._failures.clear() yield @pytest.fixture def anyio_backend() -> str: return "asyncio" _WIRE_CONTEXT = MSignContext( handle="test-user-wire", identity_id="wire-test-user-id", is_agent=False, is_admin=False, ) @pytest.fixture def wire_headers() -> Generator[dict[str, str], None, None]: """Override auth deps to inject a fake MSignContext for wire protocol tests.""" app.dependency_overrides[require_signed_request] = lambda: _WIRE_CONTEXT app.dependency_overrides[optional_signed_request] = lambda: _WIRE_CONTEXT yield { "Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack", } app.dependency_overrides.pop(require_signed_request, None) app.dependency_overrides.pop(optional_signed_request, None) @pytest.fixture(autouse=True) def _reset_variation_store() -> Generator[None, None, None]: """Reset the singleton VariationStore between tests to prevent cross-test pollution. Gracefully no-ops if the variation module has been removed (MuseHub extraction). """ yield try: from musehub.variation.storage.variation_store import reset_variation_store reset_variation_store() except ModuleNotFoundError: pass _TEST_DATABASE_URL = os.environ.get( "TEST_DATABASE_URL", "postgresql+asyncpg://musehub:musehub@localhost:5434/musehub_test", ) # Sync URL for psycopg2 — used by the session-scoped schema fixture. _TEST_DATABASE_URL_SYNC = _TEST_DATABASE_URL.replace("+asyncpg", "") # Shared async engine for the whole test session (NullPool = no connection # reuse between tests, but engine object creation is cheap so we create it # once and share it). _TEST_ENGINE = create_async_engine(_TEST_DATABASE_URL, poolclass=NullPool) _TEST_SESSION_FACTORY = async_sessionmaker( bind=_TEST_ENGINE, class_=AsyncSession, expire_on_commit=False, ) # Pre-compute the TRUNCATE statement for all tables so we don't rebuild it # each test. Reversed sorted_tables respects FK dependency order. _TRUNCATE_SQL = "TRUNCATE {} RESTART IDENTITY CASCADE".format( ", ".join(t.name for t in reversed(Base.metadata.sorted_tables)) ) @pytest.fixture(scope="session", autouse=True) def _db_schema() -> Generator[None, None, None]: """Create the test schema once per test session using a sync psycopg2 engine. This replaces per-test drop_all/create_all (which took ~3 s per test on PostgreSQL) with a single DDL pass at session start and end. Individual tests get a clean slate via TRUNCATE in the db_session fixture instead. """ from sqlalchemy import create_engine as _create_engine from sqlalchemy import text as _text # connect_timeout=10: if postgres is unreachable or still starting (e.g. # Docker container not ready), fail fast instead of blocking in C forever. # Without this, Ctrl+C cannot kill the process because psycopg2's socket # read is a non-interruptible C-level call. sync_engine = _create_engine( _TEST_DATABASE_URL_SYNC, connect_args={"connect_timeout": 10}, ) # Terminate any leftover connections from interrupted test runs before # running drop_all. If a previous pytest session was killed with SIGQUIT # (Ctrl+\) it leaves postgres backends idle-in-transaction holding locks on # the test tables. drop_all then waits forever for those locks, which # makes the next test run freeze with Ctrl+C unresponsive. with sync_engine.connect() as _conn: _conn.execute(_text( "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " "WHERE datname = current_database() AND pid != pg_backend_pid()" )) _conn.commit() Base.metadata.drop_all(sync_engine) Base.metadata.create_all(sync_engine) sync_engine.dispose() yield sync_engine2 = _create_engine( _TEST_DATABASE_URL_SYNC, connect_args={"connect_timeout": 10}, ) Base.metadata.drop_all(sync_engine2) sync_engine2.dispose() @pytest_asyncio.fixture async def db_session(_db_schema: None) -> AsyncGenerator[AsyncSession, None]: """Provide a clean DB session for each test. Tables are truncated (not dropped/recreated) between tests — a single TRUNCATE … CASCADE is ~100× faster than drop_all + create_all on PostgreSQL, cutting per-test overhead from ~3 s to ~30 ms. """ from sqlalchemy import text as _text async with _TEST_ENGINE.begin() as conn: # Terminate any idle-in-transaction backends left by prior tests # (e.g. aborted SSE streams, asyncio.run() calls in migration tests) # before acquiring the lock needed for TRUNCATE. await conn.execute(_text( "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " "WHERE datname = current_database() AND pid != pg_backend_pid() " "AND state = 'idle in transaction'" )) await conn.execute(_text(_TRUNCATE_SQL)) old_engine = database._engine old_factory = database._async_session_factory database._engine = _TEST_ENGINE database._async_session_factory = _TEST_SESSION_FACTORY try: async with _TEST_SESSION_FACTORY() as session: async def override_get_db() -> AsyncGenerator[AsyncSession, None]: # Each request gets its own session so concurrent requests # (e.g. stress tests) don't share a single connection and # raise "concurrent operations are not permitted". # All test setup data is committed, so independent sessions # see it without needing to share the test session. async with _TEST_SESSION_FACTORY() as req_session: yield req_session app.dependency_overrides[get_db] = override_get_db yield session app.dependency_overrides.clear() finally: database._engine = old_engine database._async_session_factory = old_factory class _Asgi24Wrapper: """Inject spec_version='2.4' into every HTTP scope. Without this, Starlette's StreamingResponse (spec_version < 2.4 path) runs listen_for_disconnect concurrently with stream_response via anyio task_group. listen_for_disconnect calls receive() and steals the request body chunks before _AsyncExactReader can read them — causing a deadlock where both tasks block on response_complete.wait() waiting for each other. ASGI 2.4 tells Starlette to skip listen_for_disconnect and just stream the response directly, which is correct for our streaming push handler. """ def __init__(self, app: typing.Any) -> None: self._app = app async def __call__(self, scope: dict, receive: typing.Any, send: typing.Any) -> None: if scope.get("type") == "http": scope.setdefault("asgi", {})["spec_version"] = "2.4" await self._app(scope, receive, send) @pytest_asyncio.fixture async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]: """Create an async test client. Depends on db_session so auth revocation check uses test DB.""" transport = ASGITransport(app=_Asgi24Wrapper(app)) async with AsyncClient(transport=transport, base_url="http://test") as ac: yield ac # ----------------------------------------------------------------------------- # Auth fixtures for API contract and integration tests # Uses dependency_overrides to inject a fake MSignContext so tests don't need # real Ed25519 key pairs. Only active for tests that request auth_headers. # ----------------------------------------------------------------------------- _TEST_IDENTITY_ID = "550e8400-e29b-41d4-a716-446655440000" _TEST_HANDLE = "testuser" _TEST_CONTEXT = MSignContext( handle=_TEST_HANDLE, identity_id=_TEST_IDENTITY_ID, is_agent=False, is_admin=False, ) @pytest_asyncio.fixture async def test_user(db_session: AsyncSession) -> MusehubIdentity: """Create a test identity in the DB for authenticated route tests.""" identity = MusehubIdentity( identity_id=_TEST_IDENTITY_ID, handle=_TEST_HANDLE, display_name="Test User", identity_type="human", ) db_session.add(identity) await db_session.commit() await db_session.refresh(identity) # Close the autobegin transaction started by refresh() so subsequent # test-body commits don't hit "another operation is in progress". await db_session.commit() return identity @pytest.fixture def auth_headers(test_user: MusehubIdentity) -> Generator[dict[str, str], None, None]: """Override auth dependencies to inject a fake MSignContext for the test duration. Tests that need to verify 401 behaviour for *unauthenticated* requests should use a separate client call without passing ``auth_headers`` — note that while this fixture is active the app-level dep overrides are set globally, so any request made within the same test function will be treated as authenticated. Tests that need to distinguish authed/unauthed flows within one function should use ``app.dependency_overrides`` directly or split into two test functions. """ app.dependency_overrides[require_signed_request] = lambda: _TEST_CONTEXT app.dependency_overrides[optional_signed_request] = lambda: _TEST_CONTEXT yield {"Content-Type": "application/json"} app.dependency_overrides.pop(require_signed_request, None) app.dependency_overrides.pop(optional_signed_request, None)