"""canary 를 원본에서 **떼어 놓는** 자물쇠. ★유료 0 · 원본에 한 글자도 안 쓴다.

Codex 2026-08-31 —

> 같은 theroad DB 에 canary project_id 를 넣지 마십시오. 별도 PostgreSQL
> database 를 run-id 이름으로 만드십시오. `DATABASE_URL`/engine 가 prod DB 와
> 다르고 DB name 에 canary run_id 가 있는지 **import·SessionLocal 생성 전에**
> fail-closed. 자동 cleanup/drop 금지.

★**파괴적 작업 이력이 있다.** 그래서 이 시험은 「막는가」를 하나씩 잰다.
"""
from __future__ import annotations

import ast
import inspect
from pathlib import Path

import pytest

from tools.grounding_audit import canary_isolation as ci

RID = "a1b2c3d4"
#: ★접속 정보를 **시험에도 안 박는다** — 본을 주는 것이 계약이다
TEMPLATE = "postgresql+psycopg2://사용자:암호@어딘가:5432/theroad?sslmode=disable"


def _env(**over):
    e = {"DATABASE_URL": ci.db_url(RID, template=TEMPLATE),
         "PROJECTS_DIR": str(ci.root_dir(RID) / "projects")}
    e.update(over)
    return {k: v for k, v in e.items() if v is not None}


class TestAProperSetupPasses:
    """★양성 대조 — 막기만 하고 못 지나가면 무엇을 재는지 모른다."""

    def test_the_canary_env_is_accepted(self):
        got = ci.assert_isolated(RID, env=_env())
        assert got["ok"] is True
        assert got["db_name"] == f"theroad_canary_{RID}"
        assert RID in got["projects_dir"]

    def test_the_trace_coordinates_are_canary_only(self):
        got = ci.assert_isolated(RID, env=_env())
        assert "canary" in got["trace_tag"] and "canary" in got["trace_thread"]


class TestItRefusesToTouchTheRealThing:
    def test_no_database_url_stops(self):
        with pytest.raises(ci.IsolationRefused):
            ci.assert_isolated(RID, env=_env(DATABASE_URL=None))

    def test_the_production_database_stops(self):
        url = "postgresql://사용자:암호@어딘가:5432/theroad"
        with pytest.raises(ci.IsolationRefused) as e:
            ci.assert_isolated(RID, env=_env(DATABASE_URL=url))
        assert ci.PROD_DB_NAME in str(e.value)

    def test_a_database_without_the_run_id_stops(self):
        url = "postgresql://사용자:암호@어딘가:5432/theroad_canary_zzzz"
        with pytest.raises(ci.IsolationRefused):
            ci.assert_isolated(RID, env=_env(DATABASE_URL=url))

    def test_no_projects_dir_stops(self):
        with pytest.raises(ci.IsolationRefused):
            ci.assert_isolated(RID, env=_env(PROJECTS_DIR=None))

    def test_the_real_projects_dir_stops(self, tmp_path):
        prod = tmp_path / "projects"
        (prod / RID).mkdir(parents=True)
        with pytest.raises(ci.IsolationRefused) as e:
            ci.assert_isolated(RID, env=_env(PROJECTS_DIR=str(prod / RID)),
                               prod_projects_dir=str(prod))
        assert "원본 자리" in str(e.value)

    def test_a_dir_outside_this_runs_root_stops(self, tmp_path):
        other = tmp_path / f"어딘가-{RID}"
        other.mkdir()
        with pytest.raises(ci.IsolationRefused):
            ci.assert_isolated(RID, env=_env(PROJECTS_DIR=str(other)))

    @pytest.mark.parametrize("bad", ["", "..", "theroad", "XYZ", "a1b2",
                                     "../../etc", "a" * 33, "A1B2C3D4"])
    def test_a_junk_run_id_stops(self, bad):
        """★이름이 DB 이름에 그대로 들어간다 — 꼴을 좁혀 둔다."""
        with pytest.raises(ci.IsolationRefused):
            ci.db_name(bad)


