From 6249f5874da1e6ff3cd6abf01ece7ca62080f4b0 Mon Sep 17 00:00:00 2001 From: cdeust Date: Thu, 30 Jul 2026 20:37:24 +0200 Subject: [PATCH] refactor(tests): split conftest.py under the 300-line cap (boy-scout, issue #276/#287) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tests_py/conftest.py was 501 lines — over both the repo's local 300-line cap and the global 500-line cap (coding-standards.md §4.1) — discovered while fixing #287 (a small, additive-only touch there would have left the pre-existing violation unaddressed and unfiled, which §14 forbids without a citing PR; this is that PR). Split along the file's own pre-existing section-comment boundaries, into three new modules with zero behavior change — closures over conftest.py's module globals became explicit-parameter functions (an incidental DI improvement, not the goal): - tests_py/_pg_throwaway_db.py — per-process throwaway PG database creation/cleanup (create_isolated_test_database, drop_isolated_database, drop_dead_orphaned_databases, maintenance_url). - tests_py/_pg_safety_guards.py — the two fail-closed data-safety guards (guard_against_populated_db, the anti-537k-row-deletion guard from incident 2026-06-10; guard_against_real_data_roots, from issue #219) plus looks_like_test_db and _resolved_real_data_roots. - tests_py/_domain_mapping_isolation.py — the dev-root filesystem-scan isolation from issue #196. conftest.py itself drops from 501 to 281 lines: the pytest hooks (pytest_sessionfinish, the autouse _test_isolation fixture) and the module-level bootstrap sequence stay in place, now delegating to the extracted functions in the exact same order. Two existing regression-guard test files hard-coded the guard's prior location/signature (scanning conftest.py's raw source, or calling a zero-arg closure) and broke on the move — fixed to import the real, now-standalone modules directly, which also let them drop the old exec-from-source-text workaround entirely. Both, plus the new tests added for the extracted modules, then pushed test_conftest_guard.py (496 lines pre-existing) and test_store_isolation.py (350 pre-existing) over the same cap; split each along its natural per-function boundary rather than leave a second freshly-introduced violation: test_conftest_guard.py -> test_looks_like_test_db.py + test_guard_against_populated_db.py + test_guard_blocks_populated_db.py + test_guard_against_populated_db_structural.py test_store_isolation.py -> test_store_isolation.py (trimmed) + test_real_data_root_guard.py Mutation-tested (scoped, CORTEX_TEST_ISOLATE_DB=0 to keep the per-process throwaway-DB bootstrap itself out of the mutated call graph): 204/211 mutants killed; 7 survivors are provable-equivalent split/rsplit/maxsplit choices (documented at each use site, verified by construction and by property fuzzing — one is domain-scoped to the exact cortex_test_pw_ format the function is only ever called against, not a general claim). Full local suite: 6854 passed, 5 skipped, 116 subtests passed — no regressions. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012Yu6EnWspTfqHoGkExyS6u --- tests_py/_domain_mapping_isolation.py | 61 +++ tests_py/_pg_safety_guards.py | 125 +++++ tests_py/_pg_throwaway_db.py | 146 ++++++ tests_py/conftest.py | 264 +--------- tests_py/invariants/test_conftest_guard.py | 496 ------------------ .../test_domain_mapping_isolation.py | 42 ++ .../test_guard_against_populated_db.py | 270 ++++++++++ ...t_guard_against_populated_db_structural.py | 95 ++++ .../test_guard_blocks_populated_db.py | 160 ++++++ .../invariants/test_looks_like_test_db.py | 65 +++ tests_py/invariants/test_pg_throwaway_db.py | 289 ++++++++++ .../invariants/test_real_data_root_guard.py | 213 ++++++++ tests_py/invariants/test_store_isolation.py | 183 ++----- 13 files changed, 1525 insertions(+), 884 deletions(-) create mode 100644 tests_py/_domain_mapping_isolation.py create mode 100644 tests_py/_pg_safety_guards.py create mode 100644 tests_py/_pg_throwaway_db.py delete mode 100644 tests_py/invariants/test_conftest_guard.py create mode 100644 tests_py/invariants/test_domain_mapping_isolation.py create mode 100644 tests_py/invariants/test_guard_against_populated_db.py create mode 100644 tests_py/invariants/test_guard_against_populated_db_structural.py create mode 100644 tests_py/invariants/test_guard_blocks_populated_db.py create mode 100644 tests_py/invariants/test_looks_like_test_db.py create mode 100644 tests_py/invariants/test_pg_throwaway_db.py create mode 100644 tests_py/invariants/test_real_data_root_guard.py diff --git a/tests_py/_domain_mapping_isolation.py b/tests_py/_domain_mapping_isolation.py new file mode 100644 index 00000000..263a25db --- /dev/null +++ b/tests_py/_domain_mapping_isolation.py @@ -0,0 +1,61 @@ +"""Isolate `shared.domain_mapping`'s dev-root scan from the real filesystem. + +Extracted from `tests_py/conftest.py` (issue #276/#287 boy-scout follow-up +— see `tests_py/_pg_throwaway_db.py`'s docstring for the size-cap +rationale this split serves). No behavior change: `conftest.py` calls +`isolate_dev_root_scan()` at the exact point in its module-level sequence +where this code used to run inline. + +ROOT CAUSE (reproduced deterministically while measuring coverage for +issue #196, 2026-07-27): `shared/domain_mapping._build_registry()` (used +transitively by `handlers/consolidation/wiki_backlog_pass.py` -> +`core/wiki_drift.py`'s `audit_wiki_drift` -> `_project_source_root` -> +`_file_exists_under`'s `os.walk` fallback) has NO test override by +default. `_candidate_dev_roots()` falls back to `$HOME/Developments`, +`$HOME/Documents/Developments`, `$HOME/dev`, `$HOME/code` — real developer +directories — and (this is the part that defeats a naive fix) is +ADDITIVE: even with `$CORTEX_DEV_ROOT` set, every one of those defaults +that exists on disk is STILL appended as an extra candidate (see +`shared/domain_mapping.py::_candidate_dev_roots` — the env var is one more +candidate, not a replacement). Setting the env var alone does NOT isolate +the scan. On a machine where one of the defaults exists and holds real git +repos (any contributor's normal dev machine), any test that exercises the +full consolidate handler (e.g. +`test_consolidate_stage_exception_does_not_crash_siblings`) transitively +walks those REAL, unrelated, potentially enormous repo trees via +`os.walk` — observed killing a `pytest --cov` run at the 300s per-test +timeout (`pyproject.toml`'s `timeout = 300`). This is invisible in CI +because none of the candidate dev roots exist on a CI runner +(`$HOME=/home/runner`), so `_candidate_dev_roots()` returns `[]` there and +`_project_source_root` always returns `None` — CI's isolation is +accidental, not designed. This plausibly explains the previously +unexplained "CI stall on 148d5a1... hung >3h... no offender identified" +that same `timeout` setting was added to catch: a local run with a +populated dev root would reproduce this hang with no per-test timeout to +bound it. + +Fix: replace `_candidate_dev_roots` itself (not just the env var) so it +always returns an empty list for the whole test session, matching the +exact isolation pattern `tests_py/shared/test_domain_mapping.py` already +uses per-test (monkeypatch `_candidate_dev_roots` directly, then +`_build_registry.cache_clear()`). Those dedicated tests monkeypatch this +same attribute again inside their own test body — monkeypatch always +restores to whatever was in place when the test started, so patching it +here at collection time (not via the `monkeypatch` fixture, which only +exists inside a running test) is a permanent module-level replacement for +the whole session; the per-test monkeypatch calls inside +`test_domain_mapping.py` override it for the duration of those specific +tests and their own `_build_registry.cache_clear()` calls put the real +cache state back in sync afterward. +""" + +from __future__ import annotations + + +def isolate_dev_root_scan() -> None: + """Permanently replace `_candidate_dev_roots` with an empty-list stub + for the whole test session (see module docstring for why).""" + from mcp_server.shared import domain_mapping as dm + + dm._candidate_dev_roots = lambda: [] + dm._build_registry.cache_clear() diff --git a/tests_py/_pg_safety_guards.py b/tests_py/_pg_safety_guards.py new file mode 100644 index 00000000..65707446 --- /dev/null +++ b/tests_py/_pg_safety_guards.py @@ -0,0 +1,125 @@ +"""Fail-closed guards against running the destructive test suite against a +real database or a real `~/.claude` tree. + +Extracted from `tests_py/conftest.py` (issue #276/#287 boy-scout follow-up) +as part of bringing that file under the repo's 300-line cap +(coding-standards.md §4.1) — see `tests_py/_pg_throwaway_db.py`'s docstring +for the full rationale. Behavior is unchanged: both guards below still +call `pytest.exit(..., returncode=2)` themselves (that call does not need +to happen inside a named pytest hook — conftest.py invokes these as plain +functions at module-import time, exactly as it invoked the code before the +extraction), with the same messages and the same conditions. The only +change is that they take their inputs as explicit parameters instead of +reading `tests_py/conftest.py`'s module globals directly — an incidental +DI improvement (these are independently testable now), not the goal. +""" + +from __future__ import annotations + +import os + +import pytest + + +def looks_like_test_db(url: str) -> bool: + """True when the database NAME declares itself a test/bench database. + + `.rsplit("/", 1)[-1]` and `.split("?", 1)[0]` are both selected for + their FIRST/LAST-element indexing, not their maxsplit count: `[-1]` + from `rsplit(sep, N)` is the same for every N>=1 or unlimited (it is + always "everything after the rightmost separator"), and `[0]` from + `split(sep, N)` is likewise invariant to N (always "everything before + the leftmost separator"). Only swapping `rsplit` for `split` (or vice + versa) on the SAME call changes the result, and only when 2+ + occurrences of that separator are present. + """ + name = url.rsplit("/", 1)[-1].split("?", 1)[0] + return name.endswith("_test") or name.endswith("_bench") or "_test_" in name + + +def guard_against_populated_db(test_db_url: str) -> None: + """Refuse to run destructive test isolation against a populated DB. + + INCIDENT 2026-06-10: an agent ran ``DATABASE_URL= CI=true + pytest`` — the CI branch trusted the URL verbatim and + ``_clean_all_tables`` deleted 537,396 production memories (plus all + entities, relationships, and prospective triggers) between tests. + + Two-tier guard, fail-closed: + * a database whose NAME marks it as a test DB (``*_test`` / + ``*_bench``) is trusted — local cortex_test and bench DBs; + * any OTHER name (e.g. the fresh runner-local ``cortex`` that real + CI creates) must hold ZERO memories at session start. A genuine + test database is created empty; pre-existing rows mean this is + somebody's real store. No threshold constant — empty is empty. + + Override (explicit, never default): CORTEX_TEST_ALLOW_POPULATED=1. + """ + if os.environ.get("CORTEX_TEST_ALLOW_POPULATED") == "1": + return + if looks_like_test_db(test_db_url): + return + try: + import psycopg + + with psycopg.connect(test_db_url, autocommit=True, connect_timeout=3) as conn: + row = conn.execute("SELECT COUNT(*) FROM memories").fetchone() + count = row[0] if row else 0 + except Exception: + return # unreachable/uninitialized DB — the SQLite fallback handles it + if count > 0: + pytest.exit( + f"REFUSING to run: test DB '{test_db_url}' is not named like a " + f"test database and already holds {count} memories. The test " + "suite DELETEs every table between tests — pointing it at a real " + "store destroys it (incident 2026-06-10: 537k rows lost). Use a " + "*_test database, or set CORTEX_TEST_ALLOW_POPULATED=1 if you " + "really mean it.", + returncode=2, + ) + + +def _resolved_real_data_roots() -> list[tuple[str, str]]: + """`(name, resolved value)` for every root that must stay inside the + isolated tree — read fresh from `mcp_server`, never from the env vars + `conftest.py`'s `_redirect_real_data_roots` set: an import that happened + too early, or a future edit that moves the redirection below the first + import, would leave these constants bound to the real `~/.claude` + while the env vars still look correct.""" + from mcp_server.infrastructure.config import CLAUDE_DIR, WIKI_ROOT + from mcp_server.infrastructure.memory_config import get_memory_settings + + settings = get_memory_settings() + return [ + ("CLAUDE_DIR", str(CLAUDE_DIR)), + ("WIKI_ROOT", str(WIKI_ROOT)), + ("MemorySettings.DB_PATH", settings.DB_PATH), + ("MemorySettings.SQLITE_FALLBACK_PATH", settings.SQLITE_FALLBACK_PATH), + ] + + +# Fail-closed and unconditional: there is no override. A suite that DELETEs +# every table between tests and rewrites the wiki has no safe way to run +# against a real tree (issue #219 — the previous isolation was conditional +# and its failure was silent for as long as it took to notice a 0-row store). +def guard_against_real_data_roots(isolated_claude_dir: str) -> None: + """Refuse to run if any resolved root still points at real user data.""" + isolated = os.path.realpath(isolated_claude_dir) + offenders = [ + (name, value) + for name, value in _resolved_real_data_roots() + if not os.path.realpath(os.path.expanduser(value)).startswith(isolated) + ] + if not offenders: + return + detail = "\n".join(f" {name} -> {value}" for name, value in offenders) + real_root = os.path.realpath(os.path.expanduser("~/.claude")) + pytest.exit( + "REFUSING to run: test isolation is not in effect. These roots " + f"still resolve outside the throwaway tree {isolated}:\n{detail}\n" + f"(real user root: {real_root}). The suite DELETEs every table " + "between tests and regenerates wiki dashboards — running it " + "against a real tree destroys data (issue #219). Ensure " + "_redirect_real_data_roots() runs before any mcp_server import.", + returncode=2, + ) diff --git a/tests_py/_pg_throwaway_db.py b/tests_py/_pg_throwaway_db.py new file mode 100644 index 00000000..42c29a86 --- /dev/null +++ b/tests_py/_pg_throwaway_db.py @@ -0,0 +1,146 @@ +"""Per-process throwaway PostgreSQL test-database lifecycle. + +Extracted from `tests_py/conftest.py` (issue #276/#287 boy-scout follow-up): +that file was 501 lines, over both the repo's local 300-line cap and the +global 500-line cap (coding-standards.md §4.1), with these functions +accounting for roughly a third of it. Split along the concern boundary that +was already visible in the file's own section comments — no behavior +change, only explicit-parameter functions in place of closures over +conftest.py's module globals (a DI improvement as a side effect, not the +goal: these are independently testable now). + +Ordering/behavior preserved exactly: `conftest.py` still creates the +isolated database (if any) before running its data-safety guards, still +drops it in `pytest_sessionfinish`, and every message/exception-swallowing +choice below is unchanged from the code this was extracted from. + +Why per-process isolation exists at all +---------------------------------------- +ROOT CAUSE (reproduced deterministically 3/3 + 3/3, 2026-07-10): every +worktree/agent used to default to the SAME physical DB (``cortex_test``). +``conftest.py``'s autouse ``_test_isolation`` fixture runs an unconditional +``DELETE FROM `` before/after EVERY test, in EVERY concurrently +running pytest process. Two pytest invocations sharing that DB race: +process A's between-test purge can delete the row process B just stored, +between B's ``remember()`` and its later ``recall()``/``get_memory()`` +call. This produced two previously-reported flakes with unrelated +proximate symptoms (``test_store_consolidate_recall``: +``recall_result["count"] == 0``; ``test_memory_count_unchanged_after_validation``: +``store.get_memory() is None``) — neither an intra-suite ordering bug nor +an embeddings-backend issue, both inter-process contention on one shared +table set. Retrying or skipping would hide the race, not fix it (forbidden +anti-pattern). The correct fix is to give each local pytest process its +own throwaway database — CI already gets a fresh service-container DB per +run, and an explicit ``CORTEX_TEST_DATABASE_URL`` override is respected +verbatim (the caller has taken responsibility for isolation). +""" + +from __future__ import annotations + +import os + + +def maintenance_url(url: str) -> str: + """Same host/creds as `url`, database swapped to the `postgres` maintenance DB.""" + base, _, _query = url.partition("?") + prefix, _, _dbname = base.rpartition("/") + return f"{prefix}/postgres" + + +def _orphan_pid(name: str) -> int | None: + """The pid encoded in a `cortex_test_pw_<8-hex>` database name, or + `None` when the name doesn't carry a numeric pid.""" + rest = name[len("cortex_test_pw") :] + pid_str = rest.split("_", 1)[0] + return int(pid_str) if pid_str.isdigit() else None + + +def _pid_is_definitely_dead(pid: int) -> bool: + """True only when `pid` is confirmed gone. Best-effort, non-fatal: every + ambiguous case (still alive, owned by another user, any other OS error) + counts as "alive" — never risk a false-positive drop of a database + still in use.""" + try: + os.kill(pid, 0) + return False # still alive + except ProcessLookupError: + return True + except Exception: + return False # permission-denied or any other ambiguity — leave it + + +def drop_dead_orphaned_databases(conn) -> None: + """Drop leftover `cortex_test_pw_` databases whose owning PID + is no longer alive. + + A pytest process killed abnormally (SIGTERM/SIGKILL — e.g. a CI job + cancellation or a shell timeout) never reaches `pytest_sessionfinish`, + so its throwaway database is never dropped (measured: 1 orphan produced + by a bash-tool 2-minute timeout during this investigation, 2026-07-10). + """ + try: + rows = conn.execute( + "SELECT datname FROM pg_database WHERE datname LIKE 'cortex_test_pw%';" + ).fetchall() + except Exception: + return + for row in rows: + name = row[0] if isinstance(row, tuple) else row.get("datname") + if not name: + continue + pid = _orphan_pid(name) + if pid is None or not _pid_is_definitely_dead(pid): + continue + try: + conn.execute( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " + "WHERE datname = %s AND pid <> pg_backend_pid()", + (name,), + ) + conn.execute(f'DROP DATABASE IF EXISTS "{name}"') + except Exception: + pass + + +def create_isolated_test_database(base_url: str) -> tuple[str, str, str] | None: + """Create a throwaway PG database for this pytest process. + + Returns `(isolated_url, maintenance_url, db_name)` on success, or `None` + on any failure (unreachable server, no CREATEDB privilege, etc.) so the + caller falls back to the shared `base_url` — degrading to prior + (contention-prone but functional) behavior rather than failing the + whole session. + """ + try: + import psycopg + + db_name = f"cortex_test_pw{os.getpid()}_{os.urandom(4).hex()}" + maint_url = maintenance_url(base_url) + with psycopg.connect(maint_url, autocommit=True, connect_timeout=3) as conn: + drop_dead_orphaned_databases(conn) + conn.execute(f'CREATE DATABASE "{db_name}"') + except Exception: + return None + prefix = base_url.rpartition("/")[0] + return f"{prefix}/{db_name}", maint_url, db_name + + +def drop_isolated_database(maint_url: str, db_name: str) -> None: + """Drop the throwaway per-process database created above, if any. + + Called from `conftest.py`'s `pytest_sessionfinish` hook (which must + stay a named hook function in conftest.py itself for pytest to find + it — this is the extracted body, not the hook). + """ + try: + import psycopg + + with psycopg.connect(maint_url, autocommit=True, connect_timeout=3) as conn: + conn.execute( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " + "WHERE datname = %s AND pid <> pg_backend_pid()", + (db_name,), + ) + conn.execute(f'DROP DATABASE IF EXISTS "{db_name}"') + except Exception: + pass diff --git a/tests_py/conftest.py b/tests_py/conftest.py index 2c1ba1bf..989429a5 100644 --- a/tests_py/conftest.py +++ b/tests_py/conftest.py @@ -118,6 +118,13 @@ def _redirect_real_data_roots() -> str: # database — CI already gets a fresh service-container DB per run, and an # explicit CORTEX_TEST_DATABASE_URL override is respected verbatim (the # caller has taken responsibility for isolation). +# +# The creation/cleanup mechanics live in tests_py/_pg_throwaway_db.py (moved +# out to bring this file under coding-standards.md §4.1's 300-line cap; +# issue #276/#287 boy-scout follow-up) — no behavior change, see that +# module's docstring. +from tests_py import _pg_safety_guards, _pg_throwaway_db # noqa: E402 + _CORTEX_TEST_ISOLATE_DB = os.environ.get("CORTEX_TEST_ISOLATE_DB", "1") not in ( "0", "false", @@ -125,207 +132,19 @@ def _redirect_real_data_roots() -> str: ) _OWNED_ISOLATED_DB: tuple[str, str] | None = None # (maintenance_url, db_name) - -def _maintenance_url(url: str) -> str: - """Same host/creds as `url`, database swapped to the `postgres` maintenance DB.""" - base, _, _query = url.partition("?") - prefix, _, _dbname = base.rpartition("/") - return f"{prefix}/postgres" - - -def _drop_dead_orphaned_databases(conn) -> None: - """Drop leftover `cortex_test_pw_` databases whose owning PID - is no longer alive. - - A pytest process killed abnormally (SIGTERM/SIGKILL — e.g. a CI job - cancellation or a shell timeout) never reaches `pytest_sessionfinish`, - so its throwaway database is never dropped (measured: 1 orphan produced - by a bash-tool 2-minute timeout during this investigation, 2026-07-10). - Best-effort, non-fatal: an unreachable/ambiguous entry is left alone - rather than risking a false-positive drop of a database still in use. - """ - try: - rows = conn.execute( - "SELECT datname FROM pg_database WHERE datname LIKE 'cortex_test_pw%';" - ).fetchall() - except Exception: - return - for row in rows: - name = row[0] if isinstance(row, tuple) else row.get("datname") - if not name: - continue - # Format: cortex_test_pw_<8-hex> - rest = name[len("cortex_test_pw") :] - pid_str = rest.split("_", 1)[0] - if not pid_str.isdigit(): - continue - pid = int(pid_str) - try: - os.kill(pid, 0) - continue # process still alive — not an orphan - except ProcessLookupError: - pass - except PermissionError: - continue # alive but owned by another user — leave it - except Exception: - continue - try: - conn.execute( - "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " - "WHERE datname = %s AND pid <> pg_backend_pid()", - (name,), - ) - conn.execute(f'DROP DATABASE IF EXISTS "{name}"') - except Exception: - pass - - -def _create_isolated_test_database(base_url: str) -> str | None: - """Create a throwaway PG database for this pytest process; return its URL. - - Returns None on any failure (unreachable server, no CREATEDB privilege, - etc.) so the caller falls back to the shared `base_url` — degrading to - prior (contention-prone but functional) behavior rather than failing the - whole session. - """ - try: - import psycopg - - db_name = f"cortex_test_pw{os.getpid()}_{os.urandom(4).hex()}" - maint_url = _maintenance_url(base_url) - with psycopg.connect(maint_url, autocommit=True, connect_timeout=3) as conn: - _drop_dead_orphaned_databases(conn) - conn.execute(f'CREATE DATABASE "{db_name}"') - except Exception: - return None - global _OWNED_ISOLATED_DB - _OWNED_ISOLATED_DB = (maint_url, db_name) - prefix = base_url.rpartition("/")[0] - return f"{prefix}/{db_name}" - - if not _IS_CI and _CORTEX_TEST_ISOLATE_DB and not _EXPLICIT_TEST_DB_URL: - _isolated_url = _create_isolated_test_database(_TEST_DB_URL) - if _isolated_url is not None: - _TEST_DB_URL = _isolated_url + _created = _pg_throwaway_db.create_isolated_test_database(_TEST_DB_URL) + if _created is not None: + _TEST_DB_URL, _maint_url, _db_name = _created + _OWNED_ISOLATED_DB = (_maint_url, _db_name) os.environ["DATABASE_URL"] = _TEST_DB_URL def pytest_sessionfinish(session, exitstatus) -> None: # noqa: ARG001 """Drop the throwaway per-process database created above, if any.""" - if _OWNED_ISOLATED_DB is None: - return - maint_url, db_name = _OWNED_ISOLATED_DB - try: - import psycopg - - with psycopg.connect(maint_url, autocommit=True, connect_timeout=3) as conn: - conn.execute( - "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " - "WHERE datname = %s AND pid <> pg_backend_pid()", - (db_name,), - ) - conn.execute(f'DROP DATABASE IF EXISTS "{db_name}"') - except Exception: - pass - - -def _looks_like_test_db(url: str) -> bool: - """True when the database NAME declares itself a test/bench database.""" - name = url.rsplit("/", 1)[-1].split("?", 1)[0] - return name.endswith("_test") or name.endswith("_bench") or "_test_" in name - - -def _guard_against_populated_db() -> None: - """Refuse to run destructive test isolation against a populated DB. - - INCIDENT 2026-06-10: an agent ran ``DATABASE_URL= CI=true - pytest`` — the CI branch above trusted the URL verbatim and - ``_clean_all_tables`` deleted 537,396 production memories (plus all - entities, relationships, and prospective triggers) between tests. - - Two-tier guard, fail-closed: - * a database whose NAME marks it as a test DB (``*_test`` / - ``*_bench``) is trusted — local cortex_test and bench DBs; - * any OTHER name (e.g. the fresh runner-local ``cortex`` that real - CI creates) must hold ZERO memories at session start. A genuine - test database is created empty; pre-existing rows mean this is - somebody's real store. No threshold constant — empty is empty. - - Override (explicit, never default): CORTEX_TEST_ALLOW_POPULATED=1. - """ - if os.environ.get("CORTEX_TEST_ALLOW_POPULATED") == "1": - return - if _looks_like_test_db(_TEST_DB_URL): - return - try: - import psycopg - - with psycopg.connect(_TEST_DB_URL, autocommit=True, connect_timeout=3) as conn: - row = conn.execute("SELECT COUNT(*) FROM memories").fetchone() - count = row[0] if row else 0 - except Exception: - return # unreachable/uninitialized DB — the SQLite fallback handles it - if count > 0: - import pytest - - pytest.exit( - f"REFUSING to run: test DB '{_TEST_DB_URL}' is not named like a " - f"test database and already holds {count} memories. The test " - "suite DELETEs every table between tests — pointing it at a real " - "store destroys it (incident 2026-06-10: 537k rows lost). Use a " - "*_test database, or set CORTEX_TEST_ALLOW_POPULATED=1 if you " - "really mean it.", - returncode=2, - ) - - -def _guard_against_real_data_roots() -> None: - """Refuse to run if any resolved root still points at real user data. - - `_redirect_real_data_roots` sets the env vars, but the values that - actually matter are the ones `mcp_server` RESOLVED from them — an import - that happened too early, or a future edit that moves the redirection - below the first import, would leave the constants bound to the real - `~/.claude` while the env vars look correct. This checks the resolved - values, so it cannot be fooled by a correct-looking environment. - - Fail-closed and unconditional: there is no override. A suite that DELETEs - every table between tests and rewrites the wiki has no safe way to run - against a real tree. - - source: issue #219 — the previous isolation was conditional and its - failure was silent for as long as it took to notice a 0-row store. - """ - from mcp_server.infrastructure.config import CLAUDE_DIR, WIKI_ROOT - from mcp_server.infrastructure.memory_config import get_memory_settings - - settings = get_memory_settings() - real_root = os.path.realpath(os.path.expanduser("~/.claude")) - isolated = os.path.realpath(_TEST_CLAUDE_DIR) - - offenders = [ - (name, value) - for name, value in ( - ("CLAUDE_DIR", str(CLAUDE_DIR)), - ("WIKI_ROOT", str(WIKI_ROOT)), - ("MemorySettings.DB_PATH", settings.DB_PATH), - ("MemorySettings.SQLITE_FALLBACK_PATH", settings.SQLITE_FALLBACK_PATH), - ) - if not os.path.realpath(os.path.expanduser(value)).startswith(isolated) - ] - if offenders: - detail = "\n".join(f" {name} -> {value}" for name, value in offenders) - pytest.exit( - "REFUSING to run: test isolation is not in effect. These roots " - f"still resolve outside the throwaway tree {isolated}:\n{detail}\n" - f"(real user root: {real_root}). The suite DELETEs every table " - "between tests and regenerates wiki dashboards — running it " - "against a real tree destroys data (issue #219). Ensure " - "_redirect_real_data_roots() runs before any mcp_server import.", - returncode=2, - ) + if _OWNED_ISOLATED_DB is not None: + _pg_throwaway_db.drop_isolated_database(*_OWNED_ISOLATED_DB) def _guard_against_venv_lock_drift() -> None: @@ -356,8 +175,8 @@ def _pg_available() -> bool: return False -_guard_against_populated_db() -_guard_against_real_data_roots() +_pg_safety_guards.guard_against_populated_db(_TEST_DB_URL) +_pg_safety_guards.guard_against_real_data_roots(_TEST_CLAUDE_DIR) _guard_against_venv_lock_drift() _USE_PG = _pg_available() @@ -409,52 +228,13 @@ def _effective_backend() -> str: # ── Isolate domain_mapping's dev-root scan from the real filesystem ────── # -# ROOT CAUSE (reproduced deterministically while measuring coverage for -# issue #196, 2026-07-27): shared/domain_mapping._build_registry() (used -# transitively by handlers/consolidation/wiki_backlog_pass.py -> -# core/wiki_drift.py's audit_wiki_drift -> _project_source_root -> -# _file_exists_under's os.walk fallback) has NO test override by -# default. _candidate_dev_roots() falls back to $HOME/Developments, -# $HOME/Documents/Developments, $HOME/dev, $HOME/code — real developer -# directories — and (this is the part that defeats a naive fix) is -# ADDITIVE: even with $CORTEX_DEV_ROOT set, every one of those defaults -# that exists on disk is STILL appended as an extra candidate (see -# shared/domain_mapping.py::_candidate_dev_roots — the env var is one -# more candidate, not a replacement). Setting the env var alone does -# NOT isolate the scan. On a machine where one of the defaults exists -# and holds real git repos (any contributor's normal dev machine), any -# test that exercises the full consolidate handler (e.g. -# test_consolidate_stage_exception_does_not_crash_siblings) transitively -# walks those REAL, unrelated, potentially enormous repo trees via -# os.walk — observed killing a `pytest --cov` run at the 300s per-test -# timeout (pyproject.toml's `timeout = 300`). This is invisible in CI -# because none of the candidate dev roots exist on a CI runner -# ($HOME=/home/runner), so _candidate_dev_roots() returns [] there and -# _project_source_root always returns None — CI's isolation is -# accidental, not designed. This plausibly explains the previously -# unexplained "CI stall on 148d5a1... hung >3h... no offender -# identified" this same `timeout` setting was added to catch (see the -# comment on `timeout_method` above): a local run with a populated dev -# root would reproduce this hang with no per-test timeout to bound it. -# -# Fix: replace `_candidate_dev_roots` itself (not just the env var) so -# it always returns an empty list for the whole test session, matching -# the exact isolation pattern tests_py/shared/test_domain_mapping.py -# already uses per-test (monkeypatch `_candidate_dev_roots` directly, -# then `_build_registry.cache_clear()`). Those dedicated tests -# monkeypatch this same attribute again inside their own test body — -# monkeypatch always restores to whatever was in place when the test -# started, so patching it here at collection time (not via the -# `monkeypatch` fixture, which only exists inside a running test) is a -# permanent module-level replacement for the whole session; the -# per-test monkeypatch calls inside test_domain_mapping.py override it -# for the duration of those specific tests and their own -# `_build_registry.cache_clear()` calls put the real cache state back -# in sync afterward. -from mcp_server.shared import domain_mapping as _dm # noqa: E402 - -_dm._candidate_dev_roots = lambda: [] -_dm._build_registry.cache_clear() +# Without this, a test exercising the full consolidate handler walks the +# real, unrelated (potentially enormous) dev-root trees on a contributor's +# machine via os.walk — see tests_py/_domain_mapping_isolation.py's +# docstring for the full incident history (issue #196). +from tests_py._domain_mapping_isolation import isolate_dev_root_scan # noqa: E402 + +isolate_dev_root_scan() def _get_raw_connection(): diff --git a/tests_py/invariants/test_conftest_guard.py b/tests_py/invariants/test_conftest_guard.py deleted file mode 100644 index 79f738e1..00000000 --- a/tests_py/invariants/test_conftest_guard.py +++ /dev/null @@ -1,496 +0,0 @@ -"""Regression guard for tests_py/conftest.py:_guard_against_populated_db. - -Incident 2026-06-10: 537,396 production memories were deleted because the -guard was absent. This test suite prevents a future change from silently -disabling or weakening the guard by verifying both branches in complete -isolation — WITHOUT a real PostgreSQL connection. - -Contract under test (_guard_against_populated_db): - 1. CORTEX_TEST_ALLOW_POPULATED=1 -> always returns (bypass override). - 2. DB URL name is *_test / *_bench / *_test_* -> returns (trusted name). - 3. DB URL is not a test name, try PG: - a. psycopg.connect raises any exception -> returns (SQLite fallback). - b. memories count == 0 -> returns (empty DB is safe). - c. memories count > 0 -> pytest.exit(returncode=2). - -All branches use unittest.mock only. No real PG connection is opened. - -Mocking strategy ----------------- -The guard body does ``import psycopg`` and ``import pytest`` as IMPORT_NAME -opcodes, which always resolve through ``sys.modules``. The guard function is -exec'd from the raw source into a fresh namespace that carries the controlled -``_TEST_DB_URL`` — avoiding the full conftest module-level setup code that -would overwrite the URL with the real test DB URL. - -``sys.modules['psycopg']`` is replaced with a fake module for the duration of -each test so the ``import psycopg`` inside the guard returns the fake. -``pytest.exit`` is patched on the real ``pytest`` object in sys.modules so the -``import pytest; pytest.exit(...)`` path inside the guard calls the mock. -""" - -from __future__ import annotations - -import os -import sys -import types -import unittest.mock as mock - -import pytest - - -# ── guard loader ───────────────────────────────────────────────────────────── - -_CONFTEST_PATH = os.path.abspath( - os.path.join(os.path.dirname(__file__), "..", "conftest.py") -) - -with open(_CONFTEST_PATH, encoding="utf-8") as _fh: - _CONFTEST_LINES = _fh.readlines() - - -# Extract just the two guard-related function definitions. We locate them by -# scanning for the ``def`` lines rather than hard-coding line numbers, so a -# shift from future edits above the functions does not break this test file. -def _find_function_block(lines: list[str], func_name: str) -> tuple[int, int]: - """Return (start, end) 0-indexed line indices for a top-level function.""" - start = next( - i for i, line in enumerate(lines) if line.startswith(f"def {func_name}(") - ) - # Find the next top-level def/class/assignment (col 0) after start. - end = len(lines) - for i in range(start + 1, len(lines)): - if lines[i] and lines[i][0] not in (" ", "\t", "\n", "#", '"', "'"): - end = i - break - return start, end - - -_looks_start, _looks_end = _find_function_block(_CONFTEST_LINES, "_looks_like_test_db") -_guard_start, _guard_end = _find_function_block( - _CONFTEST_LINES, "_guard_against_populated_db" -) -_FUNC_SOURCE = "".join( - _CONFTEST_LINES[_looks_start:_looks_end] + _CONFTEST_LINES[_guard_start:_guard_end] -) - - -def _make_guard(test_db_url: str) -> types.FunctionType: - """Compile and return the guard function bound to the given test_db_url. - - Executes only the two function definitions (not the full conftest module - startup code) so ``_TEST_DB_URL`` in the guard's globals is precisely the - value we supply, independent of the actual test-session DB URL. - """ - ns: dict = { - "os": os, - "pytest": pytest, - "_TEST_DB_URL": test_db_url, - } - exec(compile(_FUNC_SOURCE, _CONFTEST_PATH, "exec"), ns) # noqa: S102 — test only - return ns["_guard_against_populated_db"] - - -def _fake_psycopg(row_count: int | None = None, raise_exc: Exception | None = None): - """Build a fake psycopg module for sys.modules injection. - - Args: - row_count: if given, the mock connection returns this count. - raise_exc: if given, psycopg.connect raises this exception. - """ - fake = types.ModuleType("psycopg") - if raise_exc is not None: - fake.connect = mock.Mock(side_effect=raise_exc) - return fake - - mock_conn = mock.MagicMock() - mock_conn.__enter__ = mock.Mock(return_value=mock_conn) - mock_conn.__exit__ = mock.Mock(return_value=False) - fetchone_result = (row_count,) if row_count is not None else None - mock_conn.execute.return_value.fetchone.return_value = fetchone_result - fake.connect = mock.Mock(return_value=mock_conn) - return fake - - -# ── Branch 1: CORTEX_TEST_ALLOW_POPULATED=1 ────────────────────────────────── - - -class TestAllowPopulatedOverride: - """Guard must short-circuit immediately when the override env-var is set.""" - - def test_override_env_bypasses_guard(self, monkeypatch: pytest.MonkeyPatch): - """With CORTEX_TEST_ALLOW_POPULATED=1 the guard returns without - inspecting the DB, even for a non-test-named URL with a high row count. - - Postcondition: psycopg.connect is never called and pytest.exit is never - called when the override is active. - """ - monkeypatch.setenv("CORTEX_TEST_ALLOW_POPULATED", "1") - guard = _make_guard("postgresql://host/production") - fake_pg = _fake_psycopg(row_count=537396) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - fake_pg.connect.assert_not_called() - mock_exit.assert_not_called() - - def test_override_not_set_proceeds(self, monkeypatch: pytest.MonkeyPatch): - """Without the override, the guard must continue past the first check. - - We confirm this by using a test-named URL (triggers branch 2 short- - circuit) and verifying the guard returns without calling pytest.exit. - This is an indirect confirmation that branch 1 did NOT fire. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard("postgresql://host/cortex_test") - fake_pg = _fake_psycopg(row_count=0) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - mock_exit.assert_not_called() - - -# ── Branch 2: URL name check (_looks_like_test_db) ─────────────────────────── - - -class TestUrlNameCheck: - """Guard must trust URLs whose database name declares it a test/bench DB.""" - - @pytest.mark.parametrize( - "url", - [ - "postgresql://localhost/cortex_test", - "postgresql://localhost/mydb_bench", - "postgresql://localhost/cortex_test_2026", - "postgresql://user:pass@host:5432/cortex_test?sslmode=require", - ], - ) - def test_test_named_url_skips_connection( - self, url: str, monkeypatch: pytest.MonkeyPatch - ): - """A *_test / *_bench / *_test_* database name is trusted — the guard - returns without attempting any DB connection. - - Postcondition: psycopg.connect is not called and pytest.exit is not - called, regardless of what the DB might contain. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard(url) - fake_pg = _fake_psycopg(row_count=999999) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - fake_pg.connect.assert_not_called() - mock_exit.assert_not_called() - - @pytest.mark.parametrize( - "url", - [ - "postgresql://localhost/cortex", - "postgresql://localhost/production", - "postgresql://localhost/myapp_db", - ], - ) - def test_non_test_named_url_attempts_connection( - self, url: str, monkeypatch: pytest.MonkeyPatch - ): - """A URL whose database name is not a test name must attempt a DB - connection to inspect the row count. - - We make the fake psycopg raise to simulate PG unavailability (branch 3a). - The assertion is that connect WAS attempted — proving the guard did not - short-circuit at branch 2 for a non-test-named URL. - - Postcondition: psycopg.connect is called exactly once. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard(url) - fake_pg = _fake_psycopg(raise_exc=Exception("no PG available")) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() # must not raise (exception is caught inside the guard) - fake_pg.connect.assert_called_once() - mock_exit.assert_not_called() - - -# ── Branch 3a: PG unreachable -> guard allows (SQLite fallback path) ────────── - - -class TestPgUnreachable: - """When psycopg.connect raises any exception the guard must return silently. - - This is the normal path in environments without PG (including this CI). - The guard must NEVER propagate the connection exception — that would break - every test-collection run in environments without PG. - """ - - def test_operational_error_does_not_raise_or_block( - self, monkeypatch: pytest.MonkeyPatch - ): - """OperationalError from psycopg must be caught; guard returns silently. - - Postcondition: guard() returns without raising, pytest.exit not called. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard("postgresql://localhost/mydb") - fake_pg = _fake_psycopg(raise_exc=Exception("Connection refused")) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - mock_exit.assert_not_called() - - def test_any_exception_type_is_caught(self, monkeypatch: pytest.MonkeyPatch): - """The guard catches a bare Exception base class, so any subclass must - be swallowed — including RuntimeError, OSError, TimeoutError. - - Postcondition: guard() returns without raising for RuntimeError. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard("postgresql://localhost/mydb") - fake_pg = _fake_psycopg(raise_exc=RuntimeError("timeout")) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - mock_exit.assert_not_called() - - -# ── Branch 3b: PG reachable, count == 0 -> guard allows ────────────────────── - - -class TestEmptyDb: - """An empty non-test-named database is safe to wipe — guard must allow.""" - - def test_zero_count_allows_proceed(self, monkeypatch: pytest.MonkeyPatch): - """memories table with 0 rows must not trigger pytest.exit. - - Postcondition: pytest.exit is not called when count == 0. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard("postgresql://localhost/mydb") - fake_pg = _fake_psycopg(row_count=0) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - mock_exit.assert_not_called() - - def test_fetchone_none_allows_proceed(self, monkeypatch: pytest.MonkeyPatch): - """fetchone() returning None (e.g. table absent) must also allow. - - The guard treats None as count=0: - count = row[0] if row else 0 - - Postcondition: pytest.exit is not called when fetchone returns None. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard("postgresql://localhost/mydb") - # row_count=None instructs _fake_psycopg to return fetchone_result=None - fake_pg = _fake_psycopg(row_count=None) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - mock_exit.assert_not_called() - - -# ── Branch 3c: PG reachable, count > 0 -> BLOCKED (the anti-537k path) ─────── - - -class TestPopulatedDbBlocked: - """A populated non-test-named database MUST be blocked by the guard. - - This class directly tests the critical branch that the 2026-06-10 incident - demonstrated was absent. Any change that weakens this branch will cause - these tests to fail, making that change visible in CI before it ships. - """ - - def test_populated_db_calls_pytest_exit(self, monkeypatch: pytest.MonkeyPatch): - """count=537396 must trigger pytest.exit(returncode=2). - - Postcondition: pytest.exit is called exactly once with returncode=2. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard("postgresql://localhost/mydb") - fake_pg = _fake_psycopg(row_count=537396) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - mock_exit.assert_called_once() - call_kwargs = mock_exit.call_args[1] - assert call_kwargs.get("returncode") == 2, ( - f"pytest.exit not called with returncode=2 — got {mock_exit.call_args}" - ) - - def test_exit_message_mentions_row_count(self, monkeypatch: pytest.MonkeyPatch): - """The exit message must include the row count so operators can identify - which database was pointed at. - - Postcondition: the first positional argument to pytest.exit contains the - string representation of the count. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard("postgresql://localhost/mydb") - row_count = 42 - fake_pg = _fake_psycopg(row_count=row_count) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - assert mock_exit.called, "pytest.exit was not called for count=42" - exit_message = mock_exit.call_args[0][0] - assert str(row_count) in exit_message, ( - f"Exit message does not mention row count {row_count}: {exit_message!r}" - ) - - def test_single_row_is_enough_to_block(self, monkeypatch: pytest.MonkeyPatch): - """Even a single pre-existing memory must block the suite. - - The guard comment states 'No threshold constant — empty is empty.' - count=1 must be treated identically to 537,396. - - Postcondition: pytest.exit called with returncode=2 for count=1. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard("postgresql://localhost/mydb") - fake_pg = _fake_psycopg(row_count=1) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - assert mock_exit.called, ( - "Guard did not call pytest.exit for count=1 — " - "'empty is empty' invariant violated." - ) - call_kwargs = mock_exit.call_args[1] - assert call_kwargs.get("returncode") == 2 - - def test_exit_not_called_for_count_zero_after_populated_check( - self, monkeypatch: pytest.MonkeyPatch - ): - """Boundary: count=0 must NOT trigger the block even for a non-test URL. - - This guards against an off-by-one change (count >= 0) that would - block on every fresh empty database. - - Postcondition: pytest.exit is not called when count == 0. - """ - monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) - guard = _make_guard("postgresql://localhost/mydb") - fake_pg = _fake_psycopg(row_count=0) - - with ( - mock.patch.dict(sys.modules, {"psycopg": fake_pg}), - mock.patch.object(pytest, "exit") as mock_exit, - ): - guard() - mock_exit.assert_not_called() - - -# ── Structural guard: function present and called at module level ───────────── - - -class TestGuardStructuralIntegrity: - """Static-analysis tests that verify the guard cannot be silently removed. - - These tests read the conftest source and assert structural properties — - the function definition exists and the unconditional module-level call - site is present. A change that deletes the function or wraps the call - in a conditional will fail one of these tests. - """ - - def test_function_definition_present(self): - """_guard_against_populated_db must still be defined in conftest.py. - - Postcondition: the function definition line appears in the source. - """ - with open(_CONFTEST_PATH, encoding="utf-8") as fh: - source = fh.read() - - assert "def _guard_against_populated_db(" in source, ( - "conftest.py no longer defines _guard_against_populated_db — " - "the anti-537k guard has been removed." - ) - - def test_module_level_call_present(self): - """_guard_against_populated_db() must be called at module (col=0) level. - - The call must be unconditional — not inside an if-block or a function. - A future refactor that moves the call inside a conditional or removes it - entirely will cause the guard to stop running at collection time. - - Postcondition: at least one line at zero indentation contains exactly - '_guard_against_populated_db()'. - """ - with open(_CONFTEST_PATH, encoding="utf-8") as fh: - lines = fh.readlines() - - module_level_calls = [ - (i + 1, line.rstrip()) - for i, line in enumerate(lines) - if line.strip() == "_guard_against_populated_db()" - and not line.startswith((" ", "\t")) - ] - assert module_level_calls, ( - "conftest.py no longer calls _guard_against_populated_db() at " - "module level — the anti-537k guard is no longer active at " - "test collection time. The call must exist at zero indentation." - ) - - def test_returncode_2_present_in_source(self): - """The guard must still use returncode=2 (not 0 or 1) so CI can - distinguish a deliberate guard-abort from a normal test failure. - - Postcondition: 'returncode=2' appears in conftest.py. - """ - with open(_CONFTEST_PATH, encoding="utf-8") as fh: - source = fh.read() - - assert "returncode=2" in source, ( - "conftest.py _guard_against_populated_db no longer uses " - "returncode=2 — operators will not be able to distinguish a " - "guard-abort from a normal test failure." - ) - - def test_incident_date_mentioned_in_guard(self): - """The guard docstring must reference the incident date (2026-06-10) - so future maintainers understand why the guard exists. - - Postcondition: '2026-06-10' appears in conftest.py. - """ - with open(_CONFTEST_PATH, encoding="utf-8") as fh: - source = fh.read() - - assert "2026-06-10" in source, ( - "The incident reference date '2026-06-10' has been removed from " - "conftest.py — maintainers will lose context for why this guard " - "exists." - ) diff --git a/tests_py/invariants/test_domain_mapping_isolation.py b/tests_py/invariants/test_domain_mapping_isolation.py new file mode 100644 index 00000000..24680c22 --- /dev/null +++ b/tests_py/invariants/test_domain_mapping_isolation.py @@ -0,0 +1,42 @@ +"""Tests for tests_py/_domain_mapping_isolation.py (issue #276/#287 +boy-scout follow-up — extracted from tests_py/conftest.py). + +The function under test replaces `domain_mapping._candidate_dev_roots` +with a permanent empty-list stub for the whole session — see that module's +docstring for the incident (#196) this prevents. `conftest.py` already +calls it unconditionally at import time (which is what makes every other +test in the suite run isolated from the real filesystem); this file pins +the mechanism itself. +""" + +from __future__ import annotations + +from mcp_server.shared import domain_mapping as dm +from tests_py._domain_mapping_isolation import isolate_dev_root_scan + + +class TestIsolateDevRootScan: + def test_replaces_candidate_dev_roots_with_an_empty_list(self) -> None: + original = dm._candidate_dev_roots + try: + dm._candidate_dev_roots = lambda: ["/some/real/dev/root"] + isolate_dev_root_scan() + assert dm._candidate_dev_roots() == [] + finally: + dm._candidate_dev_roots = original + dm._build_registry.cache_clear() + + def test_clears_the_build_registry_cache(self) -> None: + """A stale cached registry (built before the stub was installed) + would still report the real dev roots' domains until cleared.""" + original = dm._candidate_dev_roots + try: + dm._candidate_dev_roots = lambda: [] + dm._build_registry() # populate the cache with the stub in place + assert dm._build_registry.cache_info().currsize == 1 + + isolate_dev_root_scan() + assert dm._build_registry.cache_info().currsize == 0 + finally: + dm._candidate_dev_roots = original + dm._build_registry.cache_clear() diff --git a/tests_py/invariants/test_guard_against_populated_db.py b/tests_py/invariants/test_guard_against_populated_db.py new file mode 100644 index 00000000..ee257017 --- /dev/null +++ b/tests_py/invariants/test_guard_against_populated_db.py @@ -0,0 +1,270 @@ +"""Regression guard for tests_py/_pg_safety_guards.py:guard_against_populated_db +— the non-blocking behavioral branches. + +Incident 2026-06-10: 537,396 production memories were deleted because the +guard was absent. This test suite prevents a future change from silently +disabling or weakening the guard by verifying each branch in complete +isolation — WITHOUT a real PostgreSQL connection. + +Contract under test: + 1. CORTEX_TEST_ALLOW_POPULATED=1 -> always returns (bypass override). + 2. DB URL name is *_test / *_bench / *_test_* -> returns (trusted name). + 3. DB URL is not a test name, try PG: + a. psycopg.connect raises any exception -> returns (SQLite fallback). + b. memories count == 0 -> returns (empty DB is safe). + +The count > 0 (BLOCKED) branch — the one the incident demonstrated was +absent — has its own file, `test_guard_blocks_populated_db.py`: it is the +single most safety-critical branch here, and splitting it out keeps this +file (and that one) under coding-standards.md §4.1's 300-line cap. The +guard's structural integrity (can the function itself be silently removed +or its module-level call site weakened?) is pinned separately in +`test_guard_against_populated_db_structural.py`. + +All branches use unittest.mock only. No real PG connection is opened. + +Mocking strategy +---------------- +Issue #276/#287 boy-scout follow-up moved the guard out of +`tests_py/conftest.py` into the standalone `tests_py/_pg_safety_guards.py` +(bringing conftest.py under coding-standards.md §4.1's 300-line cap) and +turned its DB-URL dependency from a module-global closure into an explicit +parameter. That module imports only `os` and `pytest` — no heavy +conftest.py bootstrapping (DB connections, filesystem redirection) runs on +import — so this test file can import it directly and call the real +function. + +``sys.modules['psycopg']`` is replaced with a fake module for the duration of +each test so the ``import psycopg`` inside the guard returns the fake. +``pytest.exit`` is patched on the real ``pytest`` object in sys.modules so the +``import pytest; pytest.exit(...)`` path inside the guard calls the mock. +""" + +from __future__ import annotations + +import sys +import types +import unittest.mock as mock + +import pytest + +from tests_py import _pg_safety_guards as guards + + +def _fake_psycopg(row_count: int | None = None, raise_exc: Exception | None = None): + """Build a fake psycopg module for sys.modules injection. + + Args: + row_count: if given, the mock connection returns this count. + raise_exc: if given, psycopg.connect raises this exception. + """ + fake = types.ModuleType("psycopg") + if raise_exc is not None: + fake.connect = mock.Mock(side_effect=raise_exc) + return fake + + mock_conn = mock.MagicMock() + mock_conn.__enter__ = mock.Mock(return_value=mock_conn) + mock_conn.__exit__ = mock.Mock(return_value=False) + fetchone_result = (row_count,) if row_count is not None else None + mock_conn.execute.return_value.fetchone.return_value = fetchone_result + fake.connect = mock.Mock(return_value=mock_conn) + return fake + + +# ── Branch 1: CORTEX_TEST_ALLOW_POPULATED=1 ────────────────────────────────── + + +class TestAllowPopulatedOverride: + """Guard must short-circuit immediately when the override env-var is set.""" + + def test_override_env_bypasses_guard(self, monkeypatch: pytest.MonkeyPatch): + """With CORTEX_TEST_ALLOW_POPULATED=1 the guard returns without + inspecting the DB, even for a non-test-named URL with a high row count. + + Postcondition: psycopg.connect is never called and pytest.exit is never + called when the override is active. + """ + monkeypatch.setenv("CORTEX_TEST_ALLOW_POPULATED", "1") + fake_pg = _fake_psycopg(row_count=537396) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db("postgresql://host/production") + fake_pg.connect.assert_not_called() + mock_exit.assert_not_called() + + def test_override_not_set_proceeds(self, monkeypatch: pytest.MonkeyPatch): + """Without the override, the guard must continue past the first check. + + We confirm this by using a test-named URL (triggers branch 2 short- + circuit) and verifying the guard returns without calling pytest.exit. + This is an indirect confirmation that branch 1 did NOT fire. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + fake_pg = _fake_psycopg(row_count=0) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db("postgresql://host/cortex_test") + mock_exit.assert_not_called() + + +# ── Branch 2: URL name check (looks_like_test_db) ──────────────────────────── + + +class TestUrlNameCheck: + """Guard must trust URLs whose database name declares it a test/bench DB.""" + + @pytest.mark.parametrize( + "url", + [ + "postgresql://localhost/cortex_test", + "postgresql://localhost/mydb_bench", + "postgresql://localhost/cortex_test_2026", + "postgresql://user:pass@host:5432/cortex_test?sslmode=require", + ], + ) + def test_test_named_url_skips_connection( + self, url: str, monkeypatch: pytest.MonkeyPatch + ): + """A *_test / *_bench / *_test_* database name is trusted — the guard + returns without attempting any DB connection. + + Postcondition: psycopg.connect is not called and pytest.exit is not + called, regardless of what the DB might contain. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + fake_pg = _fake_psycopg(row_count=999999) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db(url) + fake_pg.connect.assert_not_called() + mock_exit.assert_not_called() + + @pytest.mark.parametrize( + "url", + [ + "postgresql://localhost/cortex", + "postgresql://localhost/production", + "postgresql://localhost/myapp_db", + ], + ) + def test_non_test_named_url_attempts_connection( + self, url: str, monkeypatch: pytest.MonkeyPatch + ): + """A URL whose database name is not a test name must attempt a DB + connection to inspect the row count, with the exact args the guard + promises (the url itself, autocommit so the COUNT query needs no + explicit commit, and a bounded connect_timeout so an unreachable + host fails fast instead of hanging collection). + + We make the fake psycopg raise to simulate PG unavailability (branch 3a). + The assertion is that connect WAS attempted — proving the guard did not + short-circuit at branch 2 for a non-test-named URL. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + fake_pg = _fake_psycopg(raise_exc=Exception("no PG available")) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db(url) # must not raise + fake_pg.connect.assert_called_once_with( + url, autocommit=True, connect_timeout=3 + ) + mock_exit.assert_not_called() + + +# ── Branch 3a: PG unreachable -> guard allows (SQLite fallback path) ────────── + + +class TestPgUnreachable: + """When psycopg.connect raises any exception the guard must return silently. + + This is the normal path in environments without PG (including this CI). + The guard must NEVER propagate the connection exception — that would break + every test-collection run in environments without PG. + """ + + def test_operational_error_does_not_raise_or_block( + self, monkeypatch: pytest.MonkeyPatch + ): + """OperationalError from psycopg must be caught; guard returns silently. + + Postcondition: guard() returns without raising, pytest.exit not called. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + fake_pg = _fake_psycopg(raise_exc=Exception("Connection refused")) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db("postgresql://localhost/mydb") + mock_exit.assert_not_called() + + def test_any_exception_type_is_caught(self, monkeypatch: pytest.MonkeyPatch): + """The guard catches a bare Exception base class, so any subclass must + be swallowed — including RuntimeError, OSError, TimeoutError. + + Postcondition: guard() returns without raising for RuntimeError. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + fake_pg = _fake_psycopg(raise_exc=RuntimeError("timeout")) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db("postgresql://localhost/mydb") + mock_exit.assert_not_called() + + +# ── Branch 3b: PG reachable, count == 0 -> guard allows ────────────────────── + + +class TestEmptyDb: + """An empty non-test-named database is safe to wipe — guard must allow.""" + + def test_zero_count_allows_proceed(self, monkeypatch: pytest.MonkeyPatch): + """memories table with 0 rows must not trigger pytest.exit. + + Postcondition: pytest.exit is not called when count == 0. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + fake_pg = _fake_psycopg(row_count=0) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db("postgresql://localhost/mydb") + mock_exit.assert_not_called() + + def test_fetchone_none_allows_proceed(self, monkeypatch: pytest.MonkeyPatch): + """fetchone() returning None (e.g. table absent) must also allow. + + The guard treats None as count=0: + count = row[0] if row else 0 + + Postcondition: pytest.exit is not called when fetchone returns None. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + # row_count=None instructs _fake_psycopg to return fetchone_result=None + fake_pg = _fake_psycopg(row_count=None) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db("postgresql://localhost/mydb") + mock_exit.assert_not_called() diff --git a/tests_py/invariants/test_guard_against_populated_db_structural.py b/tests_py/invariants/test_guard_against_populated_db_structural.py new file mode 100644 index 00000000..6cbfa49e --- /dev/null +++ b/tests_py/invariants/test_guard_against_populated_db_structural.py @@ -0,0 +1,95 @@ +"""Static-analysis regression guard: `guard_against_populated_db` cannot be +silently removed or its module-level call site weakened. + +Split out of test_conftest_guard.py (issue #276/#287 boy-scout follow-up +— see test_guard_against_populated_db.py's docstring for the size-cap +rationale this split serves). These tests read the source of +`_pg_safety_guards.py` (the guard's definition) and `conftest.py` (its +unconditional call site) and assert structural properties. A change that +deletes the function, changes its name, or wraps the call in a +conditional will fail one of these tests. +""" + +from __future__ import annotations + +import os + +_CONFTEST_PATH = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "conftest.py") +) +_GUARDS_PATH = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "_pg_safety_guards.py") +) + + +class TestGuardStructuralIntegrity: + def test_function_definition_present(self): + """guard_against_populated_db must still be defined in + _pg_safety_guards.py. + + Postcondition: the function definition line appears in the source. + """ + with open(_GUARDS_PATH, encoding="utf-8") as fh: + source = fh.read() + + assert "def guard_against_populated_db(" in source, ( + "_pg_safety_guards.py no longer defines guard_against_populated_db " + "— the anti-537k guard has been removed." + ) + + def test_module_level_call_present(self): + """guard_against_populated_db(...) must be called from conftest.py at + module (col=0) level. + + The call must be unconditional — not inside an if-block or a function. + A future refactor that moves the call inside a conditional or removes it + entirely will cause the guard to stop running at collection time. + + Postcondition: at least one line at zero indentation calls + `_pg_safety_guards.guard_against_populated_db(...)`. + """ + with open(_CONFTEST_PATH, encoding="utf-8") as fh: + lines = fh.readlines() + + module_level_calls = [ + (i + 1, line.rstrip()) + for i, line in enumerate(lines) + if line.strip().startswith("_pg_safety_guards.guard_against_populated_db(") + and not line.startswith((" ", "\t")) + ] + assert module_level_calls, ( + "conftest.py no longer calls " + "_pg_safety_guards.guard_against_populated_db(...) at module " + "level — the anti-537k guard is no longer active at test " + "collection time. The call must exist at zero indentation." + ) + + def test_returncode_2_present_in_source(self): + """The guard must still use returncode=2 (not 0 or 1) so CI can + distinguish a deliberate guard-abort from a normal test failure. + + Postcondition: 'returncode=2' appears in _pg_safety_guards.py. + """ + with open(_GUARDS_PATH, encoding="utf-8") as fh: + source = fh.read() + + assert "returncode=2" in source, ( + "_pg_safety_guards.py's guard_against_populated_db no longer " + "uses returncode=2 — operators will not be able to distinguish " + "a guard-abort from a normal test failure." + ) + + def test_incident_date_mentioned_in_guard(self): + """The guard docstring must reference the incident date (2026-06-10) + so future maintainers understand why the guard exists. + + Postcondition: '2026-06-10' appears in _pg_safety_guards.py. + """ + with open(_GUARDS_PATH, encoding="utf-8") as fh: + source = fh.read() + + assert "2026-06-10" in source, ( + "The incident reference date '2026-06-10' has been removed from " + "_pg_safety_guards.py — maintainers will lose context for why " + "this guard exists." + ) diff --git a/tests_py/invariants/test_guard_blocks_populated_db.py b/tests_py/invariants/test_guard_blocks_populated_db.py new file mode 100644 index 00000000..4327ac8d --- /dev/null +++ b/tests_py/invariants/test_guard_blocks_populated_db.py @@ -0,0 +1,160 @@ +"""Regression guard for the anti-537k branch of +tests_py/_pg_safety_guards.py:guard_against_populated_db. + +Split out of test_conftest_guard.py (issue #276/#287 boy-scout follow-up +— see test_guard_against_populated_db.py's docstring for the size-cap +rationale this split serves). This class directly tests the critical +branch the 2026-06-10 incident (537,396 production memories deleted) +demonstrated was absent: a populated, non-test-named database MUST be +blocked. Any change that weakens this branch must fail one of these +tests, making that change visible in CI before it ships. +""" + +from __future__ import annotations + +import sys +import types +import unittest.mock as mock + +import pytest + +from tests_py import _pg_safety_guards as guards + + +def _fake_psycopg(row_count: int | None = None, raise_exc: Exception | None = None): + """Build a fake psycopg module for sys.modules injection. + + Args: + row_count: if given, the mock connection returns this count. + raise_exc: if given, psycopg.connect raises this exception. + """ + fake = types.ModuleType("psycopg") + if raise_exc is not None: + fake.connect = mock.Mock(side_effect=raise_exc) + return fake + + mock_conn = mock.MagicMock() + mock_conn.__enter__ = mock.Mock(return_value=mock_conn) + mock_conn.__exit__ = mock.Mock(return_value=False) + fetchone_result = (row_count,) if row_count is not None else None + mock_conn.execute.return_value.fetchone.return_value = fetchone_result + fake.connect = mock.Mock(return_value=mock_conn) + return fake + + +class TestPopulatedDbBlocked: + """A populated non-test-named database MUST be blocked by the guard.""" + + def test_populated_db_calls_pytest_exit(self, monkeypatch: pytest.MonkeyPatch): + """count=537396 must trigger pytest.exit(returncode=2), read from the + exact COUNT query against the exact table the incident deleted from. + + Postcondition: pytest.exit is called exactly once with returncode=2. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + fake_pg = _fake_psycopg(row_count=537396) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db("postgresql://localhost/mydb") + mock_exit.assert_called_once() + call_kwargs = mock_exit.call_args[1] + assert call_kwargs.get("returncode") == 2, ( + f"pytest.exit not called with returncode=2 — got {mock_exit.call_args}" + ) + conn = fake_pg.connect.return_value + conn.execute.assert_called_once_with("SELECT COUNT(*) FROM memories") + + def test_exit_message_is_exact(self, monkeypatch: pytest.MonkeyPatch): + """Exact equality (not `assertIn`): a wording tweak anywhere in this + message is a real change a reviewer should see reflected here. + + Postcondition: pytest.exit's first positional argument matches the + message byte-for-byte. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + row_count = 42 + url = "postgresql://localhost/mydb" + fake_pg = _fake_psycopg(row_count=row_count) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db(url) + mock_exit.assert_called_once_with( + f"REFUSING to run: test DB '{url}' is not named like a " + f"test database and already holds {row_count} memories. The " + "test suite DELETEs every table between tests — pointing it " + "at a real store destroys it (incident 2026-06-10: 537k rows " + "lost). Use a *_test database, or set " + "CORTEX_TEST_ALLOW_POPULATED=1 if you really mean it.", + returncode=2, + ) + + def test_exit_message_mentions_row_count(self, monkeypatch: pytest.MonkeyPatch): + """The exit message must include the row count so operators can identify + which database was pointed at. + + Postcondition: the first positional argument to pytest.exit contains the + string representation of the count. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + row_count = 42 + fake_pg = _fake_psycopg(row_count=row_count) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db("postgresql://localhost/mydb") + assert mock_exit.called, "pytest.exit was not called for count=42" + exit_message = mock_exit.call_args[0][0] + assert str(row_count) in exit_message, ( + f"Exit message does not mention row count {row_count}: {exit_message!r}" + ) + + def test_single_row_is_enough_to_block(self, monkeypatch: pytest.MonkeyPatch): + """Even a single pre-existing memory must block the suite. + + The guard comment states 'No threshold constant — empty is empty.' + count=1 must be treated identically to 537,396. + + Postcondition: pytest.exit called with returncode=2 for count=1. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + fake_pg = _fake_psycopg(row_count=1) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db("postgresql://localhost/mydb") + assert mock_exit.called, ( + "Guard did not call pytest.exit for count=1 — " + "'empty is empty' invariant violated." + ) + call_kwargs = mock_exit.call_args[1] + assert call_kwargs.get("returncode") == 2 + + def test_exit_not_called_for_count_zero_after_populated_check( + self, monkeypatch: pytest.MonkeyPatch + ): + """Boundary: count=0 must NOT trigger the block even for a non-test URL. + + This guards against an off-by-one change (count >= 0) that would + block on every fresh empty database. + + Postcondition: pytest.exit is not called when count == 0. + """ + monkeypatch.delenv("CORTEX_TEST_ALLOW_POPULATED", raising=False) + fake_pg = _fake_psycopg(row_count=0) + + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(pytest, "exit") as mock_exit, + ): + guards.guard_against_populated_db("postgresql://localhost/mydb") + mock_exit.assert_not_called() diff --git a/tests_py/invariants/test_looks_like_test_db.py b/tests_py/invariants/test_looks_like_test_db.py new file mode 100644 index 00000000..e7048a0c --- /dev/null +++ b/tests_py/invariants/test_looks_like_test_db.py @@ -0,0 +1,65 @@ +"""Direct unit coverage of tests_py/_pg_safety_guards.py:looks_like_test_db. + +Split out of test_conftest_guard.py (issue #276/#287 boy-scout follow-up): +that file, itself already over coding-standards.md §4.1's 300-line cap +before this session's edits (496 lines on origin/main), grew further with +the new tests this session added and needed decomposing along its natural +per-function boundaries — this file covers `looks_like_test_db` on its +own; `test_guard_against_populated_db.py` and its structural-integrity +sibling cover `guard_against_populated_db`. +""" + +from __future__ import annotations + +from tests_py import _pg_safety_guards as guards + + +class TestLooksLikeTestDb: + """Direct unit coverage of the URL -> database-name parser. + + The trailing `_test`/`_bench`/`_test_` checks are suffix/substring + checks, so what matters is which slice of the URL `looks_like_test_db` + feeds them — a wrong slice can produce a false positive (a stray + "_test_" elsewhere in the URL wrongly trusted) or a false negative (the + real name missed because the query string wasn't stripped). + """ + + def test_simple_test_named_db(self) -> None: + assert guards.looks_like_test_db("postgresql://localhost/cortex_test") + + def test_plain_db_name_is_not_test_named(self) -> None: + assert not guards.looks_like_test_db("postgresql://localhost/cortex") + + def test_query_string_is_stripped_before_the_suffix_check(self) -> None: + """Without stripping "?sslmode=require", the name would end in the + query string, not "_test", and wrongly report False.""" + assert guards.looks_like_test_db( + "postgresql://localhost/cortex_test?sslmode=require" + ) + + def test_a_test_looking_host_with_a_real_db_name_is_not_trusted(self) -> None: + """The database NAME must be checked, never the host: a host that + happens to contain "_test_" (e.g. a shared test-labeled server used + for a real database) must not make an unrelated real DB look safe. + This is exactly what an rsplit-vs-split parsing bug would get wrong + — split() on the first "/" would fold the host into the checked + text; rsplit() on the LAST "/" correctly isolates only the name.""" + assert not guards.looks_like_test_db( + "postgresql://cortex_test_host:5432/production" + ) + + def test_query_string_split_takes_the_first_question_mark_not_the_last( + self, + ) -> None: + """A (malformed but possible) URL with two "?" characters must be + split at the FIRST one — `split`, not `rsplit` — or "cortex_test" + gets a stray "?a=1" appended and no longer ends in "_test".""" + assert guards.looks_like_test_db("postgresql://localhost/cortex_test?a=1?b=2") + + def test_a_url_with_no_slash_at_all_does_not_raise(self) -> None: + """A bare name with no "postgresql://" prefix is a degenerate but + valid input for this pure string function; it must resolve `name` + to the input itself rather than raising (rsplit finds nothing to + split on and returns a single-element result — indexing that + result at a position other than 0/-1 would raise `IndexError`).""" + assert guards.looks_like_test_db("cortex_test") diff --git a/tests_py/invariants/test_pg_throwaway_db.py b/tests_py/invariants/test_pg_throwaway_db.py new file mode 100644 index 00000000..73099510 --- /dev/null +++ b/tests_py/invariants/test_pg_throwaway_db.py @@ -0,0 +1,289 @@ +"""Tests for tests_py/_pg_throwaway_db.py — per-process throwaway PG database +lifecycle (issue #276/#287 boy-scout follow-up). + +Extracted from `tests_py/conftest.py`'s module-level bootstrap, these +functions previously had no dedicated unit coverage of their own — only +whatever the full test-session bootstrap exercised implicitly. This file +closes that gap for the pieces that matter most: which orphan databases get +dropped (a wrong answer here either leaks throwaway databases forever or +drops one still in use), and that database creation/drop always degrade to +`None`/no-op on any failure rather than raising into collection. + +Assertions are exact (`assert_called_once_with`, full SQL/message text) +rather than substring checks wherever a call site is otherwise mocked +permissively enough that a wrong argument or a corrupted literal would +still "work" against the fake — e.g. a fake `psycopg.connect` that accepts +any arguments cannot itself catch a real `test_db_url` silently swapped for +`None`; only asserting the exact call can. +""" + +from __future__ import annotations + +import os +import re +import sys +import types +import unittest.mock as mock + +from tests_py import _pg_throwaway_db as throwaway + + +class TestMaintenanceUrl: + def test_swaps_the_database_name_for_postgres(self) -> None: + assert ( + throwaway.maintenance_url("postgresql://host:5432/cortex_test") + == "postgresql://host:5432/postgres" + ) + + def test_drops_any_query_string(self) -> None: + assert ( + throwaway.maintenance_url("postgresql://host/cortex_test?sslmode=require") + == "postgresql://host/postgres" + ) + + def test_query_string_is_removed_before_finding_the_last_slash(self) -> None: + """A query string containing its own "/" (e.g. a file-path option) + must not be mistaken for the database-name separator: the "?" split + has to happen BEFORE the "/" rpartition, not after.""" + assert ( + throwaway.maintenance_url("postgresql://host/cortex_test?options=/tmp/foo") + == "postgresql://host/postgres" + ) + + +class TestOrphanPid: + """`rest.split("_", 1)[0]` — every name this function is ever called + against (`cortex_test_pw_<8-lowercase-hex-chars>`, produced by + `create_isolated_test_database` and matched by the `LIKE + 'cortex_test_pw%'` query `drop_dead_orphaned_databases` runs) has + EXACTLY ONE "_" in `rest` — the hex suffix is `os.urandom(4).hex()`, + whose alphabet (0-9a-f) never contains one. With only one separator to + split on, `split(sep, 1)`, an unbounded `split(sep)`, `split(sep, 2)`, + and even `rsplit(sep, 1)` all agree, so mutating between them is a + provable equivalent HERE, not merely untested — scoped to this domain, + not a claim that `split`/`rsplit` are equivalent in general (a 200k-case + fuzz over the real `cortex_test_pw_` format confirms + agreement; an unrestricted fuzz allowing extra underscores anywhere + does NOT agree, which is why this equivalence is domain-scoped rather + than structural, unlike the `check_venv_lock_parity.py` regex case in + issue #287's PR discussion).""" + + def test_parses_the_pid_out_of_a_well_formed_name(self) -> None: + assert throwaway._orphan_pid("cortex_test_pw12345_ab12cd34") == 12345 + + def test_returns_none_for_a_non_numeric_suffix(self) -> None: + assert throwaway._orphan_pid("cortex_test_pw_not_a_pid") is None + + +class TestPidIsDefinitelyDead: + def test_calls_os_kill_with_signal_zero_on_the_given_pid(self) -> None: + """Signal 0 sends no actual signal — it only probes whether the pid + exists. `os.kill` is mocked here rather than exercised for real + (even with the correct, harmless signal 0): a mutation to the + hardcoded signal number would otherwise make an UNMOCKED call to + this same test send a real, possibly fatal, signal to the pytest + process itself when run against that mutant (observed: mutating + `0` to `1`/SIGHUP under mutmut left the run in a "suspicious", + not cleanly killed-or-survived, state). Asserting the exact call + pins both arguments at once: a pid swapped for the wrong value, or + the probe signal changed to a real one, are both caught here.""" + with mock.patch.object(throwaway.os, "kill") as mock_kill: + assert throwaway._pid_is_definitely_dead(4242) is False + mock_kill.assert_called_once_with(4242, 0) + + def test_process_lookup_error_means_dead(self) -> None: + with mock.patch.object(throwaway.os, "kill", side_effect=ProcessLookupError): + assert throwaway._pid_is_definitely_dead(999999) is True + + def test_permission_error_is_treated_as_alive(self) -> None: + """Ambiguous cases must never be treated as dead — dropping a + database another user's live process owns would be a false + positive the docstring explicitly refuses to risk.""" + with mock.patch.object(throwaway.os, "kill", side_effect=PermissionError): + assert throwaway._pid_is_definitely_dead(1) is False + + +class TestDropDeadOrphanedDatabases: + def _conn(self, rows): + conn = mock.MagicMock() + conn.execute.return_value.fetchall.return_value = rows + return conn + + def test_selects_with_the_exact_query(self) -> None: + conn = self._conn([]) + throwaway.drop_dead_orphaned_databases(conn) + conn.execute.assert_called_once_with( + "SELECT datname FROM pg_database WHERE datname LIKE 'cortex_test_pw%';" + ) + + def test_reads_the_datname_key_from_a_dict_style_row(self) -> None: + """psycopg can return dict-like rows (row_factory=dict_row) as well + as tuples; the `row.get("datname")` branch needs its own case, + not just the tuple-row one the other tests exercise. Reading the + WRONG key would make `name` fall back to `.get`'s default `None` + for a real dict that has no such key — falsy either way — so the + distinguishing observation is a DROP that only happens when the + correct key is actually read, not merely "does not raise".""" + conn = self._conn([{"datname": "cortex_test_pw999999_deadbeef"}]) + with mock.patch.object(throwaway, "_pid_is_definitely_dead", return_value=True): + throwaway.drop_dead_orphaned_databases(conn) + dropped = " ".join(str(call) for call in conn.execute.call_args_list) + assert 'DROP DATABASE IF EXISTS "cortex_test_pw999999_deadbeef"' in dropped + + def test_a_row_with_no_name_does_not_stop_processing_the_rest(self) -> None: + """A row whose name resolves to falsy (e.g. a dict-style row missing + the "datname" key entirely) must `continue`, never `break` — the + SAME invariant as `test_a_dead_row_does_not_stop_processing_the_rest` + but for the guard clause immediately after the name is read, not the + one after `_orphan_pid`.""" + conn = self._conn( + [ + {"unrelated_column": "x"}, # no "datname" key -> name is None + ("cortex_test_pw999999_deadbeef",), # a genuine, dead orphan + ] + ) + with mock.patch.object(throwaway, "_pid_is_definitely_dead", return_value=True): + throwaway.drop_dead_orphaned_databases(conn) + dropped = " ".join(str(call) for call in conn.execute.call_args_list) + assert "cortex_test_pw999999_deadbeef" in dropped + + def test_drops_a_database_whose_owning_pid_is_gone(self) -> None: + conn = self._conn([("cortex_test_pw999999_deadbeef",)]) + with mock.patch.object( + throwaway, "_pid_is_definitely_dead", return_value=True + ) as mock_is_dead: + throwaway.drop_dead_orphaned_databases(conn) + mock_is_dead.assert_called_once_with(999999) + calls = [call.args[0] for call in conn.execute.call_args_list] + assert ( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " + "WHERE datname = %s AND pid <> pg_backend_pid()" + ) in calls + terminate_call = next( + c + for c in conn.execute.call_args_list + if c.args[0].startswith("SELECT pg_terminate_backend") + ) + assert terminate_call.args[1] == ("cortex_test_pw999999_deadbeef",) + assert 'DROP DATABASE IF EXISTS "cortex_test_pw999999_deadbeef"' in calls + + def test_leaves_a_database_whose_owning_pid_is_alive(self) -> None: + conn = self._conn([(f"cortex_test_pw{os.getpid()}_deadbeef",)]) + throwaway.drop_dead_orphaned_databases(conn) + dropped = " ".join(str(call) for call in conn.execute.call_args_list) + assert "DROP DATABASE" not in dropped + + def test_a_dead_row_does_not_stop_processing_the_rest(self) -> None: + """A malformed/skippable row must `continue`, never `break` — one + bad row anywhere in the result set must not silently stop dropping + every orphan listed after it.""" + conn = self._conn( + [ + ("cortex_test_pw_not_a_pid",), # no numeric pid -> skip + ("cortex_test_pw999999_deadbeef",), # a genuine, dead orphan + ] + ) + with mock.patch.object(throwaway, "_pid_is_definitely_dead", return_value=True): + throwaway.drop_dead_orphaned_databases(conn) + dropped = " ".join(str(call) for call in conn.execute.call_args_list) + assert "cortex_test_pw999999_deadbeef" in dropped + + def test_a_selection_failure_is_swallowed(self) -> None: + conn = mock.MagicMock() + conn.execute.side_effect = Exception("connection reset") + throwaway.drop_dead_orphaned_databases(conn) # must not raise + + +class TestCreateAndDropIsolatedDatabase: + def _fake_psycopg(self, connect_side_effect=None): + fake = types.ModuleType("psycopg") + conn = mock.MagicMock() + conn.__enter__ = mock.Mock(return_value=conn) + conn.__exit__ = mock.Mock(return_value=False) + conn.execute.return_value.fetchall.return_value = [] + if connect_side_effect is not None: + fake.connect = mock.Mock(side_effect=connect_side_effect) + else: + fake.connect = mock.Mock(return_value=conn) + return fake, conn + + def test_returns_none_when_the_server_is_unreachable(self) -> None: + fake_pg, _conn = self._fake_psycopg( + connect_side_effect=Exception("connection refused") + ) + with mock.patch.dict(sys.modules, {"psycopg": fake_pg}): + assert ( + throwaway.create_isolated_test_database( + "postgresql://localhost/cortex_test" + ) + is None + ) + + def test_connects_to_the_maintenance_db_with_exact_args(self) -> None: + fake_pg, _conn = self._fake_psycopg() + with mock.patch.dict(sys.modules, {"psycopg": fake_pg}): + throwaway.create_isolated_test_database( + "postgresql://localhost/cortex_test" + ) + fake_pg.connect.assert_called_once_with( + "postgresql://localhost/postgres", autocommit=True, connect_timeout=3 + ) + + def test_reaps_orphans_on_the_same_connection_before_creating(self) -> None: + fake_pg, conn = self._fake_psycopg() + with ( + mock.patch.dict(sys.modules, {"psycopg": fake_pg}), + mock.patch.object(throwaway, "drop_dead_orphaned_databases") as mock_reap, + ): + throwaway.create_isolated_test_database( + "postgresql://localhost/cortex_test" + ) + mock_reap.assert_called_once_with(conn) + + def test_returns_the_isolated_url_and_owner_tuple_on_success(self) -> None: + fake_pg, conn = self._fake_psycopg() + with mock.patch.dict(sys.modules, {"psycopg": fake_pg}): + result = throwaway.create_isolated_test_database( + "postgresql://localhost/cortex_test" + ) + assert result is not None + isolated_url, maint_url, db_name = result + assert maint_url == "postgresql://localhost/postgres" + # cortex_test_pw_<8 lowercase hex chars> — the exact format + # _orphan_pid parses back out, and os.urandom(4).hex() is always 8 + # chars (4 bytes * 2 hex digits/byte). + assert re.fullmatch(r"cortex_test_pw\d+_[0-9a-f]{8}", db_name) + assert isolated_url == f"postgresql://localhost/{db_name}" + # conn.execute is also called once by drop_dead_orphaned_databases's + # own SELECT (asserted separately in TestDropDeadOrphanedDatabases); + # the CREATE DATABASE is always the LAST call this function makes. + conn.execute.assert_called_with(f'CREATE DATABASE "{db_name}"') + + def test_drop_isolated_database_swallows_any_failure(self) -> None: + fake_pg, _conn = self._fake_psycopg( + connect_side_effect=Exception("connection refused") + ) + with mock.patch.dict(sys.modules, {"psycopg": fake_pg}): + throwaway.drop_isolated_database( + "postgresql://x/postgres", "db1" + ) # no raise + + def test_drop_isolated_database_connects_with_exact_args(self) -> None: + fake_pg, _conn = self._fake_psycopg() + with mock.patch.dict(sys.modules, {"psycopg": fake_pg}): + throwaway.drop_isolated_database("postgresql://x/postgres", "db1") + fake_pg.connect.assert_called_once_with( + "postgresql://x/postgres", autocommit=True, connect_timeout=3 + ) + + def test_drop_isolated_database_terminates_and_drops_with_exact_sql(self) -> None: + fake_pg, conn = self._fake_psycopg() + with mock.patch.dict(sys.modules, {"psycopg": fake_pg}): + throwaway.drop_isolated_database("postgresql://x/postgres", "db1") + terminate_call, drop_call = conn.execute.call_args_list + assert terminate_call.args[0] == ( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " + "WHERE datname = %s AND pid <> pg_backend_pid()" + ) + assert terminate_call.args[1] == ("db1",) + assert drop_call.args[0] == 'DROP DATABASE IF EXISTS "db1"' diff --git a/tests_py/invariants/test_real_data_root_guard.py b/tests_py/invariants/test_real_data_root_guard.py new file mode 100644 index 00000000..9c06ffe9 --- /dev/null +++ b/tests_py/invariants/test_real_data_root_guard.py @@ -0,0 +1,213 @@ +"""The `guard_against_real_data_roots` guard itself cannot be removed or +weakened. + +Split out of test_store_isolation.py (issue #276/#287 boy-scout follow-up): +that file, together with the tests this session added, grew past +coding-standards.md §4.1's 300-line cap and needed decomposing along its +natural boundary — the RESOLVED-roots/live-write-path checks stayed in +test_store_isolation.py; the guard's own behavior and structural integrity +(can it be silently removed or its module-level call site weakened?) live +here. + +The guard checks resolved values so it cannot be fooled by a +correct-looking environment. These tests drive it against a deliberately +un-isolated root and assert it aborts the session, then pin that both the +guard function and its unconditional call site in conftest.py survive a +future refactor. +""" + +from __future__ import annotations + +import os +import unittest.mock as mock +from pathlib import Path + +import pytest + +_REAL_CLAUDE_DIR = Path(os.path.expanduser("~/.claude")).resolve() + +_CONFTEST_PATH = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "conftest.py") +) +_GUARDS_PATH = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "_pg_safety_guards.py") +) + + +class TestIsolationGuard: + """`guard_against_real_data_roots` must fire on a real root.""" + + def test_guard_exits_when_a_root_escapes_isolation(self): + """A WIKI_ROOT pointing at the real tree must abort with returncode=2. + + Postcondition: pytest.exit called once, returncode=2, message names + the offending root so an operator can see which one escaped. + """ + import tests_py.conftest as conftest + from mcp_server.infrastructure import config + from tests_py import _pg_safety_guards + + real_wiki = _REAL_CLAUDE_DIR / "methodology" / "wiki" + with ( + mock.patch.object(config, "WIKI_ROOT", real_wiki), + mock.patch.object(pytest, "exit") as mock_exit, + ): + _pg_safety_guards.guard_against_real_data_roots(conftest._TEST_CLAUDE_DIR) + + mock_exit.assert_called_once() + assert mock_exit.call_args[1].get("returncode") == 2 + assert "WIKI_ROOT" in mock_exit.call_args[0][0], ( + "Guard message does not name the offending root — an operator " + f"cannot tell what escaped: {mock_exit.call_args[0][0]!r}" + ) + + def test_guard_is_silent_when_isolation_holds(self): + """The live session is isolated, so the guard must NOT fire. + + Without this, a guard that always exits would pass the test above + while making the suite unrunnable. + """ + import tests_py.conftest as conftest + from tests_py import _pg_safety_guards + + with mock.patch.object(pytest, "exit") as mock_exit: + _pg_safety_guards.guard_against_real_data_roots(conftest._TEST_CLAUDE_DIR) + mock_exit.assert_not_called() + + def test_guard_exit_message_is_exact_with_two_offenders(self): + """Exact equality, with 2+ offenders so the join separator between + them is exercised (a single offender can't distinguish a real "\\n" + from a corrupted one). `expanduser`/`realpath` on "~/.claude" run + for REAL, not mocked: a fixed `return_value` mock would return the + same string for the correct literal and a mutated one alike.""" + from tests_py import _pg_safety_guards + + fake_offenders = [ + ("CLAUDE_DIR", "/real/claude"), + ("WIKI_ROOT", "/real/claude/wiki"), + ] + expected_real_root = os.path.realpath(os.path.expanduser("~/.claude")) + with ( + mock.patch.object( + _pg_safety_guards, + "_resolved_real_data_roots", + return_value=fake_offenders, + ), + mock.patch.object(pytest, "exit") as mock_exit, + ): + _pg_safety_guards.guard_against_real_data_roots("/isolated/tree") + + mock_exit.assert_called_once_with( + "REFUSING to run: test isolation is not in effect. These roots " + "still resolve outside the throwaway tree /isolated/tree:\n" + " CLAUDE_DIR -> /real/claude\n" + " WIKI_ROOT -> /real/claude/wiki\n" + f"(real user root: {expected_real_root}). The suite DELETEs " + "every table between tests and regenerates wiki dashboards — " + "running it against a real tree destroys data (issue #219). " + "Ensure _redirect_real_data_roots() runs before any mcp_server " + "import.", + returncode=2, + ) + + def test_guard_has_no_bypass_env_var(self): + """Unlike the PG guard, this one must offer no override. + + A suite that DELETEs every table and regenerates wiki pages has no + safe way to run against a real tree, so an escape hatch would only + ever be used to re-enable the incident. + """ + with open(_GUARDS_PATH, encoding="utf-8") as fh: + lines = fh.readlines() + + start = next( + i + for i, line in enumerate(lines) + if line.startswith("def guard_against_real_data_roots(") + ) + end = next( + ( + i + for i in range(start + 1, len(lines)) + if lines[i] and lines[i][0] not in (" ", "\t", "\n") + ), + len(lines), + ) + body = "".join(lines[start:end]) + assert "CORTEX_TEST_ALLOW" not in body, ( + "An override env-var was added to the isolation guard. There is " + "no safe way to run this suite against a real data tree." + ) + + +class TestGuardStructuralIntegrity: + """The redirection and the guard must both stay unconditional.""" + + def test_redirect_runs_at_module_level(self): + """`_redirect_real_data_roots()` must be called at zero indentation. + + Inside an `if`, it reintroduces exactly the conditional-isolation bug + of issue #219. + """ + with open(_CONFTEST_PATH, encoding="utf-8") as fh: + lines = fh.readlines() + + calls = [ + line + for line in lines + if line.strip().endswith("_redirect_real_data_roots()") + and not line.startswith((" ", "\t")) + ] + assert calls, ( + "conftest.py no longer calls _redirect_real_data_roots() at " + "module level — real-data roots are unisolated again (#219)." + ) + + def test_guard_call_present_at_module_level(self): + with open(_CONFTEST_PATH, encoding="utf-8") as fh: + lines = fh.readlines() + + calls = [ + line + for line in lines + if line.strip().startswith( + "_pg_safety_guards.guard_against_real_data_roots(" + ) + and not line.startswith((" ", "\t")) + ] + assert calls, ( + "conftest.py no longer calls " + "_pg_safety_guards.guard_against_real_data_roots(...) at module " + "level — isolation failures become silent again (#219)." + ) + + def test_redirect_precedes_first_mcp_server_import(self): + """Ordering is load-bearing: constants bind at import time. + + `from mcp_server.infrastructure.config import METHODOLOGY_DIR` copies + a value. If any mcp_server import runs before the env var is set, the + copied value is the real path and no later env change can fix it. + """ + with open(_CONFTEST_PATH, encoding="utf-8") as fh: + lines = fh.readlines() + + redirect_lines = [ + i + for i, line in enumerate(lines) + if line.strip().endswith("_redirect_real_data_roots()") + and not line.startswith((" ", "\t")) + ] + assert redirect_lines, ( + "conftest.py has no module-level _redirect_real_data_roots() " + "call, so nothing redirects the real-data roots at all (#219)." + ) + redirect_line = redirect_lines[0] + early_imports = [ + (i + 1, line.strip()) + for i, line in enumerate(lines[:redirect_line]) + if line.startswith(("import mcp_server", "from mcp_server")) + ] + assert not early_imports, ( + "mcp_server is imported before the roots are redirected, so path " + f"constants bind to the real tree: {early_imports}" + ) diff --git a/tests_py/invariants/test_store_isolation.py b/tests_py/invariants/test_store_isolation.py index 2640d863..f404de07 100644 --- a/tests_py/invariants/test_store_isolation.py +++ b/tests_py/invariants/test_store_isolation.py @@ -20,10 +20,13 @@ Both roots derive from `config.CLAUDE_DIR`, which had no override seam at all. The fix adds `CORTEX_CLAUDE_DIR` and redirects it unconditionally. -These tests pin that fix from three angles, so a regression cannot be -silent: the RESOLVED roots are isolated (not merely the env vars), a live -write-path handler leaves the real tree untouched, and the conftest guard -that enforces this cannot be deleted or made conditional. +These tests pin that fix from two angles: the RESOLVED roots are isolated +(not merely the env vars), and a live write-path handler leaves the real +tree untouched. The guard itself — `guard_against_real_data_roots`, +whether it can be silently removed or its module-level call site +weakened — has its own file, `test_real_data_root_guard.py` (issue +#276/#287 boy-scout follow-up, splitting this file under +coding-standards.md §4.1's 300-line cap). """ from __future__ import annotations @@ -36,10 +39,6 @@ _REAL_CLAUDE_DIR = Path(os.path.expanduser("~/.claude")).resolve() -_CONFTEST_PATH = os.path.abspath( - os.path.join(os.path.dirname(__file__), "..", "conftest.py") -) - def _real(path: str | Path) -> str: return os.path.realpath(os.path.expanduser(str(path))) @@ -204,147 +203,39 @@ async def test_consolidate_does_not_touch_the_real_sqlite_store(self): ) -# ── The guard itself cannot be removed or weakened ─────────────────────────── - - -class TestIsolationGuard: - """`_guard_against_real_data_roots` must fire on a real root. - - The guard checks resolved values so it cannot be fooled by a - correct-looking environment. These tests drive it against a deliberately - un-isolated root and assert it aborts the session. - """ - - def test_guard_exits_when_a_root_escapes_isolation(self): - """A WIKI_ROOT pointing at the real tree must abort with returncode=2. - - Postcondition: pytest.exit called once, returncode=2, message names - the offending root so an operator can see which one escaped. - """ - import tests_py.conftest as conftest - from mcp_server.infrastructure import config - - real_wiki = _REAL_CLAUDE_DIR / "methodology" / "wiki" - with ( - mock.patch.object(config, "WIKI_ROOT", real_wiki), - mock.patch.object(pytest, "exit") as mock_exit, - ): - conftest._guard_against_real_data_roots() - - mock_exit.assert_called_once() - assert mock_exit.call_args[1].get("returncode") == 2 - assert "WIKI_ROOT" in mock_exit.call_args[0][0], ( - "Guard message does not name the offending root — an operator " - f"cannot tell what escaped: {mock_exit.call_args[0][0]!r}" - ) - - def test_guard_is_silent_when_isolation_holds(self): - """The live session is isolated, so the guard must NOT fire. - - Without this, a guard that always exits would pass the test above - while making the suite unrunnable. - """ - import tests_py.conftest as conftest +# ── The exact set of roots the guard checks ────────────────────────────────── - with mock.patch.object(pytest, "exit") as mock_exit: - conftest._guard_against_real_data_roots() - mock_exit.assert_not_called() - def test_guard_has_no_bypass_env_var(self): - """Unlike the PG guard, this one must offer no override. +class TestResolvedRealDataRoots: + """`_pg_safety_guards._resolved_real_data_roots` names the exact four + roots `guard_against_real_data_roots` checks. A wrong label here would + still let the guard fire correctly but tell an operator the wrong name + for which root escaped.""" - A suite that DELETEs every table and regenerates wiki pages has no - safe way to run against a real tree, so an escape hatch would only - ever be used to re-enable the incident. - """ - with open(_CONFTEST_PATH, encoding="utf-8") as fh: - lines = fh.readlines() + def test_returns_the_four_named_roots_in_order(self): + from mcp_server.infrastructure import config, memory_config + from tests_py import _pg_safety_guards - start = next( - i - for i, line in enumerate(lines) - if line.startswith("def _guard_against_real_data_roots(") + fake_settings = mock.Mock( + DB_PATH="/fake/db.sqlite", + SQLITE_FALLBACK_PATH="/fake/fallback.sqlite", ) - end = next( - ( - i - for i in range(start + 1, len(lines)) - if lines[i] and lines[i][0] not in (" ", "\t", "\n") + with ( + mock.patch.object(config, "CLAUDE_DIR", "/fake/claude"), + mock.patch.object(config, "WIKI_ROOT", "/fake/claude/wiki"), + mock.patch.object( + memory_config, "get_memory_settings", return_value=fake_settings ), - len(lines), - ) - body = "".join(lines[start:end]) - assert "CORTEX_TEST_ALLOW" not in body, ( - "An override env-var was added to the isolation guard. There is " - "no safe way to run this suite against a real data tree." - ) - - -class TestGuardStructuralIntegrity: - """The redirection and the guard must both stay unconditional.""" - - def test_redirect_runs_at_module_level(self): - """`_redirect_real_data_roots()` must be called at zero indentation. - - Inside an `if`, it reintroduces exactly the conditional-isolation bug - of issue #219. - """ - with open(_CONFTEST_PATH, encoding="utf-8") as fh: - lines = fh.readlines() - - calls = [ - line - for line in lines - if line.strip().endswith("_redirect_real_data_roots()") - and not line.startswith((" ", "\t")) - ] - assert calls, ( - "conftest.py no longer calls _redirect_real_data_roots() at " - "module level — real-data roots are unisolated again (#219)." - ) - - def test_guard_call_present_at_module_level(self): - with open(_CONFTEST_PATH, encoding="utf-8") as fh: - lines = fh.readlines() - - calls = [ - line - for line in lines - if line.strip() == "_guard_against_real_data_roots()" - and not line.startswith((" ", "\t")) - ] - assert calls, ( - "conftest.py no longer calls _guard_against_real_data_roots() at " - "module level — isolation failures become silent again (#219)." - ) - - def test_redirect_precedes_first_mcp_server_import(self): - """Ordering is load-bearing: constants bind at import time. - - `from mcp_server.infrastructure.config import METHODOLOGY_DIR` copies - a value. If any mcp_server import runs before the env var is set, the - copied value is the real path and no later env change can fix it. - """ - with open(_CONFTEST_PATH, encoding="utf-8") as fh: - lines = fh.readlines() - - redirect_lines = [ - i - for i, line in enumerate(lines) - if line.strip().endswith("_redirect_real_data_roots()") - and not line.startswith((" ", "\t")) - ] - assert redirect_lines, ( - "conftest.py has no module-level _redirect_real_data_roots() " - "call, so nothing redirects the real-data roots at all (#219)." - ) - redirect_line = redirect_lines[0] - early_imports = [ - (i + 1, line.strip()) - for i, line in enumerate(lines[:redirect_line]) - if line.startswith(("import mcp_server", "from mcp_server")) - ] - assert not early_imports, ( - "mcp_server is imported before the roots are redirected, so path " - f"constants bind to the real tree: {early_imports}" - ) + ): + assert _pg_safety_guards._resolved_real_data_roots() == [ + ("CLAUDE_DIR", "/fake/claude"), + ("WIKI_ROOT", "/fake/claude/wiki"), + ("MemorySettings.DB_PATH", "/fake/db.sqlite"), + ("MemorySettings.SQLITE_FALLBACK_PATH", "/fake/fallback.sqlite"), + ] + + +# The guard itself (TestIsolationGuard, TestGuardStructuralIntegrity) moved +# to test_real_data_root_guard.py — issue #276/#287 boy-scout follow-up, +# splitting this file under coding-standards.md §4.1's 300-line cap (see +# that file's docstring for the size-cap rationale).