"""격리 near-miss (2026-09-02) 를 문으로 잠근다. `scenario()` 가 배경 술어를 물으며 `app.core.config` 를 **env 갱신 전에** 올렸다. settings 는 그 순간의 .env(원본 DB URL) 로 굳고, 뒤에 flip 검사가 `app.core.steps` → `app.core.database` 를 올려 `SessionLocal` 이 **원본에** 붙었다. 부트스트랩이 canary 프로젝트를 못 찾아 선 것이 유일한 방어였다. 세 문: ①env 가 settings 보다 먼저(main) ②flip 검사는 자식 프로세스(in-process import 0) ③부트스트랩·pipeline 직전에 **실제 엔진 URL** 을 본다. """ from __future__ import annotations import ast import inspect import subprocess import sys import textwrap from pathlib import Path import pytest from tools.grounding_audit import canary_isolation as ci from tools.grounding_audit import canary_run as cr BACKEND = Path(__file__).resolve().parents[2] class TestNoInProcessDatabaseImport: def test_scenario_and_flip_scan_leave_the_database_module_unloaded(self): """★깨끗한 인터프리터에서 — 이 시험 프로세스는 conftest 가 이미 올려 뒀다.""" code = textwrap.dedent(""" import sys from tools.grounding_audit import canary_run as cr cr.scenario(mode="v2_chunk", fixture="period_episode", target="world_guide") cr.steps_folding_setting("background_mode") print("DB_LOADED", "app.core.database" in sys.modules) """) got = subprocess.run([sys.executable, "-c", code], cwd=str(BACKEND), capture_output=True, text=True, timeout=180) assert got.returncode == 0, got.stderr[-600:] assert "DB_LOADED False" in got.stdout, got.stdout def test_the_flip_scan_still_sees_the_registry(self): got = cr.steps_folding_setting("background_mode") assert "floor_plan_render" in got and "shot_director" not in got class TestMainPreparesTheEnvBeforeSettings: def test_prepare_env_comes_before_scenario_in_main(self): src = textwrap.dedent(inspect.getsource(cr.main)) tree = ast.parse(src) order = [] for node in ast.walk(tree): if isinstance(node, ast.Call): name = ast.unparse(node.func) if name.endswith("prepare_env") or name == "scenario" \ or name.endswith("assert_app_modules_not_loaded"): order.append((node.lineno, name)) order.sort() names = [n for _, n in order] assert names.index("cbs.prepare_env") < names.index("scenario"), names assert names.index("ci.assert_app_modules_not_loaded") < names.index("scenario") class TestTheSettingsAndEngineGuards: def test_settings_guard_passes_when_settings_follow_the_env(self, monkeypatch): from app.core.config import settings monkeypatch.setattr(ci, "db_url", lambda _rid: "postgresql://u:p@h/theroad_canary_x") monkeypatch.setattr(settings, "database_url", "postgresql://u:p@h/theroad_canary_x") assert ci.assert_settings_follow_env("x")["checked"] is True def test_settings_guard_refuses_a_foreign_settings(self, monkeypatch): """★양성 대조 — settings 가 원본을 보면 선다.""" from app.core.config import settings monkeypatch.setattr(ci, "db_url", lambda _rid: "postgresql://u:p@h/theroad_canary_x") monkeypatch.setattr(settings, "database_url", "postgresql://u:p@h/theroad") with pytest.raises(ci.IsolationRefused, match="settings"): ci.assert_settings_follow_env("x") def _real(self): import app.core.database as dbm from sqlalchemy.engine import make_url u = make_url(str(dbm.engine.url)) return u.host, u.port, u.database def test_engine_guard_reads_the_real_engine(self, monkeypatch): """★settings·engine·SessionLocal bind 셋이 다 같은 곳이면 지난다.""" from app.core.config import settings host, port, db = self._real() same = f"postgresql://u:p@{host}:{port}/{db}" monkeypatch.setattr(settings, "database_url", same) monkeypatch.setattr(ci, "db_url", lambda _rid: same) got = ci.assert_engine_is_canary("x") assert got["checked"] is True and got["database"] == db assert got["compared"] == ["engine", "session_bind", "settings"] def test_engine_guard_refuses_a_foreign_engine(self, monkeypatch): """★양성 대조 — env·settings 가 canary 라 해도 엔진·bind 가 다르면 선다.""" from app.core.config import settings host, port, _db = self._real() other = f"postgresql://u:p@{host}:{port}/theroad_canary_x" monkeypatch.setattr(settings, "database_url", other) monkeypatch.setattr(ci, "db_url", lambda _rid: other) with pytest.raises(ci.IsolationRefused, match="붙어 있다") as exc: ci.assert_engine_is_canary("x") assert "engine" in str(exc.value) and "session_bind" in str(exc.value) def test_the_comparison_is_by_identity_not_by_string(self): """★비밀번호 마스킹 문자열 비교가 아니다 — host·port·database 만.""" a = ci._db_identity("postgresql://u:secret@Localhost:5432/theroad_canary_x") b = ci._db_identity("postgresql://u:***@localhost:5432/theroad_canary_x") assert a == b == {"host": "localhost", "port": "5432", "database": "theroad_canary_x"} def test_run_checks_the_engine_before_bootstrap_and_pipeline(self): src = textwrap.dedent(inspect.getsource(cr.run)) i_guard1 = src.index('got["stages"]["engine_before_bootstrap"]') i_boot = src.index("boot = cbs.bootstrap(") i_guard2 = src.index('got["stages"]["engine_before_pipeline"]') i_pipe = src.index("cp.run_pipeline(") assert i_guard1 < i_boot and i_guard2 < i_pipe class TestImportingTheCanaryDoesNotLoadSettings: """★근본 원인 — `canary_run` import 가 `app.core.config` 를 올리면 그 뒤 env 를 갈아도 settings 는 원본 URL 이다. 지난 tip 은 안 올렸고 오늘 올렸다.""" def test_a_clean_interpreter_imports_canary_run_without_config(self): code = ("import sys, tools.grounding_audit.canary_run\n" "print('CFG', 'app.core.config' in sys.modules, " "'DB', 'app.core.database' in sys.modules)") got = subprocess.run([sys.executable, "-c", code], cwd=str(BACKEND), capture_output=True, text=True, timeout=120) assert got.returncode == 0, got.stderr[-400:] assert "CFG False DB False" in got.stdout, got.stdout class TestTheWrongOrderIsRefusedAndTheRightOrderReuses: """Codex 조건 4 — 잘못 묶인 자식은 provider 0·원본 쓰기 0 에서 **먼저** 서고, 올바른 자식만 canary DB 의 기존 프로젝트를 되쓴다.""" RID = "69e821758f3d" def _child(self, code: str, *, env: dict) -> subprocess.CompletedProcess: return subprocess.run([sys.executable, "-c", textwrap.dedent(code)], cwd=str(BACKEND), capture_output=True, text=True, timeout=180, env=env) def test_settings_loaded_before_env_is_refused_before_any_db_use(self): import os code = """ import os, sys import app.core.config # ★env 보다 먼저 — 잘못된 순서 from tools.grounding_audit import canary_bootstrap as cbs, canary_isolation as ci os.environ.update(cbs.prepare_env("%s", grounding_mode="v2_chunk")) try: ci.assert_app_modules_not_loaded() print("PASSED_GUARD") except ci.IsolationRefused as e: print("REFUSED", str(e)[:80]) try: ci.assert_engine_is_canary("%s") print("ENGINE_OK") except ci.IsolationRefused as e: print("ENGINE_REFUSED") """ % (self.RID, self.RID) got = self._child(code, env=dict(os.environ)) assert got.returncode == 0, got.stderr[-600:] assert "REFUSED" in got.stdout and "PASSED_GUARD" not in got.stdout, got.stdout assert "ENGINE_REFUSED" in got.stdout, got.stdout def test_env_first_then_import_reuses_the_canary_project(self): import os from tools.grounding_audit import canary_isolation as ci if not (ci.root_dir(self.RID) / "_bootstrap_journal.json").is_file(): pytest.skip("canary 산출이 이 기계에 없다") code = """ import json, os from tools.grounding_audit import canary_bootstrap as cbs, canary_isolation as ci os.environ.update(cbs.prepare_env("%s", grounding_mode="v2_chunk")) ci.assert_app_modules_not_loaded() got = ci.assert_engine_is_canary("%s") print("ENGINE", got["database"]) from app.core.database import SessionLocal jr = json.load(open(str(ci.root_dir("%s") / "_bootstrap_journal.json"))) hit = jr["calls"][0]["response"] print("RESUME", bool(cbs._resume(SessionLocal(), hit))) """ % (self.RID, self.RID, self.RID) got = self._child(code, env=dict(os.environ)) assert got.returncode == 0, got.stderr[-600:] assert f"ENGINE theroad_canary_{self.RID}" in got.stdout, got.stdout assert "RESUME True" in got.stdout, got.stdout