class TestTheLockComesBeforeTheConnection:
    def test_it_notices_the_module_already_loaded(self, monkeypatch):
        """★`SessionLocal` 은 **import 때** 만들어진다."""
        import sys

        monkeypatch.setitem(sys.modules, "app.core.database", object())
        with pytest.raises(ci.IsolationRefused):
            ci.assert_database_module_not_loaded()

    def test_it_passes_when_not_loaded(self, monkeypatch):
        import sys

        monkeypatch.delitem(sys.modules, "app.core.database", raising=False)
        ci.assert_database_module_not_loaded()


class TestThereIsNoWayToDeleteFromHere:
    def test_the_module_has_no_drop(self):
        """★★지우는 길이 **없다**. 주행 뒤 DB 를 그대로 둔다 (Codex).

        ★글자가 아니라 **AST 로** 본다 — 설명을 적은 주석이 걸리면 안 된다.
        """
        tree = ast.parse(inspect.getsource(ci))
        sql = []
        for n in ast.walk(tree):
            if isinstance(n, ast.Constant) and isinstance(n.value, str):
                sql.append(n.value.upper())
            elif isinstance(n, ast.JoinedStr):
                sql.append("".join(
                    v.value.upper() for v in n.values
                    if isinstance(v, ast.Constant)
                    and isinstance(v.value, str)))
        # ★문자열 **리터럴**에만 본다. docstring 도 리터럴이므로 아래는
        #  「DROP 이라는 낱말이 SQL 로 쓰였나」를 본다 — 설명에는 안 쓴다.
        assert not [s for s in sql if s.strip().startswith("DROP ")]
        assert not [s for s in sql if "TRUNCATE" in s or "DELETE FROM" in s]

    def test_creating_is_add_only(self):
        """★이미 있으면 **그대로 쓴다** — 덮어쓰지 않는다."""
        src = inspect.getsource(ci.create_database)
        tree = ast.parse(src.lstrip())
        made = [n for n in ast.walk(tree)
                if isinstance(n, ast.JoinedStr)]
        joined = "".join(v.value for n in made for v in n.values
                         if isinstance(v, ast.Constant)
                         and isinstance(v.value, str))
        assert "CREATE DATABASE" in joined.upper()
        assert "DROP" not in joined.upper()


class TestTheRootHoldsEverything:
    def test_outputs_and_journal_live_under_the_run_root(self):
        root = ci.root_dir(RID)
        assert root.name == f"canary_{RID}"
        assert Path(ci.assert_isolated(RID, env=_env())["projects_dir"]) \
            .is_relative_to(root)


class TestNoConnectionDetailsAreBakedIn:
    """★★Codex BLOCK 4 — 하드코딩 URL·비밀번호를 지웠다."""

    def test_without_an_env_var_it_falls_back_to_dotenv(self, monkeypatch):
        """★사람에게 묻기 전에 `.env` 에서 파생한다 — ★값은 안 찍는다."""
        monkeypatch.delenv("THEROAD_CANARY_TEMPLATE_URL", raising=False)
        got = ci.db_url(RID)
        assert got.endswith(ci.db_name(RID))

    def test_with_neither_it_stops(self, monkeypatch):
        monkeypatch.delenv("THEROAD_CANARY_TEMPLATE_URL", raising=False)
        monkeypatch.setattr(ci, "_dotenv_database_url", lambda: None)
        with pytest.raises(ci.IsolationRefused):
            ci.db_url(RID)

    def test_the_url_is_built_by_the_parser_not_by_string_surgery(self):
        """★query·escaping 이 살아남아야 한다."""
        got = ci.db_url(RID, template=TEMPLATE)
        assert got.endswith("?sslmode=disable")
        assert f"/theroad_canary_{RID}?" in got
        assert "postgresql+psycopg2://" in got

    def test_the_module_has_no_password(self):
        """★소스에 접속 정보가 남아 있지 않다 (AST 문자열만 본다)."""
        import ast
        import inspect

        tree = ast.parse(inspect.getsource(ci))
        lits = [n.value for n in ast.walk(tree)
                if isinstance(n, ast.Constant) and isinstance(n.value, str)]

        def _is_credential(x: str) -> bool:
            """★글자가 아니라 **뜻**으로 본다 — 진짜 접속 주소인가.

            앞 판은 `"://" in x and "@" in x` 로 봐서 **가리는 함수의
            정규식**을 위반으로 읽었다
            ([[feedback-my-guard-caught-its-own-explanation]]).
            """
            from sqlalchemy.engine import make_url

            try:
                u = make_url(x)
            except Exception:               # noqa: BLE001
                return False                # 파싱 안 되면 주소가 아니다
            return bool(u.password) or bool(u.host and u.username)

        bad = [x for x in lits if _is_credential(x)]
        assert bad == [], f"★접속 정보가 박혀 있다: {bad}"


