from __future__ import annotations

import os
from pathlib import Path
import re
import subprocess
import xml.etree.ElementTree as ET


ROOT = Path(__file__).resolve().parents[3]
CONFIG = ROOT / "ops/signoz/clickhouse/config.d/backup_disk.xml"
OVERRIDE = ROOT / "ops/signoz/docker-compose.clickhouse-backup.override.yaml"
DEPLOYER = ROOT / "scripts/ops/deploy-signoz-clickhouse-backup-disk.sh"
BACKUP_SCRIPT = ROOT / "scripts/backup/backup-signoz.sh"


def write_executable(path: Path, text: str) -> None:
    path.write_text(text, encoding="utf-8")
    path.chmod(0o755)


def executable_text() -> str:
    return "\n".join(
        line
        for line in DEPLOYER.read_text(encoding="utf-8").splitlines()
        if not line.lstrip().startswith("#")
    )


def test_clickhouse_config_declares_one_allowed_local_backup_disk() -> None:
    root = ET.parse(CONFIG).getroot()
    assert root.tag == "clickhouse"

    disks = root.findall("./storage_configuration/disks/*")
    assert [disk.tag for disk in disks] == ["backups"]
    assert disks[0].findtext("type") == "local"
    assert disks[0].findtext("path") == "/backups/"

    backups = root.find("./backups")
    assert backups is not None
    assert backups.findtext("allowed_disk") == "backups"
    assert backups.findtext("allowed_path") == "/backups/"
    assert backups.findtext("allow_concurrent_backups") == "false"
    assert backups.findtext("allow_concurrent_restores") == "false"
    assert CONFIG.read_text(encoding="utf-8").count("<allowed_disk>") == 1


def test_compose_override_is_additive_clickhouse_only_and_uses_dedicated_mounts() -> (
    None
):
    text = OVERRIDE.read_text(encoding="utf-8")
    assert text.count("  clickhouse:") == 1
    assert "otel-collector:" not in text
    assert "signoz:" not in text
    assert "zookeeper:" not in text
    assert "source: /backup/staging/signoz-clickhouse" in text
    assert "target: /backups" in text
    assert (
        "source: /home/wooo/signoz/deploy/docker/awoooi-clickhouse-backup-disk.xml"
    ) in text
    assert ("target: /etc/clickhouse-server/config.d/awoooi-backup-disk.xml") in text
    assert text.count("read_only: true") == 1
    assert text.count("read_only: false") == 1
    for forbidden in ("image:", "build:", "command:", "environment:"):
        assert forbidden not in text


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


def test_deployer_requires_all_three_controlled_apply_identifiers_before_ssh() -> 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_check_mode_validates_candidate_via_stdin_without_remote_staging() -> None:
    text = DEPLOYER.read_text(encoding="utf-8")
    check_terminal = text.index('if [ "${MODE}" = "--check" ]')
    stage_assignment = text.index('STAGE_DIR="/tmp/awoooi-signoz-clickhouse-backup.')
    first_transfer = text.index('bounded_scp "${LOCAL_OVERRIDE}"')

    assert check_terminal < stage_assignment < first_transfer
    assert "-f '${REMOTE_BASE}' -f - config -q" in text
    assert '< "${LOCAL_OVERRIDE}"' in text
    assert "check_pass_no_write" in text


def test_apply_is_single_service_bounded_no_pull_and_env_isolated() -> None:
    text = DEPLOYER.read_text(encoding="utf-8")
    executable = executable_text()

    assert 'REMOTE_SERVICE="clickhouse"' in text
    assert "--env-file /dev/null" in text
    assert 'env -i PATH="${SAFE_PATH}" HOME="${SAFE_HOME}"' in text
    assert "bounded_ssh" in text
    assert "bounded_scp" in text
    assert "docker_cmd" in text
    assert "compose_config_cmd" in text
    assert "compose_apply_cmd" in text
    assert "REMOTE_APPLY_TIMEOUT_SECONDS=3600" in text
    assert "up -d --no-deps --pull never --force-recreate" in text
    assert '"${REMOTE_SERVICE}"' in text
    assert "docker compose down" not in executable
    assert "docker pull" not in executable
    assert "docker volume" not in executable
    assert "docker system prune" not in executable
    assert "github.com" not in text.lower()
    assert "ghcr.io" not in text.lower()


