from __future__ import annotations

import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace


REPO_ROOT = Path(__file__).resolve().parents[1]
RUNTIME_DIR = REPO_ROOT / "deployments" / "offline" / "runtime"
for path in (str(REPO_ROOT), str(RUNTIME_DIR)):
    if path not in sys.path:
        sys.path.insert(0, path)

import deployment_patch_replay as replay  # noqa: E402
from common import DeployError  # noqa: E402


HEADER = "Id\tIsCustomized\tOwnerPack\tAssertionState"


def write_bundle(root: Path, manifest: dict, files: dict[str, str]) -> Path:
    bundle = root / "bundle"
    bundle.mkdir(parents=True, exist_ok=True)
    (bundle / "patch-bundle.json").write_text(json.dumps(manifest))
    for name, body in files.items():
        (bundle / name).write_text(body)
    return bundle


BASE_MANIFEST = {
    "title": "test bundle",
    "created_at": "2026-01-01T00:00:00Z",
    "sql_files": ["001_noop.sql"],
}
NOOP_SQL = "UPDATE `T` SET `a` = 1 WHERE `id` = 'x';\n"
SELECT_SQL = "-- a comment mentioning SELECT in prose\nSELECT `Id` FROM `T` WHERE `Id` = 'x';\n"


class ParseAssertionsTest(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)

    def tearDown(self) -> None:
        self.tmp.cleanup()

    def bundle(self, assertions: object, extra_files: dict[str, str] | None = None) -> Path:
        manifest = dict(BASE_MANIFEST)
        if assertions is not None:
            manifest["assertions"] = assertions
        files = {"001_noop.sql": NOOP_SQL, "099_assert.sql": SELECT_SQL}
        files.update(extra_files or {})
        return write_bundle(self.root, manifest, files)

    def parse(self, bundle: Path):
        return replay.parse_patch_assertions(replay.load_patch_bundle_definition(bundle), bundle)

    def test_absent_assertions_is_backwards_compatible(self) -> None:
        self.assertEqual(self.parse(self.bundle(None)), [])

    def test_explicit_null_assertions_is_refused(self) -> None:
        # An explicit null is malformed, not absent. The D3 guard keys on presence, so treating
        # null as absent here would let staging/Test skip evaluation on a manifest D3 refuses.
        manifest = dict(BASE_MANIFEST)
        manifest["assertions"] = None
        bundle = write_bundle(self.root, manifest, {"001_noop.sql": NOOP_SQL, "099_assert.sql": SELECT_SQL})
        with self.assertRaises(DeployError) as ctx:
            self.parse(bundle)
        self.assertIn("non-empty array", str(ctx.exception))

    def test_valid_assertion_parses(self) -> None:
        parsed = self.parse(
            self.bundle([
                {
                    "id": "state",
                    "sql_file": "099_assert.sql",
                    "expect_rows": 1,
                    "expect_columns": {"AssertionState": "ok"},
                }
            ])
        )
        self.assertEqual(len(parsed), 1)
        self.assertEqual(parsed[0]["id"], "state")
        self.assertEqual(parsed[0]["expect_rows"], 1)
        self.assertEqual(parsed[0]["expect_columns"], {"AssertionState": "ok"})

    def test_traversal_outside_bundle_is_refused(self) -> None:
        (self.root / "outside.sql").write_text(SELECT_SQL)
        with self.assertRaises(DeployError) as ctx:
            self.parse(self.bundle([{"id": "x", "sql_file": "../outside.sql", "expect_rows": 1}]))
        self.assertIn("must stay inside the bundle", str(ctx.exception))

    def test_absolute_path_is_refused(self) -> None:
        outside = self.root / "abs.sql"
        outside.write_text(SELECT_SQL)
        with self.assertRaises(DeployError) as ctx:
            self.parse(self.bundle([{"id": "x", "sql_file": str(outside), "expect_rows": 1}]))
        self.assertIn("must stay inside the bundle", str(ctx.exception))

    def test_mutating_assertion_sql_is_refused(self) -> None:
        with self.assertRaises(DeployError) as ctx:
            self.parse(
                self.bundle(
                    [{"id": "x", "sql_file": "bad.sql", "expect_rows": 1}],
                    {"bad.sql": "UPDATE `T` SET `a` = 2 WHERE `id` = 'x';\n"},
                )
            )
        self.assertIn("must be read-only", str(ctx.exception))

    def test_sql_with_only_comments_is_refused(self) -> None:
        with self.assertRaises(DeployError) as ctx:
            self.parse(
                self.bundle(
                    [{"id": "x", "sql_file": "empty.sql", "expect_rows": 1}],
                    {"empty.sql": "-- SELECT is only mentioned in this comment\n"},
                )
            )
        self.assertIn("no executable statement", str(ctx.exception))

    def test_missing_sql_file_is_refused(self) -> None:
        with self.assertRaises(DeployError) as ctx:
            self.parse(self.bundle([{"id": "x", "sql_file": "nope.sql", "expect_rows": 1}]))
        self.assertIn("missing sql file", str(ctx.exception))

    def test_assertion_without_expectations_is_refused(self) -> None:
        with self.assertRaises(DeployError) as ctx:
            self.parse(self.bundle([{"id": "x", "sql_file": "099_assert.sql"}]))
        self.assertIn("must declare expect_rows or expect_columns", str(ctx.exception))