class TestTheAdminConnectionIsGated:
    def test_the_admin_dsn_is_derived_not_asked_for(self, monkeypatch):
        """★★사람에게 안 묻는다 — 본에서 **database 만** 바꾼다."""
        monkeypatch.delenv("THEROAD_CANARY_ADMIN_DSN", raising=False)
        got = ci.maintenance_dsn(explicit=None)
        assert got.rsplit("/", 1)[-1].split("?")[0] == "postgres"

    def test_deriving_stops_when_there_is_no_source(self, monkeypatch):
        monkeypatch.delenv("THEROAD_CANARY_ADMIN_DSN", raising=False)
        monkeypatch.delenv("THEROAD_CANARY_TEMPLATE_URL", raising=False)
        monkeypatch.setattr(ci, "_dotenv_database_url", lambda: None)
        with pytest.raises(ci.IsolationRefused):
            ci.maintenance_dsn()

    def test_an_admin_pointing_at_production_stops(self):
        with pytest.raises(ci.IsolationRefused) as e:
            ci.create_database(
                RID, admin_dsn="postgresql://u:p@h:5432/theroad")
        assert ci.PROD_DB_NAME in str(e.value)

    def test_an_admin_that_is_not_the_maintenance_db_stops(self):
        with pytest.raises(ci.IsolationRefused) as e:
            ci.create_database(
                RID, admin_dsn="postgresql://u:p@h:5432/무언가")
        assert "maintenance" in str(e.value)

    def test_it_verifies_current_database_after_connecting(self):
        """★★DSN 이 거짓일 수 있다 — 붙고 나서 **다시** 본다."""
        import ast
        import inspect

        tree = ast.parse(inspect.getsource(ci.create_database).lstrip())
        lits = [n.value for n in ast.walk(tree)
                if isinstance(n, ast.Constant) and isinstance(n.value, str)]
        assert any("current_database()" in x for x in lits), \
            "★붙은 뒤 확인을 안 한다"


