"""Safety checks for the disposable migration rehearsal. No production database."""

from __future__ import annotations

import os
import shutil
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "deploy" / "rehearse-production-migrations.sh"


def resolve_bash() -> str:
    found = shutil.which("bash")
    git_bash = Path(r"D:\Program Files\Git\bin\bash.exe")
    if os.name == "nt" and git_bash.exists():
        lowered = (found or "").lower()
        if not found or "system32" in lowered or "windowsapps" in lowered:
            return str(git_bash)
    if found:
        return found
    if git_bash.exists():
        return str(git_bash)
    return "bash"


BASH = resolve_bash()


def run(env: dict[str, str]) -> subprocess.CompletedProcess[str]:
    full = os.environ.copy()
    full.update(env)
    return subprocess.run(
        [BASH, str(SCRIPT)],
        cwd=ROOT,
        env=full,
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )


def test_refuses_a_non_production_path() -> None:
    completed = run({"DEPLOY_PATH": "/tmp/not-production"})
    assert completed.returncode == 1
    assert "unexpected production path" in completed.stderr


def test_refuses_the_production_database_name() -> None:
    completed = run(
        {
            "JYOTISHA_REHEARSAL_LOCAL": "1",
            "JYOTISHA_REHEARSAL_DATABASE_URL": "postgresql://postgres:secret-value@127.0.0.1:1/jyotisha",
        }
    )
    assert completed.returncode == 1
    assert "refusing to touch the production database" in completed.stderr
    assert "secret-value" not in completed.stdout
    assert "secret-value" not in completed.stderr


def test_plan_keeps_the_stop_file_out_of_the_first_batch() -> None:
    completed = run(
        {
            "JYOTISHA_REHEARSAL_LOCAL": "1",
            "JYOTISHA_REHEARSAL_PLAN": "1",
            "JYOTISHA_REHEARSAL_DATABASE_URL": "postgresql://postgres:secret-value@127.0.0.1:1/rehearse_plan",
        }
    )
    assert completed.returncode == 0, completed.stderr
    text = completed.stdout
    assert "rehearsal_plan=1" in text
    assert "database_name=rehearse_plan" in text
    assert "stop_before=20260925030000_chart_subject_trust_and_limit.sql" in text
    assert "stop_in_before=no" in text
    values = dict(line.split("=", 1) for line in text.splitlines() if "=" in line)
    assert int(values["batch_before"]) < int(values["batch_all"])
    assert "secret-value" not in completed.stdout
    assert "secret-value" not in completed.stderr


def test_local_url_with_query_is_refused() -> None:
    completed = run(
        {
            "JYOTISHA_REHEARSAL_LOCAL": "1",
            "JYOTISHA_REHEARSAL_DATABASE_URL": "postgresql://postgres:secret-value@127.0.0.1:1/rehearse_plan?database=jyotisha",
        }
    )
    assert completed.returncode == 1
    assert "query parameters" in completed.stderr
    assert "secret-value" not in completed.stdout
    assert "secret-value" not in completed.stderr


def bash_path(path: Path) -> str:
    if os.name != "nt":
        return str(path)
    completed = subprocess.run(
        [BASH, "-lc", 'cygpath -u "$1"', "cygpath", str(path)],
        check=True,
        capture_output=True,
        text=True,
    )
    return completed.stdout.strip()


