from __future__ import annotations

import os
from pathlib import Path
import subprocess
import time


ROOT = Path(__file__).resolve().parents[3]
DEPLOYER = ROOT / "scripts/ops/deploy-signoz-native-backup-toolchain.sh"
BACKUP = ROOT / "scripts/backup/backup-signoz.sh"
CANARY = ROOT / "scripts/backup/run-signoz-backup-canary.sh"


def deployer_text() -> str:
    return DEPLOYER.read_text(encoding="utf-8")


def executable_text() -> str:
    return "\n".join(
        line
        for line in deployer_text().splitlines()
        if not line.lstrip().startswith("#")
    )


def base_env() -> dict[str, str]:
    env = os.environ.copy()
    env.update(
        {
            "TRACE_ID": "P0-OBS-002",
            "RUN_ID": "toolchain-test-run",
            "WORK_ITEM_ID": "P0-OBS-002",
        }
    )
    return env


def test_deployer_shell_is_valid_and_help_is_no_write() -> None:
    syntax = subprocess.run(
        ["bash", "-n", str(DEPLOYER)],
        check=False,
        capture_output=True,
        text=True,
    )
    assert syntax.returncode == 0, syntax.stderr

    help_result = subprocess.run(
        ["bash", str(DEPLOYER), "--help"],
        check=False,
        capture_output=True,
        text=True,
        env={
            key: value
            for key, value in os.environ.items()
            if key not in {"TRACE_ID", "RUN_ID", "WORK_ITEM_ID"}
        },
    )
    assert help_result.returncode == 0
    assert "--check" in help_result.stdout
    assert "--apply" in help_result.stdout
    assert "CLICKHOUSE_RESTORE_INVENTORY_HELPER" in help_result.stdout
    assert "CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG" in help_result.stdout
    assert "CLICKHOUSE_RESTORE_CLUSTER_CONFIG" in help_result.stdout


def test_deployer_requires_controlled_apply_ids_before_network_access() -> None:
    env = {
        key: value
        for key, value in os.environ.items()
        if key not in {"TRACE_ID", "RUN_ID", "WORK_ITEM_ID"}
    }
    result = subprocess.run(
        ["bash", str(DEPLOYER), "--check"],
        check=False,
        capture_output=True,
        text=True,
        env=env,
    )
    assert result.returncode == 64
    assert "must be path-safe" in result.stderr
    assert "ssh:" not in result.stderr


def test_safe_absolute_path_overrides_fail_closed_before_network_access() -> None:
    env = base_env()
    env["SIGNOZ_TOOLCHAIN_SCRIPT_DIR"] = "/backup/scripts/../escape"
    result = subprocess.run(
        ["bash", str(DEPLOYER), "--check"],
        check=False,
        capture_output=True,
        text=True,
        env=env,
    )
    assert result.returncode == 64
    assert "safe absolute paths" in result.stderr
    assert "ssh:" not in result.stderr

    env = base_env()
    env["SIGNOZ_TOOLCHAIN_RECEIPT_ROOT"] = "/var/tmp/receipts"
    result = subprocess.run(
        ["bash", str(DEPLOYER), "--check"],
        check=False,
        capture_output=True,
        text=True,
        env=env,
    )
    assert result.returncode == 64
    assert "receipt root must stay under /backup" in result.stderr

    env = base_env()
    env["SIGNOZ_TOOLCHAIN_STAGE_ROOT"] = "/backup/deploy-staging"
    env["SIGNOZ_TOOLCHAIN_SCRIPT_DIR"] = "/backup/deploy-staging/scripts"
    result = subprocess.run(
        ["bash", str(DEPLOYER), "--check"],
        check=False,
        capture_output=True,
        text=True,
        env=env,
    )
    assert result.returncode == 64
    assert "must not overlap" in result.stderr


def test_timeout_overrides_fail_closed_before_network_access() -> None:
    for key, value, expected in (
        (
            "SIGNOZ_TOOLCHAIN_SSH_TIMEOUT_SECONDS",
            "0",
            "positive integer seconds",
        ),
        (
            "SIGNOZ_TOOLCHAIN_DOCKER_TIMEOUT_SECONDS",
            "301",
            "exceeds bounded maximum",
        ),
    ):
        env = base_env()
        env[key] = value
        result = subprocess.run(
            ["bash", str(DEPLOYER), "--check"],
            check=False,
            capture_output=True,
            text=True,
            env=env,
        )
        assert result.returncode == 64
        assert expected in result.stderr
        assert "ssh:" not in result.stderr