class TestAlembicGoesThroughTheSameGate:
    def _env(self):
        return {"DATABASE_URL": ci.db_url(RID, template=TEMPLATE),
                "PROJECTS_DIR": str(ci.root_dir(RID) / "projects"),
                "PATH": "/usr/bin", "HOME": "/tmp"}

    def test_only_the_allowlist_is_handed_down(self):
        e = {**self._env(), "OPENAI_API_KEY": "비밀", "무엇": "그밖"}
        got = ci.child_env(RID, env=e)
        assert set(got) <= set(ci.ENV_ALLOWLIST)
        assert "OPENAI_API_KEY" not in got and "무엇" not in got

    def test_the_production_pointers_never_travel(self):
        e = {**self._env(),
             "THEROAD_PROD_DATABASE_URL": "postgresql://u:p@h/theroad"}
        got = ci.child_env(RID, env=e)
        assert not [k for k in ci.ENV_DENY if k in got]

    def test_a_bad_env_stops_before_building_the_child(self):
        e = {**self._env(), "DATABASE_URL": "postgresql://u:p@h:5432/theroad"}
        with pytest.raises(ci.IsolationRefused):
            ci.child_env(RID, env=e)

    def test_the_migration_connection_is_checked_inside(self):
        """★★`env.py` 가 URL 을 덮었을 수 있다 — 붙은 뒤 다시 본다."""
        ci.assert_migration_connection(RID, current_database=ci.db_name(RID))
        with pytest.raises(ci.IsolationRefused) as e:
            ci.assert_migration_connection(RID, current_database="theroad")
        assert "env.py" in str(e.value)

    def test_a_half_applied_schema_stops(self):
        ci.assert_at_head(alembic_version="abc123", head="abc123")
        for bad in ("", None, "옛판"):
            with pytest.raises(ci.IsolationRefused):
                ci.assert_at_head(alembic_version=bad, head="abc123")

    def test_the_command_runs_the_gated_entry_not_bare_alembic(self):
        """★★`alembic upgrade head` 를 바로 부르면 확인 함수가 **아무 데서도
        안 불린다** (Codex 2026-08-31: 「만들고 안 읽음」)."""
        cmd = ci.upgrade_command(RID)
        assert cmd[0].endswith("python") or "python" in cmd[0]
        assert cmd[1].endswith("canary_alembic.py")
        assert cmd[2] == RID


class TestAlembicEnvActuallyCallsTheGate:
    """★★Codex BLOCK 1 — 「함수는 있다」와 「그 문을 지난다」는 다르다."""

    def _env_module(self):
        import importlib.util
        from pathlib import Path

        src = Path("alembic/env.py").read_text(encoding="utf-8")
        return src

    def test_the_gate_is_called_before_run_migrations(self):
        """★AST 로 본다 — 확인이 `context.run_migrations()` **앞**이어야 한다."""
        import ast

        tree = ast.parse(self._env_module())
        fn = next(n for n in ast.walk(tree)
                  if isinstance(n, ast.FunctionDef)
                  and n.name == "run_migrations_online")
        order = []
        for n in ast.walk(fn):
            if isinstance(n, ast.Call):
                if isinstance(n.func, ast.Name):
                    order.append((n.lineno, n.func.id))
                elif isinstance(n.func, ast.Attribute):
                    order.append((n.lineno, n.func.attr))
        gate = [ln for ln, name in order
                if name == "_assert_canary_db_if_asked"]
        run = [ln for ln, name in order if name == "run_migrations"]
        assert gate, "★env.py 가 문을 안 부른다 — 함수만 있는 것이다"
        assert run and min(gate) < min(run), "★확인이 migration 뒤에 있다"

    def _call_gate(self, monkeypatch, *, run_id, current):
        import importlib

        env = importlib.import_module("tools.grounding_audit.canary_isolation")
        ns: dict = {}
        import ast
        tree = ast.parse(self._env_module())
        fn = next(n for n in ast.walk(tree)
                  if isinstance(n, ast.FunctionDef)
                  and n.name == "_assert_canary_db_if_asked")
        exec(compile(ast.Module([fn], []), "<env>", "exec"), ns)

        class _Res:
            def scalar(self_inner):
                return current

        class _Con:
            def execute(self_inner, *a, **k):
                return _Res()

            def rollback(self_inner):
                """★문이 연 트랜잭션을 닫는다 — 안 닫으면 이관이 롤백된다."""

        if run_id is None:
            monkeypatch.delenv("THEROAD_CANARY_RUN_ID", raising=False)
        else:
            monkeypatch.setenv("THEROAD_CANARY_RUN_ID", run_id)
        assert env is not None
        return ns["_assert_canary_db_if_asked"](_Con())

    def test_a_wrong_database_stops_before_any_sql(self, monkeypatch):
        """★★딴 DB 면 migration SQL 이 **한 줄도** 안 나간다."""
        with pytest.raises(ci.IsolationRefused):
            self._call_gate(monkeypatch, run_id=RID, current="theroad")

    def test_the_right_database_passes(self, monkeypatch):
        self._call_gate(monkeypatch, run_id=RID, current=ci.db_name(RID))

    def test_without_the_run_id_nothing_changes(self, monkeypatch):
        """★canary 가 아니면 **기존 동작 그대로**다 — 아무것도 안 본다."""
        self._call_gate(monkeypatch, run_id=None, current="theroad")

    def test_the_run_id_travels_to_the_child(self):
        e = {"DATABASE_URL": ci.db_url(RID, template=TEMPLATE),
             "PROJECTS_DIR": str(ci.root_dir(RID) / "projects"),
             "THEROAD_CANARY_RUN_ID": RID, "PATH": "/usr/bin"}
        assert ci.child_env(RID, env=e).get("THEROAD_CANARY_RUN_ID") == RID