def test_host_mode_does_not_call_host_node_and_failed_cleanup_is_nonzero(tmp_path: Path) -> None:
    bin_dir = tmp_path / "bin"
    host = tmp_path / "host"
    (host / ".state").mkdir(parents=True)
    (host / "backups").mkdir()
    dump = host / "backups" / "production-pre-migration-20261014T000000Z.dump.enc"
    dump.write_bytes(b"not-a-real-dump")
    (host / ".env.production.database").write_text(
        "\n".join(
            [
                "POSTGRES_DB=jyotisha",
                "POSTGRES_USER=postgres",
                "POSTGRES_PASSWORD=secret-value",
                "PRODUCTION_BACKUP_ENCRYPTION_KEY=not-a-real-key",
                "",
            ]
        ),
        encoding="utf-8",
        newline="\n",
    )
    docker_log = tmp_path / "docker.log"
    node_log = tmp_path / "node.log"
    bin_dir.mkdir()
    (bin_dir / "docker").write_text(
        "\n".join(
            [
                "#!/bin/bash",
                'printf "%s\\n" "$*" >> "${DOCKER_LOG:?}"',
                'case "$*" in',
                "  *exec\\ -i*|*-i\\ *) cat >/dev/null || true ;;",
                "esac",
                'joined="$*"',
                'case "$joined" in',
                '  *ps\\ *) echo container1; exit 0 ;;',
                "esac",
                'case "$joined" in',
                "  *'{{.Image}}'*) echo sha256:0000000000000000000000000000000000000000000000000000000000000000; exit 0 ;;",
                "esac",
                'case "$joined" in',
                "  *inspect*) echo jyotisha-net; exit 0 ;;",
                "esac",
                'case "$joined" in',
                "  *DROP\\ DATABASE*) exit 1 ;;",
                "esac",
                'case "$joined" in',
                "  *has_table_privilege*)",
                "    printf '%s\\n' 'authenticated public.profiles true' 'authenticated public.chat_sessions true' 'service_role public.chart_profiles true'",
                "    exit 0",
                "    ;;",
                "esac",
                'case "$joined" in',
                "  *count*) echo 0; exit 0 ;;",
                "esac",
                "exit 0",
                "",
            ]
        ),
        encoding="utf-8",
        newline="\n",
    )
    (bin_dir / "node").write_text(
        "\n".join(
            [
                "#!/bin/bash",
                'printf "%s\\n" "$*" >> "${NODE_LOG:?}"',
                "exit 9",
                "",
            ]
        ),
        encoding="utf-8",
        newline="\n",
    )
    (bin_dir / "openssl").write_text("#!/bin/bash\nexit 0\n", encoding="utf-8", newline="\n")
    unix_bin = bash_path(bin_dir)
    python_bin = bash_path(Path(sys.executable))
    (bin_dir / "python3").write_text(
        f"#!/bin/bash\nexec '{python_bin}' \"$@\"\n",
        encoding="utf-8",
        newline="\n",
    )
    subprocess.run(
        [BASH, "-lc", f'chmod +x "{unix_bin}/docker" "{unix_bin}/node" "{unix_bin}/openssl" "{unix_bin}/python3"'],
        check=True,
    )
    full = os.environ.copy()
    full.update(
        {
            "JYOTISHA_REHEARSAL_FAKE_HOST": "1",
            "DEPLOY_PATH": bash_path(host),
            "RECOVERY_DUMP": bash_path(dump),
            "DOCKER_LOG": bash_path(docker_log),
            "NODE_LOG": bash_path(node_log),
        }
    )
    completed = subprocess.run(
        [
            BASH,
            "-lc",
            'export PATH="$1:$PATH"; exec bash "$2"',
            "bash",
            unix_bin,
            bash_path(SCRIPT),
        ],
        cwd=ROOT,
        env=full,
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    assert completed.returncode != 0, completed.stdout + completed.stderr
    assert "演练库未删除，请手动 DROP DATABASE rehearse_" in completed.stderr
    assert "secret-value" not in completed.stdout
    assert "secret-value" not in completed.stderr
    assert not node_log.exists()
    docker_text = docker_log.read_text(encoding="utf-8")
    assert "--entrypoint node" in docker_text
    assert "DROP DATABASE" in docker_text


def test_restore_keeps_backup_owners_and_compares_grants() -> None:
    text = SCRIPT.read_text(encoding="utf-8")
    assert "--no-owner" not in text
    assert "ALTER TYPE" not in text
    assert "typtype" not in text
    assert "revoked_grants" in text
    assert "chart_profiles_service_role_select_missing" in text
    assert "role%3Dschema_owner" in text
    assert "grant connect, create on database" in text