def test_target_host_is_fixed_to_host110_and_not_environment_overridable() -> None:
    text = deployer_text()
    assert 'TARGET_HOST="wooo@192.168.0.110"' in text
    assert "TARGET_HOST:-" not in text
    assert "TARGET_HOST=" not in text.replace('TARGET_HOST="wooo@192.168.0.110"', "", 1)


def test_exact_seven_artifacts_and_target_modes_are_declared() -> None:
    text = deployer_text()
    for source in (
        "scripts/backup/backup-signoz.sh",
        "scripts/backup/clickhouse-native-backup.sh",
        "scripts/backup/clickhouse-native-restore-drill.sh",
        "scripts/backup/clickhouse-restore-inventory.py",
        "scripts/backup/run-signoz-backup-canary.sh",
        "ops/signoz/clickhouse/config.d/backup_disk.xml",
        "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml",
    ):
        assert source in text

    for target in (
        "${REMOTE_SCRIPT_DIR}/backup-signoz.sh",
        "${REMOTE_SCRIPT_DIR}/clickhouse-native-backup.sh",
        "${REMOTE_SCRIPT_DIR}/clickhouse-native-restore-drill.sh",
        "${REMOTE_SCRIPT_DIR}/clickhouse-restore-inventory.py",
        "${REMOTE_SCRIPT_DIR}/run-signoz-backup-canary.sh",
        "${REMOTE_CONFIG_ROOT}/config.d/backup_disk.xml",
        "${REMOTE_CONFIG_ROOT}/restore-drill/config.d/isolated-cluster.xml",
    ):
        assert target in text
    assert "MODES=(0755 0755 0755 0755 0755 0644 0644)" in text
    assert 'REMOTE_SCRIPT_DIR="${SIGNOZ_TOOLCHAIN_SCRIPT_DIR:-/backup/scripts}"' in text
    assert (
        'REMOTE_CONFIG_ROOT="${SIGNOZ_TOOLCHAIN_CONFIG_ROOT:-'
        '/backup/config/signoz/clickhouse}"'
    ) in text


def test_local_and_remote_stage_validation_cover_hash_syntax_python_and_xml() -> None:
    text = deployer_text()
    assert "validate_local_sources" in text
    assert "SOURCE_HASHES" in text
    assert "bash -n" in text
    assert "python3 -m py_compile" in text
    assert "validate_xml_pair" in text
    for expected in (
        '"/backups/"',
        '"restore-zookeeper"',
        '"restore-clickhouse"',
        '"CLICKHOUSE_RESTORE_REPLICA"',
    ):
        assert expected in text


def test_check_mode_exits_before_remote_stage_or_scp() -> None:
    text = deployer_text()
    check_exit = text.index('if [ "${MODE}" = --check ]')
    stage_create = text.index("STAGE_CREATED=0")
    first_scp = text.index("scp -q")
    assert check_exit < stage_create < first_scp
    assert "check_pass_no_write" in text
    assert "remote_target_unchanged_no_stage_no_candidate" in text

    remote_check = text[
        text.index("remote_read_only_check() {") : text.index(
            "local_receipt sensor_source", text.index("remote_read_only_check() {")
        )
    ]
    for forbidden in ("mkdir ", "install ", "mv ", "rm ", "scp "):
        assert forbidden not in remote_check


def test_check_mode_uses_only_ssh_and_never_scp(tmp_path: Path) -> None:
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    log = tmp_path / "calls.log"
    ssh = fake_bin / "ssh"
    ssh.write_text(
        "#!/bin/sh\n"
        'printf "ssh %s\\n" "$*" >> "$FAKE_CALL_LOG"\n'
        "cat >/dev/null\n"
        'printf \'{"phase":"source_of_truth_diff","result":"pass"}\\n\'\n',
        encoding="utf-8",
    )
    scp = fake_bin / "scp"
    scp.write_text(
        '#!/bin/sh\nprintf "scp %s\\n" "$*" >> "$FAKE_CALL_LOG"\nexit 99\n',
        encoding="utf-8",
    )
    ssh.chmod(0o755)
    scp.chmod(0o755)

    env = base_env()
    env["PATH"] = f"{fake_bin}:{env['PATH']}"
    env["FAKE_CALL_LOG"] = str(log)
    result = subprocess.run(
        ["bash", str(DEPLOYER), "--check"],
        check=False,
        capture_output=True,
        text=True,
        env=env,
    )
    assert result.returncode == 0, result.stderr
    calls = log.read_text(encoding="utf-8").splitlines()
    assert len(calls) == 1
    assert calls[0].startswith("ssh ")
    assert "wooo@192.168.0.110" in calls[0]
    assert "check_pass_no_write" in result.stdout