class TestTheAlembicWrapperCanActuallyRun:
    """★★「가리킨다」와 「연다」는 다르다 (Codex 실측 2026-08-31).

    앞 판은 `backend/backend/alembic.ini` 를 가리켰는데, 시험이 명령줄만 보고
    **실제로 열어 보지 않아** 통과했다.
    """

    def test_the_config_file_is_really_there(self):
        from tools.grounding_audit import canary_alembic as ca

        assert ca.ALEMBIC_INI.is_file(), f"★없다: {ca.ALEMBIC_INI}"

    def test_alembic_can_read_it_and_find_a_head(self):
        """★진짜 `Config` 로 열어 head 를 읽는다 — DB 는 안 건드린다."""
        from alembic.config import Config
        from alembic.script import ScriptDirectory

        from tools.grounding_audit import canary_alembic as ca

        head = ScriptDirectory.from_config(
            Config(str(ca.ALEMBIC_INI))).get_current_head()
        assert head, "★head 를 못 읽었다"

    def test_the_backend_root_is_not_doubled(self):
        from tools.grounding_audit import canary_alembic as ca

        assert ca.BACKEND.name == "backend"
        assert "backend/backend" not in str(ca.ALEMBIC_INI)


class TestTheSecretSurvivesTheUrlRoundTrip:
    """★★★`str(URL)` 은 비밀번호를 `***` 로 **가린다** (실측 2026-08-31).

    그대로 넘기면 접속이 「인증 실패」로 죽는다. 유료 주행 첫 걸음에서 실제로
    그렇게 섰다 — 다만 **아무것도 안 사고** 섰다.
    """

    SECRET = "a-secret-16chars"

    def _t(self):
        return f"postgresql://사용자:{self.SECRET}@어딘가:5432/theroad"

    def test_the_canary_url_keeps_the_password(self):
        """★substring 이 아니라 **파싱해서** 본다 — URL 인코딩이 있다."""
        from sqlalchemy.engine import make_url

        got = make_url(ci.db_url(RID, template=self._t()))
        assert got.password == self.SECRET, "★비밀번호가 가려졌다"

    def test_the_admin_dsn_keeps_the_password(self, monkeypatch):
        monkeypatch.delenv("THEROAD_CANARY_ADMIN_DSN", raising=False)
        monkeypatch.setenv("THEROAD_CANARY_TEMPLATE_URL", self._t())
        from sqlalchemy.engine import make_url

        assert make_url(ci.maintenance_dsn()).password == self.SECRET

    def test_the_password_length_survives(self):
        """★길이로도 본다 — `***` 는 3글자다."""
        from sqlalchemy.engine import make_url

        got = make_url(ci.db_url(RID, template=self._t()))
        assert len(got.password) == len(self.SECRET)