class ParseTabularRowsTest(unittest.TestCase):
    def test_header_and_rows(self) -> None:
        rows = replay.parse_tabular_rows(HEADER + "\nid1\t0\tpack\tok\n")
        self.assertEqual(rows, [{"Id": "id1", "IsCustomized": "0", "OwnerPack": "pack", "AssertionState": "ok"}])

    def test_header_only_is_zero_rows(self) -> None:
        self.assertEqual(replay.parse_tabular_rows(HEADER + "\n"), [])

    def test_empty_output_is_zero_rows(self) -> None:
        self.assertEqual(replay.parse_tabular_rows(""), [])

    def test_empty_string_row_is_not_dropped(self) -> None:
        # With -B -r a row whose single column is '' is emitted as a blank line. Treating it as
        # no row would let an `expect_rows: 0` assertion pass on an unmet state.
        self.assertEqual(replay.parse_tabular_rows("Value\n\n"), [{"Value": ""}])

    def test_empty_string_row_after_a_normal_row(self) -> None:
        self.assertEqual(
            replay.parse_tabular_rows("Value\nx\n\n"), [{"Value": "x"}, {"Value": ""}]
        )

    def test_output_without_trailing_newline(self) -> None:
        self.assertEqual(replay.parse_tabular_rows("Value\nx"), [{"Value": "x"}])

    def test_expect_zero_rows_fails_when_an_empty_row_exists(self) -> None:
        artifact_dir = Path(tempfile.mkdtemp())
        sql_file = artifact_dir / "099_assert.sql"
        sql_file.write_text(SELECT_SQL)
        original = replay.query_patch_assertion_rows
        replay.query_patch_assertion_rows = lambda _e, _s: replay.parse_tabular_rows("Value\n\n")
        try:
            with self.assertRaises(DeployError) as ctx:
                replay.evaluate_patch_assertion(
                    {},
                    {"id": "z", "sql_file": sql_file, "expect_rows": 0, "expect_columns": {}},
                    artifact_dir,
                )
            self.assertIn("expected 0 row", str(ctx.exception))
        finally:
            replay.query_patch_assertion_rows = original

    def test_ragged_row_is_refused(self) -> None:
        raw = "customer@example.com\tSecretState\nid1\tprivate-value\textra\n"
        with self.assertRaises(DeployError) as ctx:
            replay.parse_tabular_rows(raw)
        self.assertEqual(
            str(ctx.exception),
            "unexpected patch assertion result shape: "
            "row_index=1 expected_columns=2 actual_columns=3",
        )
        self.assertNotIn("customer@example.com", str(ctx.exception))
        self.assertNotIn("private-value", str(ctx.exception))