def test_hung_ssh_is_bounded_and_cannot_claim_check_pass(tmp_path: Path) -> None:
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    ssh = fake_bin / "ssh"
    ssh.write_text("#!/bin/sh\nsleep 5\n", encoding="utf-8")
    scp = fake_bin / "scp"
    scp.write_text("#!/bin/sh\nexit 99\n", encoding="utf-8")
    ssh.chmod(0o755)
    scp.chmod(0o755)

    env = base_env()
    env["PATH"] = f"{fake_bin}:{env['PATH']}"
    env["SIGNOZ_TOOLCHAIN_SSH_TIMEOUT_SECONDS"] = "1"
    env["SIGNOZ_TOOLCHAIN_TIMEOUT_KILL_AFTER_SECONDS"] = "1"
    started = time.monotonic()
    result = subprocess.run(
        ["bash", str(DEPLOYER), "--check"],
        check=False,
        capture_output=True,
        text=True,
        env=env,
        timeout=5,
    )
    elapsed = time.monotonic() - started

    assert result.returncode == 124
    assert elapsed < 4
    assert "check_pass_no_write" not in result.stdout


def test_all_ssh_scp_and_docker_calls_use_bounded_wrappers() -> None:
    text = deployer_text()
    executable = executable_text()

    assert (
        'SSH_COMMAND_TIMEOUT_SECONDS="${SIGNOZ_TOOLCHAIN_SSH_TIMEOUT_SECONDS:-300}"'
        in text
    )
    assert (
        'SCP_COMMAND_TIMEOUT_SECONDS="${SIGNOZ_TOOLCHAIN_SCP_TIMEOUT_SECONDS:-120}"'
        in text
    )
    assert (
        'DOCKER_COMMAND_TIMEOUT_SECONDS="${SIGNOZ_TOOLCHAIN_DOCKER_TIMEOUT_SECONDS:-30}"'
        in text
    )
    assert (
        'TIMEOUT_KILL_AFTER_SECONDS="${SIGNOZ_TOOLCHAIN_TIMEOUT_KILL_AFTER_SECONDS:-10}"'
        in text
    )
    assert "bounded_ssh()" in text
    assert "bounded_scp()" in text
    assert text.count('ssh "$@"') == 1
    assert text.count('scp "$@"') == 1
    assert executable.count("bounded_docker()") == 2
    assert "$(docker inspect" not in executable
    assert "$(docker exec" not in executable
    assert "\n  docker inspect" not in executable
    assert "\n  docker exec" not in executable
    assert executable.count("bounded_docker inspect") == 2
    assert executable.count("bounded_docker exec") == 2
    assert "timeout --signal=TERM" in executable


def test_apply_persists_exact_prior_metadata_and_has_full_rollback() -> None:
    text = deployer_text()
    for token in (
        "prior-manifest.tsv",
        "PRIOR_PRESENT",
        "PRIOR_MODES",
        "PRIOR_UIDS",
        "PRIOR_GIDS",
        "PRIOR_HASHES",
        "cp -p",
        "rollback_targets",
        "all_prior_files_modes_hashes_owners_and_runtime_restored",
    ):
        assert token in text
    assert "trap on_exit EXIT" in text
    assert "rc=91" in text
    assert 'run_root rm -f "${target}"' in text
    assert 'run_root rmdir "${CREATED_DIRS[index]}"' in text
    assert "bounded_execution_not_started_targets_unchanged" in text


def test_apply_uses_same_filesystem_candidates_and_accurate_replace_receipt() -> None:
    text = deployer_text()
    assert '"${target}.candidate.${RUN_ID}"' in text
    assert (
        'run_root mv -f "${TARGETS[index]}.candidate.${RUN_ID}" "${TARGETS[index]}"'
    ) in text
    assert "deployed-manifest.tsv" in text
    assert (
        "seven_same_filesystem_candidates_replaced_with_full_rollback_"
        "expected_hashes_and_modes"
    ) in text
    assert "seven_candidates_atomically_replaced" not in text
    assert "atomically replaces exact targets" not in text
    assert "stage_candidate_residue_0" in text
    assert "verify_residue_zero" in text