class TestTheSecretNeverShowsUp:
    """★★넘길 때는 그대로, **찍을 때는 가린다**.

    Codex 확인 항목 (2026-08-31): 「실제 비밀번호가 산출물이나 예외 로그로
    새지 않는지」.
    """

    SECRET = "do-not-print-me16"

    def test_the_mask_removes_it(self):
        u = f"postgresql://사용자:{self.SECRET}@어딘가:5432/theroad"
        assert self.SECRET not in ci.masked(u)
        assert "어딘가" in ci.masked(u), "★가리다가 다 지웠다"

    def test_a_broken_url_is_still_masked(self):
        u = f"postgresql://사용자:{self.SECRET}@"
        assert self.SECRET not in ci.masked(u)

    def test_the_refusal_message_does_not_carry_it(self, monkeypatch):
        u = f"postgresql://사용자:{self.SECRET}@어딘가:5432/"
        with pytest.raises(ci.IsolationRefused) as e:
            ci.assert_isolated(RID, env={"DATABASE_URL": u,
                                         "PROJECTS_DIR": "/x"})
        assert self.SECRET not in str(e.value)

    def test_no_refusal_message_carries_it(self, monkeypatch, tmp_path):
        """★어느 갈래로 서든 새면 안 된다 — 문 전부를 훑는다."""
        u = f"postgresql://사용자:{self.SECRET}@어딘가:5432/theroad"
        bad_envs = [
            {"DATABASE_URL": u, "PROJECTS_DIR": "/x"},           # 원본 DB
            {"DATABASE_URL": f"postgresql://사용자:{self.SECRET}"
                             "@어딘가:5432/theroad_canary_zzzz",
             "PROJECTS_DIR": "/x"},                              # run_id 없음
            {"DATABASE_URL": ci.db_url(RID, template=u),
             "PROJECTS_DIR": "/딴곳"},                            # 디렉토리 밖
        ]
        for env in bad_envs:
            with pytest.raises(ci.IsolationRefused) as e:
                ci.assert_isolated(RID, env=env)
            assert self.SECRET not in str(e.value), f"★샜다: {env}"

    def test_the_public_return_has_no_full_url(self):
        """★★공개 반환에 **접속 주소 전체**를 안 넣는다 (Codex 2026-08-31).

        나중에 이 결과를 저장하면 비밀이 샌다.
        """
        u = f"postgresql://사용자:{self.SECRET}@어딘가:5432/theroad"
        got = ci.assert_isolated(
            RID, env={"DATABASE_URL": ci.db_url(RID, template=u),
                      "PROJECTS_DIR": str(ci.root_dir(RID) / "projects")})
        blob = repr(got)
        assert self.SECRET not in blob
        assert "database_url" not in got, "★전체 주소가 반환에 남아 있다"
        assert got["database_url_masked"].count("***") == 1

    def test_the_alembic_output_is_masked(self):
        """★바깥 프로그램 출력에 섞인 주소도 지운다."""
        line = ("INFO [alembic] url=postgresql://사용자:"
                f"{self.SECRET}@어딘가:5432/x 로 붙었다")
        assert self.SECRET not in ci.masked_text(line)
        assert "alembic" in ci.masked_text(line), "★지우다가 다 날렸다"

    def test_the_run_record_has_no_url(self, monkeypatch, tmp_path):
        """★산출물(`canary_run.json`)에 접속 주소가 안 실린다."""
        import json

        from tools.grounding_audit import canary_run as cr

        monkeypatch.setenv("THEROAD_CANARY_TEMPLATE_URL",
                           f"postgresql://사용자:{self.SECRET}@어딘가:5432/theroad")
        monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path))
        got = cr.run(live=False)
        saved = (ci.root_dir(got["run_id"]) / "canary_run.json").read_text(
            encoding="utf-8")
        assert self.SECRET not in saved
        assert "postgresql://" not in saved
        assert json.loads(saved)["stages"]["isolation"]["db"]