class EvaluateAssertionTest(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.artifact_dir = Path(self.tmp.name)
        self.sql_file = self.artifact_dir / "099_assert.sql"
        self.sql_file.write_text(SELECT_SQL)
        self.original_query = replay.query_patch_assertion_rows

    def tearDown(self) -> None:
        replay.query_patch_assertion_rows = self.original_query
        self.tmp.cleanup()

    def evaluate(self, output: str, *, expect_rows=1, expect_columns=None):
        replay.query_patch_assertion_rows = lambda _envv, _sql: replay.parse_tabular_rows(output)
        assertion = {
            "id": "state",
            "sql_file": self.sql_file,
            "expect_rows": expect_rows,
            "expect_columns": expect_columns or {},
        }
        return replay.evaluate_patch_assertion({}, assertion, self.artifact_dir)

    def test_matching_state_passes(self) -> None:
        result = self.evaluate(HEADER + "\nid1\t0\tpack\tok\n", expect_columns={"AssertionState": "ok"})
        self.assertEqual(result["status"], "passed")
        self.assertEqual(result["rows"], "1")

    def test_column_mismatch_fails_closed(self) -> None:
        with self.assertRaises(DeployError) as ctx:
            self.evaluate(HEADER + "\nid1\t1\tpack\tunexpected\n", expect_columns={"AssertionState": "ok"})
        self.assertIn("AssertionState", str(ctx.exception))

    def test_missing_row_fails_closed(self) -> None:
        with self.assertRaises(DeployError) as ctx:
            self.evaluate(HEADER + "\n", expect_columns={"AssertionState": "ok"})
        self.assertIn("expected 1 row", str(ctx.exception))

    def test_unknown_column_fails_closed(self) -> None:
        with self.assertRaises(DeployError) as ctx:
            self.evaluate(HEADER + "\nid1\t0\tpack\tok\n", expect_columns={"Nope": "ok"})
        self.assertIn("not in result set", str(ctx.exception))

    def test_bounded_result_hash_is_written_without_raw_rows(self) -> None:
        self.evaluate(HEADER + "\nid1\t0\tpack\tok\n", expect_columns={"AssertionState": "ok"})
        artifact = self.artifact_dir / "099_assert.assertion.json"
        self.assertTrue(artifact.is_file())
        evidence = json.loads(artifact.read_text())
        self.assertEqual(
            evidence["schema"], "nexuscrm.patch-bundle-assertion-evidence/v1"
        )
        self.assertEqual(evidence["row_count"], 1)
        self.assertRegex(evidence["rows_sha256"], r"^[0-9a-f]{64}$")
        self.assertNotIn("id1", artifact.read_text())
        self.assertNotIn("ok", artifact.read_text())


class ManifestHardeningTest(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)

    def tearDown(self) -> None:
        self.tmp.cleanup()

    def bundle(self, assertions: object, extra_files: dict[str, str] | None = None) -> Path:
        manifest = dict(BASE_MANIFEST)
        manifest["assertions"] = assertions
        files = {"001_noop.sql": NOOP_SQL, "099_assert.sql": SELECT_SQL}
        files.update(extra_files or {})
        return write_bundle(self.root, manifest, files)

    def parse(self, bundle: Path):
        return replay.parse_patch_assertions(replay.load_patch_bundle_definition(bundle), bundle)

    def test_boolean_expect_rows_is_refused(self) -> None:
        # bool is an int subclass, so `true` would otherwise be read as an expected count of 1.
        with self.assertRaises(DeployError) as ctx:
            self.parse(self.bundle([{"id": "x", "sql_file": "099_assert.sql", "expect_rows": True}]))
        self.assertIn("expect_rows", str(ctx.exception))

    def test_negative_expect_rows_is_refused(self) -> None:
        with self.assertRaises(DeployError):
            self.parse(self.bundle([{"id": "x", "sql_file": "099_assert.sql", "expect_rows": -1}]))

    def test_executable_version_comment_is_refused(self) -> None:
        # split_sql_statements strips block comments, but the client executes /*! ... */.
        smuggled = "SELECT `Id` FROM `T`;\n/*!40000 UPDATE `T` SET `a` = 2 */;\n"
        with self.assertRaises(DeployError) as ctx:
            self.parse(
                self.bundle([{"id": "x", "sql_file": "smuggle.sql", "expect_rows": 1}],
                            {"smuggle.sql": smuggled})
            )
        self.assertIn("executable comments", str(ctx.exception))

    def test_falsy_invalid_expect_columns_is_refused(self) -> None:
        # `or {}` would have turned [] into an empty object, silently dropping the column check.
        for bad in ([], "", 0, None):
            with self.subTest(value=bad):
                with self.assertRaises(DeployError) as ctx:
                    self.parse(
                        self.bundle([{
                            "id": "x",
                            "sql_file": "099_assert.sql",
                            "expect_rows": 1,
                            "expect_columns": bad,
                        }])
                    )
                self.assertIn("expect_columns must be an object", str(ctx.exception))

    def test_absent_expect_columns_still_defaults(self) -> None:
        parsed = self.parse(self.bundle([{"id": "x", "sql_file": "099_assert.sql", "expect_rows": 1}]))
        self.assertEqual(parsed[0]["expect_columns"], {})

    def test_backslash_escape_smuggling_is_refused(self) -> None:
        # The shared splitter does not model MySQL backslash escapes, so this parses as one
        # statement while the client would execute the trailing UPDATE.
        exploit = "SELECT 'x" + chr(92) + "''; UPDATE `T` SET `a` = 2;\n"
        with self.assertRaises(DeployError) as ctx:
            self.parse(
                self.bundle([{"id": "x", "sql_file": "smuggle2.sql", "expect_rows": 1}],
                            {"smuggle2.sql": exploit})
            )
        self.assertIn("must not contain backslashes", str(ctx.exception))

    def test_any_backslash_is_refused(self) -> None:
        with self.assertRaises(DeployError):
            self.parse(
                self.bundle([{"id": "x", "sql_file": "bs.sql", "expect_rows": 1}],
                            {"bs.sql": "SELECT 'a" + chr(92) + "n' FROM `T`;\n"})
            )

    def test_interior_semicolon_is_refused(self) -> None:
        with self.assertRaises(DeployError) as ctx:
            self.parse(
                self.bundle([{"id": "x", "sql_file": "semi.sql", "expect_rows": 1}],
                            {"semi.sql": "-- note; here\nSELECT `Id` FROM `T`;\n"})
            )
        self.assertIn("at most one trailing semicolon", str(ctx.exception))

    def test_statement_without_trailing_semicolon_is_accepted(self) -> None:
        parsed = self.parse(
            self.bundle([{"id": "x", "sql_file": "nosemi.sql", "expect_rows": 1}],
                        {"nosemi.sql": "SELECT `Id` FROM `T`\n"})
        )
        self.assertEqual(len(parsed), 1)

    def test_multiple_statements_are_refused(self) -> None:
        # The parser reads one result set; a second header line would be read as a data row.
        # Two SELECTs need an interior semicolon, so the lexical guard catches them first.
        with self.assertRaises(DeployError) as ctx:
            self.parse(
                self.bundle([{"id": "x", "sql_file": "two.sql", "expect_rows": 1}],
                            {"two.sql": "SELECT `Id` FROM `T`;\nSELECT `Name` FROM `T`;\n"})
            )
        self.assertIn("single statement", str(ctx.exception))

    def test_optimizer_hint_comment_is_refused(self) -> None:
        with self.assertRaises(DeployError):
            self.parse(
                self.bundle([{"id": "x", "sql_file": "hint.sql", "expect_rows": 1}],
                            {"hint.sql": "SELECT /*+ SOME_HINT() */ `Id` FROM `T`;\n"})
            )


class D3ReplayGuardTest(unittest.TestCase):
    """The D3 replay path executes sql_files only, so it must refuse assertion-bearing bundles."""

    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)

    def tearDown(self) -> None:
        self.tmp.cleanup()

    def bundle(self, manifest: dict) -> Path:
        return write_bundle(self.root, manifest, {"001_noop.sql": NOOP_SQL, "099_assert.sql": SELECT_SQL})

    def test_assertion_bearing_bundle_is_refused(self) -> None:
        from scripts import serviceification_d3_test_patch as d3

        manifest = dict(BASE_MANIFEST)
        manifest["assertions"] = [{"id": "x", "sql_file": "099_assert.sql", "expect_rows": 1}]
        bundle = self.bundle(manifest)
        definition = replay.load_patch_bundle_definition(bundle)
        with self.assertRaises(Exception) as ctx:
            d3.require_patch_bundle_declares_no_assertions(definition, bundle)
        self.assertIn("does not evaluate manifest assertions", str(ctx.exception))

    def test_falsy_assertions_key_is_still_refused(self) -> None:
        # parse_patch_assertions rejects a present-but-malformed block, so the D3 guard must key
        # on presence rather than truthiness or the two paths disagree.
        from scripts import serviceification_d3_test_patch as d3

        for bad in ([], 0, "", None):
            with self.subTest(value=bad):
                manifest = dict(BASE_MANIFEST)
                manifest["assertions"] = bad
                bundle = self.bundle(manifest)
                definition = replay.load_patch_bundle_definition(bundle)
                with self.assertRaises(Exception) as ctx:
                    d3.require_patch_bundle_declares_no_assertions(definition, bundle)
                self.assertIn("does not evaluate manifest assertions", str(ctx.exception))

    def test_plain_bundle_is_accepted(self) -> None:
        from scripts import serviceification_d3_test_patch as d3

        bundle = self.bundle(dict(BASE_MANIFEST))
        definition = replay.load_patch_bundle_definition(bundle)
        d3.require_patch_bundle_declares_no_assertions(definition, bundle)

    def test_seal_digest_is_outside_bundle_and_recheck_is_stable(self) -> None:
        from scripts import serviceification_d3_test_patch as d3

        source = self.bundle(dict(BASE_MANIFEST))
        sealed = self.root / "sealed" / "patch-bundle"
        d3.seal_patch_bundle(source, sealed)

        digest_path = d3.patch_seal_digest_path(sealed)
        self.assertEqual(digest_path, sealed.parent / d3.SEAL_TREE_DIGEST_NAME)
        self.assertFalse((sealed / d3.SEAL_TREE_DIGEST_NAME).exists())
        self.assertEqual(d3.require_sealed_patch_bundle(sealed), d3.patch_tree_digest(source))
        self.assertEqual(d3.seal_patch_bundle(source, sealed), sealed)