def test_apply_holds_dual_backup_locks_through_postverify_or_rollback() -> None:
    text = deployer_text()
    backup_text = BACKUP.read_text(encoding="utf-8")
    canary_text = CANARY.read_text(encoding="utf-8")
    canary_lock = "/tmp/awoooi-signoz-backup-canary.lock"
    operation_lock = "/tmp/awoooi-signoz-backup-operation.lock"

    assert f'CANARY_LOCK="{canary_lock}"' in text
    assert f'OPERATION_LOCK="{operation_lock}"' in text
    assert canary_lock in canary_text
    assert operation_lock in backup_text
    canary_acquire = text.index('exec 8>>"${CANARY_LOCK}"')
    operation_acquire = text.index('exec 7>>"${OPERATION_LOCK}"')
    active_check = text.index("if pgrep -af", operation_acquire)
    apply_started = text.index("APPLY_STARTED=1", operation_acquire)
    target_replace = text.index(
        'run_root mv -f "${TARGETS[index]}.candidate.${RUN_ID}"', apply_started
    )
    assert canary_acquire < operation_acquire < active_check < apply_started
    assert apply_started < target_replace

    on_exit = text[text.index("on_exit() {") : text.index("trap on_exit EXIT")]
    rollback = on_exit.index("rollback_targets")
    release_operation = on_exit.index("flock -u 7")
    release_canary = on_exit.index("flock -u 8")
    assert rollback < release_operation < release_canary
    assert 'rm -f "${OPERATION_LOCK}"' not in text
    assert 'rm -f "${CANARY_LOCK}"' not in text
    assert "another_signoz_backup_holds_operation_lock" in text
    assert "another_signoz_canary_holds_canary_lock" in text
    assert (
        "risk_medium_bounded_ssh_scp_docker_dual_signoz_canary_and_"
        "backup_operation_locks_"
        "no_runtime_execution_no_container_change_full_rollback"
    ) in text


def test_runtime_post_verifier_requires_exact_identity_and_lifecycle_parity() -> None:
    text = deployer_text()
    for container in (
        "signoz-clickhouse",
        "signoz-otel-collector",
        "signoz",
        "signoz-zookeeper-1",
    ):
        assert container in text
    for field in (
        ".Id",
        ".State.StartedAt",
        ".RestartCount",
        ".State.Running",
        'index .State "Health"',
    ):
        assert field in text
    assert "runtime-before.tsv" in text
    assert "runtime-after.tsv" in text
    assert 'cmp -s "${RUNTIME_BEFORE}" "${RUNTIME_AFTER}"' in text
    for port in (4317, 4318, 8080):
        assert "for port in 4317 4318 8080" in text
        assert str(port) in text
    assert "api_200_unchanged" in text


def test_apply_fails_closed_while_backup_or_restore_is_active() -> None:
    text = deployer_text()
    assert "active_signoz_backup_toolchain_process" in text
    assert "active_native_backup_or_restore" in text
    assert "CREATING_BACKUP" in text
    assert "RESTORING" in text
    assert "another_toolchain_deploy_holds_lock" in text


def test_deployer_never_executes_runtime_work_or_uses_forbidden_supply_chain() -> None:
    executable = executable_text().lower()
    for forbidden in (
        "docker restart",
        "docker compose",
        "docker pull",
        "docker run",
        "backup all",
        "restore all",
        "ansible-playbook",
        "github.com",
        "api.github.com",
        "raw.githubusercontent.com",
        "codeload.github.com",
        "ghcr.io",
        " gh ",
    ):
        assert forbidden not in executable
    assert "no_backup_restore_restart_or_pull" in executable


def test_receipts_cover_complete_controlled_apply_contract() -> None:
    text = deployer_text()
    for phase in (
        "sensor_source",
        "normalized_asset_identity",
        "source_of_truth_diff",
        "ai_decision",
        "risk_policy",
        "check",
        "execution",
        "post_verifier",
        "rollback",
        "closure_writeback",
        "terminal",
    ):
        assert phase in text
    assert "receipts.jsonl" in text
    assert '"${TRACE_ID}" "${RUN_ID}" "${WORK_ITEM_ID}"' in text
    assert "durable_manifests_and_receipts_ack" in text