class TestTheSecretIsZeroInEveryArtifact:
    """★★★알려진 비밀을 넣고 **실제 산출물**에서 0회인지 본다.

    Codex (2026-08-31): 「소스 금지어 검사가 아니라 **실제 끝점 산출**을
    보십시오.」
    """

    SECRET = "canary-secret-99x"

    def test_zero_occurrences_across_all_written_files(self, monkeypatch,
                                                       tmp_path):
        from tools.grounding_audit import canary_run as cr

        monkeypatch.setenv(
            "THEROAD_CANARY_TEMPLATE_URL",
            f"postgresql://사용자:{self.SECRET}@어딘가:5432/theroad")
        monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path))
        got = cr.run(live=False)

        # ★이 판이 쓴 **모든 파일**을 훑는다
        root = ci.root_dir(got["run_id"])
        files = [p for p in root.rglob("*") if p.is_file()]
        assert files, "★아무 파일도 안 썼다 — 무엇을 재는지 모른다"
        for p in files:
            body = p.read_text(encoding="utf-8", errors="replace")
            assert self.SECRET not in body, f"★{p.name} 에 샜다"

    def test_a_failure_path_writes_nothing_with_it(self, monkeypatch,
                                                   tmp_path):
        """★서는 갈래에서도 예외 문구에 안 담긴다."""
        u = f"postgresql://사용자:{self.SECRET}@어딘가:5432/theroad"
        with pytest.raises(ci.IsolationRefused) as e:
            ci.assert_isolated(RID, env={"DATABASE_URL": u,
                                         "PROJECTS_DIR": "/x"})
        assert self.SECRET not in str(e.value)

    def test_the_child_env_check_does_not_need_the_public_url(self):
        """★자식 env 대조는 **입력끼리** 본다 — 공개 반환에 비밀이 없어도 된다."""
        u = f"postgresql://사용자:{self.SECRET}@어딘가:5432/theroad"
        e = {"DATABASE_URL": ci.db_url(RID, template=u),
             "PROJECTS_DIR": str(ci.root_dir(RID) / "projects"),
             "PATH": "/usr/bin"}
        got = ci.child_env(RID, env=e)
        assert got["DATABASE_URL"] == e["DATABASE_URL"]


class TestTheGateMustNotVoidTheMigration:
    """★★★제 안전문이 이관을 **통째로 롤백**시켰다 (실측 2026-08-31).

    `SELECT current_database()` 는 읽기지만 그것만으로 **트랜잭션이 열린다**.
    그대로 두면 alembic 의 `begin_transaction()` 이 그 위에 얹혀, 연결이 닫힐
    때 **전부 되돌아간다** — `stamp` 가 「했다」고 찍고도 표가 안 생겼다.
    문을 다는 것과 문이 **일을 망치지 않는 것**은 다르다.
    """

    def _gate(self):
        import ast
        from pathlib import Path

        src = Path("alembic/env.py").read_text(encoding="utf-8")
        tree = ast.parse(src)
        fn = next(n for n in ast.walk(tree)
                  if isinstance(n, ast.FunctionDef)
                  and n.name == "_assert_canary_db_if_asked")
        ns: dict = {}
        exec(compile(ast.Module([fn], []), "<env>", "exec"), ns)
        return ns["_assert_canary_db_if_asked"]

    def test_it_closes_what_it_opened(self, monkeypatch):
        """★읽고 나서 **연 것을 닫는다** — 순서까지 본다."""
        calls = []

        class _Res:
            def scalar(self):
                return ci.db_name(RID)

        class _Con:
            def execute(self, *a, **k):
                calls.append("execute")
                return _Res()

            def rollback(self):
                calls.append("rollback")

        monkeypatch.setenv("THEROAD_CANARY_RUN_ID", RID)
        self._gate()(_Con())
        assert calls == ["execute", "rollback"], f"★{calls}"

    def test_it_does_nothing_when_not_a_canary(self, monkeypatch):
        """★canary 가 아니면 **연결에 손도 안 댄다**."""
        calls = []

        class _Con:
            def execute(self, *a, **k):
                calls.append("execute")
                raise AssertionError("★건드리면 안 된다")

            def rollback(self):
                calls.append("rollback")

        monkeypatch.delenv("THEROAD_CANARY_RUN_ID", raising=False)
        self._gate()(_Con())
        assert calls == []

    def test_a_wrong_database_still_stops(self, monkeypatch):
        class _Res:
            def scalar(self):
                return "theroad"

        class _Con:
            def execute(self, *a, **k):
                return _Res()

            def rollback(self):
                pass

        monkeypatch.setenv("THEROAD_CANARY_RUN_ID", RID)
        with pytest.raises(ci.IsolationRefused):
            self._gate()(_Con())