class IndependentPatchOperatorTest(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)
        self.bundle = write_bundle(
            self.root,
            dict(BASE_MANIFEST),
            {"001_noop.sql": NOOP_SQL},
        )
        self.envv = {
            "ENV_NAME": "test-slot",
            "TIDB_HOST": "db.example.invalid",
            "TIDB_PORT": "4000",
            "TIDB_USER": "operator",
            "TIDB_PASSWORD": "secret-value",
            "TIDB_DATABASE": "target_db",
        }

    def tearDown(self) -> None:
        self.tmp.cleanup()


    def test_git_digest_is_bound_to_selected_commit_blobs(self) -> None:
        from scripts import run_patch_bundle as operator

        repo = self.root / "git-source"
        selected = repo / "docs" / "migration" / "patch-bundles" / "selected"
        selected.mkdir(parents=True)
        (selected / "patch-bundle.json").write_text(json.dumps(BASE_MANIFEST))
        (selected / "001_noop.sql").write_text(NOOP_SQL)
        for command in (
            ("init",),
            ("config", "user.email", "patch-test@example.invalid"),
            ("config", "user.name", "Patch Test"),
            ("add", "."),
            ("commit", "-m", "add selected patch"),
        ):
            subprocess.run(
                ["git", "-C", str(repo), *command],
                check=True,
                capture_output=True,
            )
        commit = operator.git_capture(repo, "rev-parse", "HEAD^{commit}")
        relative = Path("docs/migration/patch-bundles/selected")
        reviewed_digest = operator.git_patch_bundle_tree_digest(
            repo, commit, relative
        )
        self.assertEqual(reviewed_digest, replay.patch_bundle_tree_digest(selected))

        (selected / "001_noop.sql").write_text(NOOP_SQL + "-- changed later\n")
        self.assertEqual(
            operator.git_patch_bundle_tree_digest(repo, commit, relative),
            reviewed_digest,
        )
        self.assertNotEqual(replay.patch_bundle_tree_digest(selected), reviewed_digest)

    def test_public_adapter_rejects_release_or_unreviewed_paths(self) -> None:
        from scripts import run_patch_bundle as operator

        for invalid in (
            "docs/migration/patch-bundles",
            "deployments/offline/runtime/patches/example",
            "../docs/migration/patch-bundles/example",
            "/tmp/example",
        ):
            with self.subTest(path=invalid), self.assertRaises(DeployError):
                operator.require_patch_request(invalid)


    def test_independent_operator_is_not_a_prod_release_entrypoint(self) -> None:
        prod_workflow = (
            REPO_ROOT / ".github" / "workflows" / "deploy-multi-prod-d3-environment.yml"
        ).read_text()
        tag_workflow = (
            REPO_ROOT / ".github" / "workflows" / "emit-staging-d3-release-tag.yml"
        ).read_text()
        for workflow in (prod_workflow, tag_workflow):
            self.assertNotIn("run_patch_bundle.py", workflow)
            self.assertNotIn("patch_dir:", workflow)
        self.assertIn('(has("patchBundle") | not)', prod_workflow)

        catalog = json.loads(
            (
                REPO_ROOT
                / "deployments"
                / "offline"
                / "runtime"
                / "seeds"
                / "deploy-seed-catalog.json"
            ).read_text()
        )
        exclusion = next(
            item
            for item in catalog["excluded_mutations"]
            if item["exclusion_id"] == "patch_bundle_replay"
        )
        self.assertEqual(exclusion["owner"], "patch_bundle_migration")
        self.assertIn("scripts/run_patch_bundle.py", exclusion["source_paths"])
        self.assertIn(
            "deployments/offline/runtime/deployment_patch_replay.py",
            exclusion["source_paths"],
        )


