gabriel / musehub public
conftest.py python
373 lines 14.9 KB
Raw
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ breaking 144 days ago
1 """Pytest configuration and fixtures."""
2 from __future__ import annotations
3
4 from pathlib import Path
5
6 import logging
7 import os
8 import typing
9 from collections.abc import AsyncGenerator, Generator
10
11 if not os.environ.get("MUSE_ENV"):
12 os.environ["MUSE_ENV"] = "test"
13
14 import pytest
15 import pytest_asyncio
16 from httpx import AsyncClient, ASGITransport
17 from sqlalchemy.ext.asyncio import (
18 AsyncSession,
19 async_sessionmaker,
20 create_async_engine,
21 )
22 from sqlalchemy.pool import NullPool
23
24 from musehub.core.genesis import compute_identity_id
25 from musehub.db import database
26 from musehub.db.database import Base, get_db
27 from musehub.db.musehub_models import MusehubIdentity
28 # Force all ORM models into Base.metadata before any create_all/drop_all.
29 # muse_cli_models is only imported inside init_db() in production; without
30 # this explicit import, Base.metadata is non-deterministic in tests (depends
31 # on import order), causing drop_all to miss tables that create_all later
32 # tries to create — resulting in duplicate-key errors on pg_type.
33 import musehub.db.muse_cli_models as _muse_cli_models # noqa: F401
34 from musehub.auth.request_signing import MSignContext, optional_signed_request, require_signed_request
35 from musehub.main import app
36 from musehub.rate_limits import limiter
37
38 type _JobPayload = dict[str, str | int | bool | None]
39 import musehub.auth.failure_limiter as _failure_limiter
40
41
42 @pytest.fixture(autouse=True)
43 def _stub_push_background_tasks(monkeypatch: pytest.MonkeyPatch) -> None:
44 """Replace enqueue_push_intel with a no-op spy during tests.
45
46 The push endpoint enqueues intel jobs into the DB. During tests we don't
47 want a live worker processing those jobs concurrently. This fixture
48 replaces enqueue_push_intel with a no-op that records calls in a
49 module-level list so integration tests can assert on what was enqueued
50 without touching the DB.
51 """
52 import musehub.services.musehub_jobs as _jobs
53
54 _jobs._test_enqueued_calls.clear()
55
56 async def _spy_enqueue(
57 session: AsyncSession, repo_id: str, head: str, domain_id: str | None = None
58 ) -> None:
59 _jobs._test_enqueued_calls.append((repo_id, "enqueue_push_intel", {"head": head, "domain_id": domain_id}))
60
61 monkeypatch.setattr(_jobs, "enqueue_push_intel", _spy_enqueue)
62
63
64 @pytest.fixture(autouse=True)
65 def _tmp_objects_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
66 """Redirect object storage to a per-test temp directory.
67
68 Prevents tests from writing to the real storage path and isolates
69 object state between tests. autouse=True so every test gets a
70 fresh, empty object store without needing to request the fixture.
71 """
72 import musehub.storage.backends as _backends
73 import musehub.services.musehub_wire as _wire_svc
74 import musehub.api.routes.wire as _wire_route
75 from musehub.config import settings
76
77 test_backend = _backends.LocalBackend()
78 monkeypatch.setattr(_wire_svc, "get_backend", lambda: test_backend)
79 monkeypatch.setattr(_wire_route, "get_backend", lambda: test_backend)
80 # Redirect musehub_repos_dir to tmp_path so disk_path containment
81 # checks in get_object_content / get_blob_meta see the right root.
82 monkeypatch.setattr(settings, "musehub_repos_dir", str(tmp_path))
83
84 # Redirect the /releases StaticFiles mount to a temp dir so tests that
85 # hit /releases/* don't fail because /data/releases doesn't exist locally.
86 releases_dir = f"{tmp_path}/releases"
87 os.makedirs(releases_dir, exist_ok=True)
88 from musehub.main import app as _app
89 for _route in _app.routes:
90 if getattr(_route, "name", None) == "releases":
91 _static = _route.app # type: ignore[attr-defined]
92 _static.directory = releases_dir
93 _static.config_checked = False # force re-check with new dir
94 break
95
96
97 def pytest_configure(config: pytest.Config) -> None:
98 """Ensure asyncio_mode is auto so async fixtures work (e.g. in Docker when pyproject not in cwd)."""
99 if hasattr(config.option, "asyncio_mode") and config.option.asyncio_mode is None:
100 config.option.asyncio_mode = "auto"
101 # Suppress verbose library loggers that flood the test output with DEBUG lines.
102 for name in ("httpcore", "httpx", "sqlalchemy", "asyncio", "faker"):
103 logging.getLogger(name).setLevel(logging.WARNING)
104
105
106 @pytest.fixture(autouse=True)
107 def reset_rate_limiter() -> Generator[None, None, None]:
108 """Reset in-memory rate-limit counters before every test.
109
110 Without this, the shared MemoryStorage accumulates hits across all tests
111 in a session. Auth endpoints cap at 20/minute; running 30+ auth tests
112 back-to-back exhausts that budget and causes 429s for legitimate calls.
113 """
114 limiter.reset()
115 _failure_limiter._failures.clear()
116 yield
117
118
119 @pytest.fixture
120 def anyio_backend() -> str:
121 return "asyncio"
122
123
124 _WIRE_CONTEXT = MSignContext(
125 handle="test-user-wire",
126 identity_id="wire-test-user-id",
127 is_agent=False,
128 is_admin=False,
129 )
130
131
132 @pytest.fixture
133 def wire_headers() -> Generator[dict[str, str], None, None]:
134 """Override auth deps to inject a fake MSignContext for wire protocol tests."""
135 app.dependency_overrides[require_signed_request] = lambda: _WIRE_CONTEXT
136 app.dependency_overrides[optional_signed_request] = lambda: _WIRE_CONTEXT
137 yield {
138 "Content-Type": "application/x-msgpack",
139 "Accept": "application/x-msgpack",
140 }
141 app.dependency_overrides.pop(require_signed_request, None)
142 app.dependency_overrides.pop(optional_signed_request, None)
143
144
145 @pytest.fixture(autouse=True)
146 def _reset_variation_store() -> Generator[None, None, None]:
147 """Reset the singleton VariationStore between tests to prevent cross-test pollution.
148
149 Gracefully no-ops if the variation module has been removed (MuseHub extraction).
150 """
151 yield
152 try:
153 from musehub.variation.storage.variation_store import reset_variation_store
154 reset_variation_store()
155 except ModuleNotFoundError:
156 pass
157
158
159 _TEST_DATABASE_URL = os.environ.get(
160 "TEST_DATABASE_URL",
161 "postgresql+asyncpg://musehub:musehub@localhost:5434/musehub_test",
162 )
163
164 # Sync URL for psycopg2 — used by the session-scoped schema fixture.
165 _TEST_DATABASE_URL_SYNC = _TEST_DATABASE_URL.replace("+asyncpg", "")
166
167 # Shared async engine for the whole test session (NullPool = no connection
168 # reuse between tests, but engine object creation is cheap so we create it
169 # once and share it).
170 _TEST_ENGINE = create_async_engine(_TEST_DATABASE_URL, poolclass=NullPool)
171 _TEST_SESSION_FACTORY = async_sessionmaker(
172 bind=_TEST_ENGINE,
173 class_=AsyncSession,
174 expire_on_commit=False,
175 )
176
177 # Pre-compute the TRUNCATE statement for all tables so we don't rebuild it
178 # each test. Reversed sorted_tables respects FK dependency order.
179 _TRUNCATE_SQL = "TRUNCATE {} RESTART IDENTITY CASCADE".format(
180 ", ".join(t.name for t in reversed(Base.metadata.sorted_tables))
181 )
182
183
184 @pytest.fixture(scope="session", autouse=True)
185 def _db_schema() -> Generator[None, None, None]:
186 """Create the test schema once per test session using a sync psycopg2 engine.
187
188 This replaces per-test drop_all/create_all (which took ~3 s per test on
189 PostgreSQL) with a single DDL pass at session start and end. Individual
190 tests get a clean slate via TRUNCATE in the db_session fixture instead.
191 """
192 from sqlalchemy import create_engine as _create_engine
193
194 from sqlalchemy import text as _text
195
196 # connect_timeout=10: if postgres is unreachable or still starting (e.g.
197 # Docker container not ready), fail fast instead of blocking in C forever.
198 # Without this, Ctrl+C cannot kill the process because psycopg2's socket
199 # read is a non-interruptible C-level call.
200 sync_engine = _create_engine(
201 _TEST_DATABASE_URL_SYNC,
202 connect_args={"connect_timeout": 10},
203 )
204 # Terminate any leftover connections from interrupted test runs before
205 # running drop_all. If a previous pytest session was killed with SIGQUIT
206 # (Ctrl+\) it leaves postgres backends idle-in-transaction holding locks on
207 # the test tables. drop_all then waits forever for those locks, which
208 # makes the next test run freeze with Ctrl+C unresponsive.
209 with sync_engine.connect() as _conn:
210 _conn.execute(_text(
211 "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
212 "WHERE datname = current_database() AND pid != pg_backend_pid()"
213 ))
214 _conn.commit()
215 # Dispose so drop_all / create_all get fresh connections — the
216 # pg_terminate_backend above may have killed pooled connections.
217 sync_engine.dispose()
218 sync_engine2 = _create_engine(
219 _TEST_DATABASE_URL_SYNC,
220 connect_args={"connect_timeout": 10},
221 )
222 Base.metadata.drop_all(sync_engine2)
223 sync_engine2.dispose()
224 sync_engine2 = _create_engine(
225 _TEST_DATABASE_URL_SYNC,
226 connect_args={"connect_timeout": 10},
227 )
228 Base.metadata.create_all(sync_engine2)
229 sync_engine2.dispose()
230 yield
231 sync_engine3 = _create_engine(
232 _TEST_DATABASE_URL_SYNC,
233 connect_args={"connect_timeout": 10},
234 )
235 with sync_engine3.connect() as _conn:
236 _conn.execute(_text(
237 "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
238 "WHERE datname = current_database() AND pid != pg_backend_pid()"
239 ))
240 _conn.commit()
241 sync_engine3.dispose()
242 sync_engine4 = _create_engine(
243 _TEST_DATABASE_URL_SYNC,
244 connect_args={"connect_timeout": 10},
245 )
246 Base.metadata.drop_all(sync_engine4)
247 sync_engine4.dispose()
248
249
250 @pytest_asyncio.fixture
251 async def db_session(_db_schema: None) -> AsyncGenerator[AsyncSession, None]:
252 """Provide a clean DB session for each test.
253
254 Tables are truncated (not dropped/recreated) between tests — a single
255 TRUNCATE … CASCADE is ~100× faster than drop_all + create_all on
256 PostgreSQL, cutting per-test overhead from ~3 s to ~30 ms.
257 """
258 from sqlalchemy import text as _text
259
260 async with _TEST_ENGINE.begin() as conn:
261 # Terminate ALL other backends before TRUNCATE. A failed test can
262 # leave a connection in any state (idle in transaction, idle in
263 # transaction (aborted), active) — filtering by state misses some
264 # cases and causes deadlocks when TRUNCATE races the stale transaction.
265 await conn.execute(_text(
266 "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
267 "WHERE datname = current_database() AND pid != pg_backend_pid()"
268 ))
269 await conn.execute(_text(_TRUNCATE_SQL))
270
271 old_engine = database._engine
272 old_factory = database._async_session_factory
273 database._engine = _TEST_ENGINE
274 database._async_session_factory = _TEST_SESSION_FACTORY
275 try:
276 async with _TEST_SESSION_FACTORY() as session:
277 async def override_get_db() -> AsyncGenerator[AsyncSession, None]:
278 # Each request gets its own session so concurrent requests
279 # (e.g. stress tests) don't share a single connection and
280 # raise "concurrent operations are not permitted".
281 # All test setup data is committed, so independent sessions
282 # see it without needing to share the test session.
283 async with _TEST_SESSION_FACTORY() as req_session:
284 yield req_session
285 app.dependency_overrides[get_db] = override_get_db
286 yield session
287 app.dependency_overrides.clear()
288 finally:
289 database._engine = old_engine
290 database._async_session_factory = old_factory
291
292
293 class _Asgi24Wrapper:
294 """Inject spec_version='2.4' into every HTTP scope.
295
296 Without this, Starlette's StreamingResponse (spec_version < 2.4 path) runs
297 listen_for_disconnect concurrently with stream_response via anyio task_group.
298 listen_for_disconnect calls receive() and steals the request body chunks
299 before _AsyncExactReader can read them — causing a deadlock where both tasks
300 block on response_complete.wait() waiting for each other.
301
302 ASGI 2.4 tells Starlette to skip listen_for_disconnect and just stream the
303 response directly, which is correct for our streaming push handler.
304 """
305
306 def __init__(self, app: typing.Any) -> None:
307 self._app = app
308
309 async def __call__(self, scope: typing.MutableMapping[str, typing.Any], receive: typing.Any, send: typing.Any) -> None:
310 if scope.get("type") == "http":
311 scope.setdefault("asgi", {})["spec_version"] = "2.4"
312 await self._app(scope, receive, send)
313
314
315 @pytest_asyncio.fixture
316 async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
317 """Create an async test client. Depends on db_session so auth revocation check uses test DB."""
318 transport = ASGITransport(app=_Asgi24Wrapper(app))
319 async with AsyncClient(transport=transport, base_url="http://test") as ac:
320 yield ac
321
322
323 # -----------------------------------------------------------------------------
324 # Auth fixtures for API contract and integration tests
325 # Uses dependency_overrides to inject a fake MSignContext so tests don't need
326 # real Ed25519 key pairs. Only active for tests that request auth_headers.
327 # -----------------------------------------------------------------------------
328
329 _TEST_IDENTITY_ID = compute_identity_id(b"testuser")
330 _TEST_HANDLE = "testuser"
331
332 _TEST_CONTEXT = MSignContext(
333 handle=_TEST_HANDLE,
334 identity_id=_TEST_IDENTITY_ID,
335 is_agent=False,
336 is_admin=False,
337 )
338
339
340 @pytest_asyncio.fixture
341 async def test_user(db_session: AsyncSession) -> MusehubIdentity:
342 """Create a test identity in the DB for authenticated route tests."""
343 identity = MusehubIdentity(
344 identity_id=_TEST_IDENTITY_ID,
345 handle=_TEST_HANDLE,
346 display_name="Test User",
347 identity_type="human",
348 )
349 db_session.add(identity)
350 await db_session.commit()
351 await db_session.refresh(identity)
352 # Close the autobegin transaction started by refresh() so subsequent
353 # test-body commits don't hit "another operation is in progress".
354 await db_session.commit()
355 return identity
356
357
358 @pytest.fixture
359 def auth_headers(test_user: MusehubIdentity) -> Generator[dict[str, str], None, None]:
360 """Override auth dependencies to inject a fake MSignContext for the test duration.
361
362 Tests that need to verify 401 behaviour for *unauthenticated* requests should
363 use a separate client call without passing ``auth_headers`` — note that while
364 this fixture is active the app-level dep overrides are set globally, so any
365 request made within the same test function will be treated as authenticated.
366 Tests that need to distinguish authed/unauthed flows within one function should
367 use ``app.dependency_overrides`` directly or split into two test functions.
368 """
369 app.dependency_overrides[require_signed_request] = lambda: _TEST_CONTEXT
370 app.dependency_overrides[optional_signed_request] = lambda: _TEST_CONTEXT
371 yield {"Content-Type": "application/json"}
372 app.dependency_overrides.pop(require_signed_request, None)
373 app.dependency_overrides.pop(optional_signed_request, None)
File History 1 commit
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ 144 days ago