class TestTheSeedWasRemovedOnPurpose:
    """★★★지문 표 복사를 **뺐다** (Codex 결정 2026-08-31).

    이 판에서 `db=` 를 넘기는 prompt 이름과 production 활성 override 의
    **교집합이 0** 이라, 복사해도 이 canary 의 outbound 를 **한 글자도**
    안 바꾼다. 그런데 유지하면 원본 DB read · `pg_dump`/`psql` · 재개 상태라는
    **새 실패면**만 유료 경로 앞에 붙는다.

    ★「production 에 더 충실해 보인다」는 이유만으로 **효과 없는 부품**을
    넣지 않는다.
    """

    def test_the_schema_step_does_not_seed(self):
        import inspect

        from tools.grounding_audit import canary_alembic as ca

        assert not hasattr(ca, "seed_prompt_templates")
        src = inspect.getsource(ca.upgrade)
        assert "prompt_template" not in src

    def test_the_module_still_runs_as_a_script(self):
        """★★helper 를 `main()` **뒤**에 두면 CLI 가 `NameError` 로 죽는다.

        import 기반 시험은 통과하는데 파일 직접 실행만 깨진다 (Codex 실측).
        그래서 **끝점**으로 본다 — 실제로 돌려 도움말이 나오는지.
        """
        import subprocess
        import sys

        got = subprocess.run(
            [sys.executable, "tools/grounding_audit/canary_alembic.py"],
            capture_output=True, text=True)
        assert got.returncode == 2, got.stderr[-400:]
        assert "NameError" not in got.stderr

    def test_no_db_consulting_prompt_is_overridden(self):
        """★교집합 0 을 잠근다 — 늘어나면 이 시험이 **먼저 깨진다**."""
        import ast
        import subprocess

        from sqlalchemy.engine import make_url

        from tools.grounding_audit import canary_cost_table as ct

        from tools.grounding_audit.canary_run import default_fixture_config
        FIXTURE_CONFIG = default_fixture_config()

        # ★사본을 만들지 않는다 — 네 번째 사본이 여기 남아 있었다
        plan = ct.execution_plan(config=FIXTURE_CONFIG)
        pairs = set()
        for st in [r["step"] for r in plan["applied"]]:
            f = ct.step_module(st)
            if not f:
                continue
            try:
                tree = ast.parse(open(f, encoding="utf-8",
                                      errors="replace").read())
            except Exception:                       # noqa: BLE001
                continue
            for n in ast.walk(tree):
                if (isinstance(n, ast.Call)
                        and getattr(n.func, "id", "") == "load_prompt"
                        and any(k.arg == "db" for k in n.keywords)
                        and len(n.args) >= 2
                        and all(isinstance(a, ast.Constant)
                                for a in n.args[:2])):
                    pairs.add((n.args[0].value, n.args[1].value))
        assert pairs, "★`db=` 를 넘기는 자리를 하나도 못 찾았다 — 탐지가 헐겁다"
        u = make_url(ci.template_url())
        out = subprocess.run(
            ["psql", "-h", str(u.host), "-p", str(u.port or 5432),
             "-U", str(u.username), "-d", "theroad", "-t", "-A",
             "-c", "SELECT module||'/'||name FROM prompt_template "
                   "WHERE is_active;"],
            env={"PGPASSWORD": str(u.password or ""),
                 "PATH": "/usr/bin:/bin:/usr/local/bin:/opt/homebrew/bin"},
            capture_output=True, text=True)
        if out.returncode != 0:
            pytest.skip("원본 DB 를 못 읽었다 — 「없다」로 읽지 않는다")
        mine = {f"{m}/{n}" for m, n in pairs}
        assert not (mine & set(out.stdout.split())), \
            f"★이제 겹친다 — seed 를 다시 봐야 한다: {sorted(mine & set(out.stdout.split()))}"