class PreflightValidationTest(unittest.TestCase):
    """A malformed assertion block must be rejected before the deploy resets anything.

    The staging deploy stages reset-branch before apply-patch, so validation that only runs
    at execution time would let a bad bundle reset the TiDB branch and fail afterwards.
    """

    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)

    def tearDown(self) -> None:
        self.tmp.cleanup()

    def bundle(self, assertions: object) -> Path:
        manifest = dict(BASE_MANIFEST)
        manifest["assertions"] = assertions
        return write_bundle(
            self.root, manifest, {"001_noop.sql": NOOP_SQL, "099_assert.sql": SELECT_SQL}
        )

    def test_valid_bundle_passes_preflight(self) -> None:
        parsed = replay.validate_patch_bundle_assertions(
            self.bundle([{"id": "x", "sql_file": "099_assert.sql", "expect_rows": 1}])
        )
        self.assertEqual(len(parsed), 1)

    def test_malformed_bundle_is_rejected_at_preflight(self) -> None:
        with self.assertRaises(DeployError):
            replay.validate_patch_bundle_assertions(
                self.bundle([{"id": "x", "sql_file": "../escape.sql", "expect_rows": 1}])
            )

    def test_staging_prepare_validates_before_reset(self) -> None:
        import prepare_staging_runtime_env as prepare

        self.assert_prepare_rejects(prepare)

    def test_retired_candidate_compile_and_publishers_are_gone(self) -> None:
        import candidate_deployment
        import deploy_staging_env as staging
        import scripts.deployment_bundle as deployment_bundle

        self.assertNotIn("build-candidate", staging.STAGING_STAGES)
        for module, name in (
            (staging, "build_staging_candidate"),
            (candidate_deployment, "seal_candidate"),
            (candidate_deployment, "seal_staging_candidate"),
            (candidate_deployment, "CandidateCompileRetired"),
            (candidate_deployment, "CANDIDATE_COMPILE_AND_UPLOAD_RETIRED"),
            (deployment_bundle, "build_deployment_bundle"),
            (deployment_bundle, "build_from_runtime_env"),
        ):
            self.assertFalse(hasattr(module, name), f"{module.__name__}.{name}")

        for retired_path in (
            "deployments/offline/runtime/emit_staging_release_tag.py",
            "deployments/offline/runtime/render_staging_release_manifest.py",
            "deployments/offline/runtime/staging_candidate_checkpoint.py",
            "deployments/offline/runtime/verify_staging_release_tag_target.py",
            "scripts/release_artifact_store.py",
            "scripts/release_authority.py",
        ):
            self.assertFalse((REPO_ROOT / retired_path).exists(), retired_path)

    def assert_prepare_rejects(self, prepare) -> None:
        bundle = self.bundle([{"id": "x", "sql_file": "../escape.sql", "expect_rows": 1}])
        args = SimpleNamespace(patch_dir=str(bundle))
        with self.assertRaises(DeployError) as ctx:
            prepare.patch_bundle_definition_for_args(args)
        self.assertIn("must stay inside the bundle", str(ctx.exception))


class QueryClientArgsTest(unittest.TestCase):
    def test_batch_and_raw_flags_are_passed_without_suppressing_headers(self) -> None:
        import inspect

        source = inspect.getsource(replay.query_patch_assertion_rows)
        self.assertIn('"-B"', source)
        self.assertIn('"-r"', source)
        self.assertNotIn('"-N"', source)


if __name__ == "__main__":
    unittest.main()