def test_apply_fails_closed_when_backup_is_active_or_managed_files_drift() -> None:
    text = DEPLOYER.read_text(encoding="utf-8")
    assert "active_signoz_backup_or_restore_process_detected" in text
    assert "active_native_backup_or_restore" in text
    assert "signoz_backup_process_probe_failed" in text
    assert "partial_managed_file_drift" in text
    assert "another_deploy_holds_lock" in text
    for process in (
        "backup-signoz",
        "clickhouse-native-backup",
        "clickhouse-native-restore-drill",
        "run-signoz-backup-canary",
    ):
        assert process in text
    probe_block = text[
        text.index("require_backup_idle()") : text.index(
            "require_backup_idle preflight"
        )
    ]
    assert "process_rc=$?" in probe_block
    assert 'case "${process_rc}" in' in probe_block
    assert "1) ;;" in probe_block
    assert "return 69" in probe_block
    assert "CREATING_BACKUP" in text
    assert "RESTORING" in text


def test_target_host_is_fixed_and_ignores_ambient_override(tmp_path: Path) -> None:
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    ssh_log = tmp_path / "ssh.log"
    write_executable(
        fake_bin / "ssh",
        """#!/bin/bash
printf '%s\\n' "$*" >> "${FAKE_SSH_LOG:?}"
cat >/dev/null
""",
    )
    write_executable(fake_bin / "scp", "#!/bin/bash\nexit 0\n")
    env = os.environ.copy()
    env.update(
        {
            "PATH": f"{fake_bin}:{env['PATH']}",
            "TRACE_ID": "P0-OBS-002-test",
            "RUN_ID": "fixed-host-test",
            "WORK_ITEM_ID": "P0-OBS-002",
            "TARGET_HOST": "attacker@example.invalid",
            "FAKE_SSH_LOG": str(ssh_log),
        }
    )

    result = subprocess.run(
        ["bash", str(DEPLOYER), "--check"],
        check=False,
        capture_output=True,
        text=True,
        env=env,
    )

    assert result.returncode == 0, result.stderr
    invocations = ssh_log.read_text(encoding="utf-8").splitlines()
    assert len(invocations) == 2
    assert all("wooo@192.168.0.110" in line for line in invocations)
    assert all("attacker@example.invalid" not in line for line in invocations)


def test_all_transport_and_remote_docker_operations_are_bounded() -> None:
    text = DEPLOYER.read_text(encoding="utf-8")
    executable = executable_text()

    assert executable.count(' ssh "${SSH_OPTS[@]}"') == 1
    assert executable.count(' scp -q "${SSH_OPTS[@]}"') == 1
    assert not re.search(r"\bdocker (inspect|exec)\b", executable)
    assert 'timeout --signal=TERM --kill-after="${KILL_AFTER_SECONDS}"' in text
    assert 'docker_cmd exec "${CLICKHOUSE_CONTAINER}"' in text
    assert "ss_cmd -H -lnt" in text
    for mutation in (
        "bounded_cmd cp -p",
        "bounded_cmd install -m",
        "bounded_cmd mv -f",
        "bounded_cmd sudo -n install",
        "bounded_cmd rm -f",
    ):
        assert mutation in text


def test_stage_cleanup_failure_changes_apply_terminal_to_failed(tmp_path: Path) -> None:
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    ssh_count = tmp_path / "ssh-count"
    write_executable(
        fake_bin / "ssh",
        """#!/bin/bash
count=0
if [ -f "${FAKE_SSH_COUNT:?}" ]; then
  count="$(cat "${FAKE_SSH_COUNT}")"
fi
count=$((count + 1))
printf '%s\\n' "${count}" > "${FAKE_SSH_COUNT}"
cat >/dev/null
if [ "${count}" -eq 5 ]; then
  exit 1
fi
""",
    )
    write_executable(fake_bin / "scp", "#!/bin/bash\nexit 0\n")
    env = os.environ.copy()
    env.update(
        {
            "PATH": f"{fake_bin}:{env['PATH']}",
            "TRACE_ID": "P0-OBS-002-test",
            "RUN_ID": "cleanup-failure-test",
            "WORK_ITEM_ID": "P0-OBS-002",
            "FAKE_SSH_COUNT": str(ssh_count),
        }
    )

    result = subprocess.run(
        ["bash", str(DEPLOYER), "--apply"],
        check=False,
        capture_output=True,
        text=True,
        env=env,
    )

    assert result.returncode == 92
    assert '"phase":"cleanup","result":"failed"' in result.stdout
    assert '"phase":"terminal","result":"failed"' in result.stdout
    assert "failed_or_residue_present" in result.stdout


def test_apply_preserves_image_data_volume_and_non_target_containers() -> None:
    text = DEPLOYER.read_text(encoding="utf-8")
    assert "BEFORE_IMAGE_ID" in text
    assert "AFTER_IMAGE_ID" in text
    assert "BEFORE_DATA_VOLUME" in text
    assert "AFTER_DATA_VOLUME" in text
    assert 'eq .Destination "/var/lib/clickhouse"' in text
    for prefix in ("COLLECTOR", "SIGNOZ", "ZOOKEEPER"):
        assert f"BEFORE_{prefix}_ID" in text
        assert f"BEFORE_{prefix}_RESTARTS" in text
    assert "image_and_data_volume_unchanged" in text


