import pytest import os from dotenv import load_dotenv from database_test_guard import validate_test_database_url as _validate_test_database_url # Load .env file to ensure TEST_DATABASE_URL is available load_dotenv() # CRITICAL: Override DATABASE_URL with TEST_DATABASE_URL for all tests # This prevents any accidental production database access test_db_url = _validate_test_database_url(os.getenv('TEST_DATABASE_URL')) os.environ['DATABASE_URL'] = test_db_url from oneirodex import create_app, db def _install_lock_timeout(): """Make a blocked test connection fail fast and say so. Postgres waits for a lock forever by default. Combined with the schema setup below, that turned one leaked transaction into a silent, permanent stall: no failing test, no timeout, no output — just a pytest process at 0% CPU that looked "slow" for hours. A short `lock_timeout` converts that into an immediate, named error on the statement that could not get its lock, which is a bug report instead of a mystery. Applied only to test connections; production waits as before, where blocking on a real lock is usually the correct behaviour. """ from sqlalchemy import event from sqlalchemy.engine import Engine @event.listens_for(Engine, 'connect') def _set_timeouts(dbapi_connection, _record): # noqa: ANN001 # In autocommit, because `SET` in Postgres is transactional: run inside # the implicit transaction and SQLAlchemy's first rollback (which # happens whenever a connection returns to the pool) quietly undoes it, # leaving exactly the silent hang this is meant to prevent. try: previous = dbapi_connection.autocommit dbapi_connection.autocommit = True try: with dbapi_connection.cursor() as cur: # `lock_timeout` stays short: no healthy test waits on a # lock, so a 15s wait is always a leak somewhere and the # named `LockNotAvailable` on the blocked statement is the # bug report we want. cur.execute("SET SESSION lock_timeout = '15s'") # `idle_in_transaction_session_timeout` is deliberately # generous. `db_session` now holds one outer transaction # open for the whole test (per-test SAVEPOINT isolation), # so between statements the connection is legitimately # "idle in transaction" for as long as the test's own # Python runs — and the slowest single tests # (`test_init_manager_themes`, ~75s of theme-file copying) # would trip a 60s ceiling and have Postgres sever their # own transaction mid-test. 300s clears the slowest test # with headroom while still killing a transaction a # crashed fixture stranded between tests (teardown always # rolls the outer transaction back, so a leak cannot # outlive one test otherwise). cur.execute("SET SESSION idle_in_transaction_session_timeout = '300s'") finally: dbapi_connection.autocommit = previous except Exception: # noqa: BLE001 # Non-Postgres or a driver without autocommit — nothing to set. pass _install_lock_timeout() @pytest.fixture(scope='function') def app(): """Create and configure a test app using the test database.""" # Ensure we have TEST_DATABASE_URL environment variable test_db_url = _validate_test_database_url(os.getenv('TEST_DATABASE_URL')) # Additional check: ensure DATABASE_URL was properly overridden current_db_url = os.getenv('DATABASE_URL') if current_db_url != test_db_url: pytest.fail( f"CRITICAL: DATABASE_URL override failed. " "This could result in tests running against production database." ) # Create app. Under pytest, create_app() auto-selects config.TestConfig, # which carries TESTING, WTF_CSRF_ENABLED=False, SERVER_NAME='localhost', # APPLICATION_ROOT, PREFERRED_URL_SCHEME, a stable SECRET_KEY, and the # TEST_DATABASE_URL URI — the post-construction instance mutations that # used to live here (wave A2.6). app = create_app() # Double-check that the app is using test database actual_db_uri = app.config['SQLALCHEMY_DATABASE_URI'] if actual_db_uri != test_db_url: pytest.fail( f"CRITICAL: App database URI mismatch. " "Tests cannot proceed with wrong database configuration." ) yield app # --------------------------------------------------------------------------- # Per-test database isolation (spec §3.2) # --------------------------------------------------------------------------- # # The schema (DDL) is still built exactly once per process, behind the # `_SCHEMA_READY` latch. `add_column_if_not_exists` issues a long series of # `ALTER TABLE`s, each wanting an ACCESS EXCLUSIVE lock; doing that once and # only once is what keeps the suite from the lock-storm deadlock recorded in # `docs/dev/archive/test-harness-2026-08-07.md`. # # What changed: `db_session` no longer hands out `db.session` directly with no # teardown. Each test now runs inside a SAVEPOINT nested in an outer # transaction that is *always rolled back* on teardown — the SQLAlchemy 2.0 # "joining a session into an external transaction" pattern, with # `join_transaction_mode="create_savepoint"` so a `db.session.commit()` inside # a test (or a product-code `begin_nested()`, e.g. # `global_settings_row_or_create`, `metadata_enrichment`, `game_enrich`) # releases and re-opens a savepoint instead of touching the real transaction. # # Net: every test starts from an empty database and leaves nothing behind. The # old per-run `TRUNCATE` is gone as an isolation mechanism; a single # process-start `TRUNCATE` remains only to sweep rows an *older* harness # committed before this fixture existed, so a local `oneirodextest` that still # carries months of residue starts where CI (fresh container) starts. _SCHEMA_READY = False def _build_schema_once(): """DDL + one residue sweep, on a dedicated connection, exactly once.""" global _SCHEMA_READY if _SCHEMA_READY: return from sqlalchemy import text connection = db.engine.connect() try: with connection.begin(): db.metadata.create_all(bind=connection) # Incremental column adds for a test DB left over from an older schema. # Idempotent, and now paid for once per process rather than per test. from oneirodex.updateschema import DatabaseManager DatabaseManager().add_column_if_not_exists() # One-time floor sweep. Per-test isolation (below) keeps the database # clean going forward, but it cannot un-commit rows a previous harness # already persisted. `ONEIRODEX_KEEP_TEST_DATA=1` skips it for when the # leftover rows are the thing under investigation. if os.getenv('ONEIRODEX_KEEP_TEST_DATA') != '1': tables = [f'"{t.name}"' for t in db.metadata.sorted_tables] if tables: with connection.begin(): connection.execute( text( 'TRUNCATE TABLE ' + ', '.join(tables) + ' RESTART IDENTITY CASCADE' ) ) finally: connection.close() _SCHEMA_READY = True @pytest.fixture(scope='function') def db_session(app): """A database session scoped to one test by a rolled-back SAVEPOINT. Everything a test writes — directly or through the Flask test client, which shares this bound connection — is visible for the life of the test and discarded on teardown. `db.session` is swapped for a scoped session bound to the test's connection because Flask-SQLAlchemy's own `Session.get_bind` returns the app's default engine before it will honour a per-session `bind=`, so `db.session.configure(bind=...)` alone does not route here. Two things keep the SAVEPOINT model honest: * **The savepoint-restart listener** (SQLAlchemy's "Joining a Session into an External Transaction" recipe). A `db.session.commit()` inside a test — or Flask-SQLAlchemy's `teardown_appcontext` firing `db.session.remove()` when a nested `app_context()` / `test_request_context()` pops mid-test — RELEASEs the per-test SAVEPOINT. With nothing re-opening one, the session keeps a dangling savepoint reference and teardown's `.remove()` emits `ROLLBACK TO SAVEPOINT` against a name Postgres has already released (`InvalidSavepointSpecification`). `_restart_savepoint` re-arms a fresh nested transaction the moment the previous one ends, so exactly one per-test SAVEPOINT is always outstanding. * **Exception-safe teardown.** Even with the listener, a slow test that let Postgres hit `idle_in_transaction_session_timeout`, or any error inside `.remove()`, must not skip `transaction.rollback()` / `connection.close()`: an un-rolled-back outer transaction strands the connection `idle in transaction` holding row locks, and the next test's first write blocks on `lock_timeout` and fails — the cascade recorded in `docs/dev/test-harness-2026-09-10.md`. Every teardown step therefore runs under its own `try`, and the connection is closed no matter what. """ from sqlalchemy import event from sqlalchemy import orm as sa_orm from flask_sqlalchemy.session import _app_ctx_id with app.app_context(): _build_schema_once() connection = db.engine.connect() transaction = connection.begin() # outer transaction — always rolled back original_session = db.session test_session = sa_orm.scoped_session( sa_orm.sessionmaker( bind=connection, join_transaction_mode='create_savepoint', future=True, # No `expire_on_commit=False`. Keeping objects unexpired after a # `db_session.commit()` hands the test the *pre-commit* Python # values — a tz-aware datetime the column stores naive, a # relationship a concurrent writer has since changed — instead # of what a fresh query returns. Flask-SQLAlchemy's own session # expires on commit; matching that keeps a test reading the # database, not its own stale identity map. ), scopefunc=_app_ctx_id, ) db.session = test_session # The session for *this* (the fixture's) app context. The per-test # SAVEPOINT and the restart listener both belong to it alone. A Flask # test-client request pushes its own app context, so `scopefunc= # _app_ctx_id` hands it a *separate* short-lived session bound to the # same connection; that one manages its own savepoints, is `.remove()`d # when the request context pops, and must NOT get the restart listener # (its product-code `with db.session.begin_nested():` blocks are often # the only nested transaction it has, and re-arming inside their # `__exit__` raises "Can't operate on closed transaction inside context # manager"). bound_session = test_session() bound_session.begin_nested() def _restart_savepoint(sess, trans): # Re-open the per-test SAVEPOINT once the previous one has fully # ended — a `db.session.commit()` in a test releases it and lands # here with `in_nested_transaction()` False. A product-code # `with bound_session.begin_nested():` in a fixture also fires this # event from its `__exit__`, but the per-test SAVEPOINT is still # outstanding under it, so `in_nested_transaction()` is True and it # is left alone. (`not trans._parent.nested` cannot be the guard: # `join_transaction_mode='create_savepoint'` makes the root join # itself report `nested`.) if trans.nested and not sess.in_nested_transaction(): sess.begin_nested() event.listen(bound_session, 'after_transaction_end', _restart_savepoint) try: yield test_session finally: try: event.remove( bound_session, 'after_transaction_end', _restart_savepoint ) except Exception: # noqa: BLE001 pass try: test_session.remove() except Exception: # noqa: BLE001 # Most likely `InvalidSavepointSpecification` if Postgres # already severed the transaction. The outer rollback below # still has to run. pass finally: db.session = original_session try: transaction.rollback() except Exception: # noqa: BLE001 pass try: connection.close() except Exception: # noqa: BLE001 pass # --------------------------------------------------------------------------- # Auto-markers (pytest.ini declares `database` and `slow`; --strict-markers is on) # --------------------------------------------------------------------------- # Conservative `slow` list: only tests whose *own* work — filesystem theme # copies, redirect-chain walks, mocked-network retry loops — pushed them past # ~20s of call time in the 2026-09-10 re-baseline (`--durations`). Not derived # from timings at collection time (those move with the box); revisit from a # fresh `--durations` when the set visibly drifts. _SLOW_NODEIDS = ( 'test_init_manager_themes.py::TestSetupDefaultTheme', 'test_init_manager_themes.py::test_install_preset_themes_uses_the_shipped_source', 'test_ssrf_hardening.py::test_redirect_chain_is_bounded', 'test_ssrf_hardening.py::test_post_body_is_not_replayed_on_a_303', 'test_ssrf_hardening.py::test_relative_redirect_resolves_against_current_hop', 'test_providers_steamgriddb.py::test_fetch_image_mocked', 'test_gaming_news_feeds.py::TestFeedApi::test_response_lists_the_configured_sources', 'test_indexer_registry.py::test_enable_presets_copies_without_mutating_pack', ) def pytest_collection_modifyitems(config, items): """Tag every test that reaches Postgres `database`, and the known-slow ones `slow`. `database` is derived, not hand-maintained: a test gets it whenever `db_session` is in its resolved fixture graph — directly, or through `configured_install` / `global_settings` / `admin_user` / any fixture that itself requests `db_session`. `-m "not database"` then selects the pure unit tests, and `-m database` the ones that need `oneirodex-review-db` up. """ for item in items: if 'db_session' in getattr(item, 'fixturenames', ()): # noqa: SIM118 item.add_marker('database') if any(frag in item.nodeid for frag in _SLOW_NODEIDS): item.add_marker('slow') @pytest.fixture(scope='function') def client(app): """Create a test client for the Flask application.""" return app.test_client() # --------------------------------------------------------------------------- # Bucket-E opt-out: no_savepoint_db (docs/dev/test-harness-2026-09-10.md) # --------------------------------------------------------------------------- # # A handful of tests need `db.session` to behave exactly as it does in # production — an ordinary Flask-SQLAlchemy scoped session pulling a fresh # connection per app context from the engine's pool — because they exercise a # second OS thread pushing its own `app.app_context()` # (`run_in_background`, `oneirodex/utils/background.py`) or a debounced digest # flushed from a context nested inside another. `db_session` binds every app # context onto ONE shared `Connection` object so a test-client request can see # a test's own uncommitted rows; a second thread touching that same # `Connection` concurrently is exactly the hazard `run_in_background` exists # to avoid (a DBAPI connection is not thread-safe), and a nested context's own # `teardown_appcontext` -> `.remove()` rolls back its `create_savepoint` # savepoint, discarding what an even-more-nested context just committed into # it. Neither is fixable with a smarter SAVEPOINT — these tests need a real # session against the real engine. # # Rows committed here are real, so there is no rollback to lean on. Cleanup is # a TRUNCATE scoped to the fixed, small set of tables the bucket-E tests # actually touch — never a global, cross-test TRUNCATE. That is safe because # these are the only tests that request this fixture and pytest runs them one # at a time in one process: nothing else can be relying on those tables' # contents while one is mid-flight, so a TRUNCATE that also cascades into a # handful of unlisted-but-referencing tables (games -> tags, etc.) removes # only rows this test itself created. _NO_SAVEPOINT_TABLES = ( 'user_notifications', 'scan_jobs', 'games', 'libraries', 'global_settings', 'users', ) @pytest.fixture(scope='function') def no_savepoint_db(app): """Opt-out of the per-test SAVEPOINT model for bucket-E tests only. Use only for a test that pushes its own nested or cross-thread app context and needs `db.session` to be the app's ordinary engine-bound session, not `db_session`'s single shared connection — see the module comment above and docs/dev/test-harness-2026-09-10.md. Everything the test commits is real; teardown truncates `_NO_SAVEPOINT_TABLES` so the next (SAVEPOINT-isolated) test still starts from an empty database. """ from sqlalchemy import text with app.app_context(): _build_schema_once() try: yield db.session finally: try: db.session.remove() except Exception: # noqa: BLE001 pass cleanup = db.engine.connect() try: with cleanup.begin(): cleanup.execute( text( 'TRUNCATE TABLE ' + ', '.join(f'"{t}"' for t in _NO_SAVEPOINT_TABLES) + ' RESTART IDENTITY CASCADE' ) ) finally: cleanup.close() @pytest.fixture(scope='function') def configured_install(db_session): """An install that is past the setup wizard. `check_setup_status` is a `before_request` hook and `is_setup_required()` means "no users exist", so on the now-empty per-test database *any* anonymous request is redirected to `/setup` — not to the login page. Every `..._requires_login` test that asserts `'/login' in response.location` needs a user row to exist first; requesting this fixture states that precondition explicitly instead of relying on an earlier file to have left one behind. Tests that drive the wizard itself must *not* use it — they need the un-configured state this removes. The insert lives inside the test's SAVEPOINT and is rolled back on teardown. """ from uuid import uuid4 from oneirodex.models import User from oneirodex.utils.global_settings import global_settings_row_or_create anchor = User( user_id=str(uuid4()), name=f'SetupAnchor_{str(uuid4())[:8]}', email=f'anchor_{str(uuid4())[:8]}@test.com', role='admin', is_email_verified=True, ) anchor.set_password('testpass123') db_session.add(anchor) settings = global_settings_row_or_create() settings.setup_in_progress = False settings.setup_completed = True db_session.commit() return settings @pytest.fixture(scope='function') def global_settings(db_session): """The `GlobalSettings` singleton for this test. Each test starts with an empty database, so this is an unconditional create — but routed through `global_settings_row_or_create()` (the same helper production code uses) so it composes with `configured_install` and with any product code that also lazily creates the row, without tripping the `global_settings_singleton` unique index. Mutate the returned object and commit for tests that need particular values on it. """ from oneirodex.utils.global_settings import global_settings_row_or_create row = global_settings_row_or_create() db_session.commit() return row