def test_post_verifier_checks_disk_mount_health_listeners_and_api() -> None:
    text = DEPLOYER.read_text(encoding="utf-8")
    assert "system.disks WHERE name='backups'" in text
    assert "Local|0|0|0|1" in text
    assert "EXISTS TABLE system.backups" in text
    assert "SHOW GRANTS" in text
    assert "BACKUP([,[:space:]]|$)" in text
    assert 'eq .Destination "/backups"' in text
    assert "101:101:750" in text
    assert "server_uid" in text
    assert "server_gid" in text
    assert "docker_cmd exec -u 101:101" in text
    assert "test -w /backups" in text
    assert "awk '/^Uid:/{print $2}' /proc/1/status" in text
    assert "awk '/^Gid:/{print $2}' /proc/1/status" in text
    identity_start = text.index('server_uid="')
    identity_block = text[
        identity_start : text.index("docker_cmd exec -u 101:101", identity_start)
    ]
    assert "sh -c" not in identity_block
    assert "concurrency_disabled" in text
    for port in (4317, 4318, 8080):
        assert f"listener_up {port}" in text
    assert "api_code" in text
    assert "disk_backups_local_rw_healthy" in text


def test_failed_apply_restores_prior_managed_compose_state() -> None:
    text = DEPLOYER.read_text(encoding="utf-8")
    assert "restore_managed_files" in text
    assert "rollback_deploy" in text
    assert "PRIOR_OVERRIDE_PRESENT" in text
    assert "PRIOR_CONFIG_PRESENT" in text
    assert "prior-override.yaml" in text
    assert "prior-backup-disk.xml" in text
    assert "prior_compose_state_restored" in text
    assert "retained_nonempty_no_delete" in text
    assert "rc=91" in text
    assert "trap on_exit EXIT" in text
    for token in (
        "PRIOR_OVERRIDE_HASH",
        "PRIOR_OVERRIDE_MODE",
        "PRIOR_OVERRIDE_UID",
        "PRIOR_OVERRIDE_GID",
        "PRIOR_CONFIG_HASH",
        "PRIOR_CONFIG_MODE",
        "PRIOR_CONFIG_UID",
        "PRIOR_CONFIG_GID",
        "file_matches_metadata",
        "restore_exact_file",
        "prior_compose_state_restored_exact_hash_mode_uid_gid",
    ):
        assert token in text
    assert "exact_stage_candidate_and_rollback_residue_absence_verified" in text


def test_deployer_and_backup_execution_share_operation_lock_contract() -> None:
    deployer = DEPLOYER.read_text(encoding="utf-8")
    backup = BACKUP_SCRIPT.read_text(encoding="utf-8")
    operation_lock = "/tmp/awoooi-signoz-backup-operation.lock"
    canary_lock = "/tmp/awoooi-signoz-backup-canary.lock"

    assert operation_lock in deployer
    assert operation_lock in backup
    assert canary_lock in deployer
    assert 'exec 8>>"${BACKUP_OPERATION_LOCK_FILE}"' in deployer
    assert 'exec 7>>"${CANARY_LOCK_FILE}"' in deployer
    assert "flock -n 8 || { receipt check failed backup_operation_lock_held" in deployer
    assert "flock -n 7 || { receipt check failed backup_canary_lock_held" in deployer
    assert 'exec 8>>"${OPERATION_LOCK_FILE}"' in backup
    assert "require_backup_idle pre_mutation" in deployer
    assert deployer.index('exec 7>>"${CANARY_LOCK_FILE}"') < deployer.index(
        'exec 8>>"${BACKUP_OPERATION_LOCK_FILE}"'
    )
    assert deployer.index('exec 8>>"${BACKUP_OPERATION_LOCK_FILE}"') < deployer.index(
        "require_backup_idle pre_mutation"
    )
    remote_cleanup_call = deployer.rindex("    cleanup_owned_residue")
    assert remote_cleanup_call < deployer.index(
        'receipt terminal "${terminal}"', remote_cleanup_call
    )


def test_deployer_never_reads_runtime_environment_or_executes_backup() -> None:
    executable = executable_text()
    lowered = executable.lower()
    assert ".config.env" not in lowered
    assert "config.env" not in lowered
    assert ".env" not in lowered
    assert "docker inspect --format '{{json .config.env}}'" not in lowered
    assert " backup all" not in lowered
    assert " restore all" not in lowered
    assert "clickhouse-client --query 'backup" not in lowered
    assert "clickhouse-client --query 'restore" not in lowered
    assert "no_backup_or_restore_executed" in executable


def test_receipts_cover_controlled_apply_and_durable_terminal() -> None:
    text = DEPLOYER.read_text(encoding="utf-8")
    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 "override_sha_" in text
    assert "config_sha_" in text