from __future__ import annotations

import hashlib
import json
import os
import shutil
import subprocess
import zipfile
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "clickhouse-native-restore-drill.sh"
INVENTORY_HELPER = ROOT / "clickhouse-restore-inventory.py"
REPO_ROOT = ROOT.parents[1]
CLICKHOUSE_IMAGE_ID = (
    "sha256:8248e5926d7304400e44ecdf0c7181fbde28e6a9b1d647780eca839f34c00cc6"
)
ZOOKEEPER_IMAGE_ID = (
    "sha256:3ab0e8f032ab58f14ac1e929ac40305a4a464cf320bdfe4151d62d6922729ba0"
)
EMPTY_SHA256 = "E3B0C44298FC1C149AFBF4C8996FB92427AE41E4649B934CA495991B7852B855"
EXPECTED_DATABASES = [
    "signoz_analytics",
    "signoz_logs",
    "signoz_metadata",
    "signoz_meter",
    "signoz_metrics",
    "signoz_traces",
]


def _inventory(
    *,
    restored_delta: int = 0,
    engine_mismatch: bool = False,
    schema_mismatch: bool = False,
) -> str:
    rows = [
        ("signoz_analytics", "events", "ReplicatedMergeTree", 10, 100),
        ("signoz_analytics", "events_all", "Distributed", 10, 100),
        ("signoz_logs", "logs_v2", "ReplicatedMergeTree", 20, 200),
        ("signoz_metadata", "resource_attrs", "ReplicatedMergeTree", 2, 20),
        ("signoz_meter", "meter", "ReplicatedMergeTree", 3, 30),
        ("signoz_metrics", "samples_v4", "ReplicatedMergeTree", 30, 300),
        ("signoz_metrics", "time_series_v4", "ReplicatedMergeTree", 12, 120),
        ("signoz_traces", "signoz_index_v3", "ReplicatedMergeTree", 40, 400),
    ]
    rendered: list[str] = []
    for index, (database, table, engine, row_count, byte_count) in enumerate(rows):
        if engine_mismatch and table == "meter":
            engine = "MergeTree"
        schema_hash = (
            hashlib.sha256(f"{database}.{table}:{engine}".encode("utf-8"))
            .hexdigest()
            .upper()
        )
        if schema_mismatch and table == "meter":
            schema_hash = "F" * 64
        rendered.append(
            "\t".join(
                (
                    database,
                    table,
                    engine,
                    str(row_count + (restored_delta if index == 0 else 0)),
                    str(byte_count + (restored_delta if index == 0 else 0)),
                    schema_hash,
                )
            )
        )
    return "\n".join(rendered) + "\n"


def _write_backup_fixture(tmp_path: Path) -> tuple[Path, Path, Path]:
    artifact = tmp_path / "clickhouse-native.zip.partial"
    with zipfile.ZipFile(artifact, "w") as archive:
        archive.writestr(".backup", "bounded native backup fixture\n")
    inventory = tmp_path / "source-inventory.tsv.partial"
    inventory.write_text(_inventory(), encoding="utf-8")
    artifact_sha = hashlib.sha256(artifact.read_bytes()).hexdigest()
    inventory_sha = hashlib.sha256(inventory.read_bytes()).hexdigest()
    manifest = tmp_path / "manifest.json.partial"
    manifest.write_text(
        json.dumps(
            {
                "trace_id": "backup-trace",
                "run_id": "backup-run",
                "work_item_id": "P0-OBS-002",
                "operation_id": "awoooi-backup-operation",
                "databases": EXPECTED_DATABASES,
                "source_inventory_sha256": inventory_sha,
                "sha256": artifact_sha,
                "size_bytes": artifact.stat().st_size,
                "status": "BACKUP_CREATED",
            },
            separators=(",", ":"),
        )
        + "\n",
        encoding="utf-8",
    )
    return artifact, manifest, inventory


def _write_fake_docker(tmp_path: Path) -> tuple[Path, Path, Path]:
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    state_dir = tmp_path / "docker-state"
    state_dir.mkdir()
    log_path = tmp_path / "docker.log"
    docker = bin_dir / "docker"
    docker.write_text(
        r"""#!/usr/bin/env bash
set -euo pipefail
printf '%s\n' "$*" >> "${FAKE_DOCKER_LOG:?}"
state="${FAKE_DOCKER_STATE:?}"
cmd="${1:?}"
shift
last_arg() { eval "printf '%s' \"\${$#}\""; }

case "$cmd" in
  info)
    [ "${FAKE_DOCKER_INFO_FAIL:-0}" = 0 ] || exit 78
    printf '25.0.0\\n'
    ;;
  image)
    sub="${1:?}"
    shift
    [ "$sub" = inspect ] || exit 90
    image="$(last_arg "$@")"
    case "$image" in
      clickhouse/clickhouse-server:25.5.6) printf '%s\n' "${FAKE_CLICKHOUSE_IMAGE_ID:?}" ;;
      signoz/zookeeper:3.7.1) printf '%s\n' "${FAKE_ZOOKEEPER_IMAGE_ID:?}" ;;
      *) exit 1 ;;
    esac
    ;;
  container)
    sub="${1:?}"
    shift
    case "$sub" in
      inspect)
        format=""
        if [ "${1:-}" = --format ]; then
          format="$2"
          shift 2
        fi
        name="${1:?}"
        [ -f "$state/container-$name" ] || exit 1
        if [[ "$format" == *Labels* ]]; then
          if [ "${FAKE_LABEL_INSPECT_HANG:-0}" = 1 ] && [[ "$name" == *ch-restore* ]]; then
            sleep 10
          fi
          cat "$state/container-label-$name"
        fi
        ;;
      ls)
        for item in "$state"/container-*; do
          [ -e "$item" ] || continue
          case "$item" in
            *-label-*|*-network-*|*-image-*) continue ;;
          esac
          basename "$item" | sed 's/^container-//'
        done
        ;;
      *) exit 90 ;;
    esac
    ;;
  inspect)
    format=""
    if [ "${1:-}" = --format ]; then
      format="$2"
      shift 2
    fi
    name="${1:?}"
    [ -f "$state/container-$name" ] || exit 1
    case "$format" in
      *State.Running*) printf 'true\n' ;;
      *HostConfig.NetworkMode*) cat "$state/container-network-$name" ;;
      *.Image*) cat "$state/container-image-$name" ;;
      *) printf '%s\n' "fake-$name" ;;
    esac
    ;;
  network)
    sub="${1:?}"
    shift
    case "$sub" in
      create)
        name="$(last_arg "$@")"
        label=""
        while [ "$#" -gt 0 ]; do
          if [ "$1" = --label ]; then label="${2#*=}"; shift 2; else shift; fi
        done
        touch "$state/network-$name"
        printf '%s\n' "$label" > "$state/network-label-$name"
        printf '%s\n' "$name" > "$state/network-name"
        printf '%s\n' "$name"
        ;;
      inspect)
        format=""
        if [ "${1:-}" = --format ]; then
          format="$2"
          shift 2
        fi
        name="${1:?}"
        [ -f "$state/network-$name" ] || exit 1
        if [[ "$format" == *Internal* ]]; then printf 'true\n'; fi
        if [[ "$format" == *Labels* ]]; then cat "$state/network-label-$name"; fi
        ;;
      ls)
        for item in "$state"/network-*; do
          [ -e "$item" ] || continue
          case "$item" in *-label-*|*/network-name) continue ;; esac
          basename "$item" | sed 's/^network-//'
        done
        ;;
      rm)
        name="${1:?}"
        rm -f "$state/network-$name" "$state/network-label-$name" "$state/network-name"
        printf '%s\n' "$name"
        ;;
      *) exit 90 ;;
    esac
    ;;
  volume)
    sub="${1:?}"
    shift
    case "$sub" in
      create)
        name="$(last_arg "$@")"
        label=""
        while [ "$#" -gt 0 ]; do
          if [ "$1" = --label ]; then label="${2#*=}"; shift 2; else shift; fi
        done
        touch "$state/volume-$name"
        printf '%s\n' "$label" > "$state/volume-label-$name"
        printf '%s\n' "$name"
        ;;
      inspect)
        format=""
        if [ "${1:-}" = --format ]; then format="$2"; shift 2; fi
        name="${1:?}"
        [ -f "$state/volume-$name" ] || exit 1
        if [[ "$format" == *Labels* ]]; then cat "$state/volume-label-$name"; fi
        ;;
      ls)
        for item in "$state"/volume-*; do
          [ -e "$item" ] || continue
          case "$item" in *-label-*) continue ;; esac
          basename "$item" | sed 's/^volume-//'
        done
        ;;
      rm)
        name="${1:?}"
        rm -f "$state/volume-$name" "$state/volume-label-$name"
        if [[ "$name" == *-backup ]]; then
          rm -f "$state/staged-artifact-sha256"
        fi
        printf '%s\n' "$name"
        ;;
      *) exit 90 ;;
    esac
    ;;
  run)
    name=""
    replica=""
    label=""
    network=""
    image=""
    expected_artifact_sha256=""
    while [ "$#" -gt 0 ]; do
      case "$1" in
        --name)
          name="$2"
          shift 2
          ;;
        --env)
          if [[ "$2" == CLICKHOUSE_RESTORE_REPLICA=* ]]; then
            replica="${2#*=}"
          fi
          if [[ "$2" == EXPECTED_ARTIFACT_SHA256=* ]]; then
            expected_artifact_sha256="${2#*=}"
          fi
          shift 2
          ;;
        --label)
          label="${2#*=}"
          shift 2
          ;;
        --network)
          network="$2"
          shift 2
          ;;
        --network-alias|--restart|--mount|--user|--entrypoint)
          shift 2
          ;;
        --detach|--pull=never)
          shift
          ;;
        clickhouse/clickhouse-server:25.5.6|signoz/zookeeper:3.7.1)
          image="$1"
          shift
          ;;
        *)
          shift
          ;;
      esac
    done
    [ -n "$name" ] || exit 91
    touch "$state/container-$name"
    printf '%s\n' "$label" > "$state/container-label-$name"
    printf '%s\n' "$network" > "$state/container-network-$name"
    case "$image" in
      clickhouse/clickhouse-server:25.5.6)
        printf '%s\n' "${FAKE_CLICKHOUSE_IMAGE_ID:?}" > "$state/container-image-$name"
        ;;
      signoz/zookeeper:3.7.1)
        printf '%s\n' "${FAKE_ZOOKEEPER_IMAGE_ID:?}" > "$state/container-image-$name"
        ;;
      *) exit 96 ;;
    esac
    if [ -n "$replica" ]; then printf '%s\n' "$replica" > "$state/replica"; fi
    if [ "$network" = none ]; then
      if [ "${FAKE_STAGING_HANG:-0}" = 1 ]; then sleep 10; fi
      [ "${FAKE_STAGING_FAIL:-0}" = 0 ] || exit 94
      staged_hash="$expected_artifact_sha256"
      if [ "${FAKE_STAGING_HASH_MISMATCH:-0}" = 1 ]; then
        staged_hash="$(printf '0%.0s' {1..64})"
      fi
      printf '%s\n' "$staged_hash" > "$state/staged-artifact-sha256"
      printf '%s\n' "$staged_hash"
    else
      printf 'fake-container-%s\n' "$name"
    fi
    ;;
  exec)
    container="${1:?}"
    shift
    [ -f "$state/container-$container" ] || exit 1
    if [ "${1:-}" = sha256sum ]; then
      [ -f "$state/staged-artifact-sha256" ] || exit 97
      artifact_hash="$(cat "$state/staged-artifact-sha256")"
      if [ "${FAKE_CLICKHOUSE_HASH_MISMATCH:-0}" = 1 ]; then
        artifact_hash="$(printf 'f%.0s' {1..64})"
      fi
      printf '%s  /backups/native-backup.zip\n' "$artifact_hash"
      exit 0
    fi
    query=""
    operation=""
    while [ "$#" -gt 0 ]; do
      case "$1" in
        --query)
          query="$2"
          shift 2
          ;;
        --param_operation_id)
          operation="$2"
          shift 2
          ;;
        *) shift ;;
      esac
    done
    case "$query" in
      CHECK\ TABLE*)
        if [ "${FAKE_CHECK_TABLE_FAIL:-0}" = 1 ]; then printf '0\n'; else printf '1\n'; fi
        ;;
      RESTORE\ *)
        case "$query" in
          *" SETTINGS id='${EXPECTED_FAKE_RESTORE_OPERATION_ID:?}' ASYNC")
            operation="${EXPECTED_FAKE_RESTORE_OPERATION_ID}"
            ;;
          *) exit 93 ;;
        esac
        touch "$state/restored"
        printf '%s\tRESTORING\n' "$operation"
        ;;
      *FROM\ system.backups*)
        printf "Disk('backups', 'native-backup.zip')\tRESTORED\t100\t1000\t%s\n" "${EMPTY_SHA256:?}"
        ;;
      *FROM\ system.disks*) printf '/backups/\n' ;;
      *FROM\ system.macros*)
        printf 'replica\t%s\nshard\t01\n' "$(cat "$state/replica")"
        ;;
      *FROM\ system.clusters*) printf 'cluster\trestore-clickhouse\t9000\n' ;;
      *FROM\ system.zookeeper*) printf '1\n' ;;
      *"SELECT count() FROM system.databases"*) printf '0\n' ;;
      *"SELECT count() FROM system.tables"*) printf '0\n' ;;
      *"SELECT database,name,engine"*) cat "${FAKE_RESTORED_INVENTORY:?}" ;;
      "SELECT 1") printf '1\n' ;;
      *) printf 'UNHANDLED QUERY: %s\n' "$query" >&2; exit 92 ;;
    esac
    ;;
  logs)
    printf 'bounded fake container log\n'
    ;;
  rm)
    if [ "${1:-}" = -f ]; then shift; fi
    name="${1:?}"
    rm -f "$state/container-$name" "$state/container-label-$name" \
      "$state/container-network-$name" "$state/container-image-$name"
    printf '%s\n' "$name"
    ;;
  *) exit 99 ;;
esac
""",
        encoding="utf-8",
    )
    docker.chmod(0o755)
    return bin_dir, state_dir, log_path


def _environment(
    tmp_path: Path,
    artifact: Path,
    manifest: Path,
    inventory: Path,
    restored_inventory: Path,
) -> dict[str, str]:
    bin_dir, state_dir, log_path = _write_fake_docker(tmp_path)
    return os.environ | {
        "PATH": f"{bin_dir}:{os.environ['PATH']}",
        "TRACE_ID": "trace-restore-test",
        "RUN_ID": "run-restore-test",
        "WORK_ITEM_ID": "P0-OBS-002",
        "CLICKHOUSE_NATIVE_ARTIFACT_PATH": str(artifact),
        "CLICKHOUSE_NATIVE_MANIFEST_PATH": str(manifest),
        "CLICKHOUSE_NATIVE_SOURCE_INVENTORY_PATH": str(inventory),
        "CLICKHOUSE_NATIVE_ARTIFACT_SHA256": hashlib.sha256(
            artifact.read_bytes()
        ).hexdigest(),
        "CLICKHOUSE_NATIVE_OPERATION_ID": "awoooi-backup-operation",
        "CLICKHOUSE_NATIVE_EXPECTED_DATABASES": ",".join(EXPECTED_DATABASES),
        "CLICKHOUSE_RESTORE_RECEIPT_ROOT": str(tmp_path / "receipts"),
        "CLICKHOUSE_RESTORE_POLL_INTERVAL_SECONDS": "1",
        "CLICKHOUSE_RESTORE_STARTUP_TIMEOUT_SECONDS": "5",
        "CLICKHOUSE_RESTORE_TIMEOUT_SECONDS": "5",
        "CLICKHOUSE_RESTORE_COMMAND_TIMEOUT_SECONDS": "5",
        "FAKE_DOCKER_LOG": str(log_path),
        "FAKE_DOCKER_STATE": str(state_dir),
        "FAKE_CLICKHOUSE_IMAGE_ID": CLICKHOUSE_IMAGE_ID,
        "FAKE_ZOOKEEPER_IMAGE_ID": ZOOKEEPER_IMAGE_ID,
        "FAKE_RESTORED_INVENTORY": str(restored_inventory),
        "EXPECTED_FAKE_RESTORE_OPERATION_ID": "restore-"
        + hashlib.sha256(
            b"trace-restore-test\0run-restore-test\0P0-OBS-002"
        ).hexdigest(),
        "EMPTY_SHA256": EMPTY_SHA256,
    }


def _run(mode: str, env: dict[str, str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["bash", str(SCRIPT), mode],
        text=True,
        capture_output=True,
        env=env,
    )


def _status(tmp_path: Path) -> dict[str, object]:
    return json.loads(
        (tmp_path / "receipts" / "run-restore-test" / "status.json").read_text(
            encoding="utf-8"
        )
    )


def _assert_ephemeral_resources_absent(state_dir: Path) -> None:
    assert not list(state_dir.glob("container-*"))
    assert not list(state_dir.glob("volume-*"))
    assert not list(state_dir.glob("network-*"))
    assert not (state_dir / "staged-artifact-sha256").exists()


def test_hook_check_uses_env_contract_without_runtime_creation(tmp_path: Path) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(restored_delta=3), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)

    result = _run("--check", env)

    assert result.returncode == 0, result.stderr
    assert "terminal=check_pass_no_runtime_write" in result.stdout
    docker_log = Path(env["FAKE_DOCKER_LOG"]).read_text(encoding="utf-8")
    assert "image inspect" in docker_log
    assert "network create" not in docker_log
    assert "volume create" not in docker_log
    assert not any(line.startswith("run ") for line in docker_log.splitlines())
    status = _status(tmp_path)
    assert status["check_pass"] is True
    assert status["ephemeral_runtime_created"] is False
    assert status["production_network_attached"] is False
    assert status["production_restore_performed"] is False


def test_apply_restores_on_internal_pair_and_cleans_every_resource(
    tmp_path: Path,
) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(restored_delta=3), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)
    check = _run("--check", env)
    assert check.returncode == 0, check.stderr

    result = _run("--apply", env)

    assert result.returncode == 0, result.stderr
    assert "terminal=verified" in result.stdout
    docker_lines = Path(env["FAKE_DOCKER_LOG"]).read_text(encoding="utf-8").splitlines()
    network_create = next(
        line for line in docker_lines if line.startswith("network create")
    )
    assert "--internal" in network_create
    run_lines = [line for line in docker_lines if line.startswith("run ")]
    assert len(run_lines) == 3
    assert all("--pull=never" in line for line in run_lines)
    assert all("--publish" not in line and " -p " not in line for line in run_lines)

    staging_run = next(line for line in run_lines if "--network none" in line)
    zookeeper_run = next(line for line in run_lines if "signoz/zookeeper:3.7.1" in line)
    clickhouse_run = next(
        line for line in run_lines if "--network-alias restore-clickhouse" in line
    )
    assert "--user 0:0" in staging_run
    assert "--entrypoint /bin/sh" in staging_run
    assert (
        f"type=bind,src={artifact},dst=/tmp/awoooi-source-native-backup.zip,readonly"
        in staging_run
    )
    assert sum(f"src={artifact}" in line for line in run_lines) == 1
    assert "type=volume,src=awoooi-ch-restore-" in staging_run
    assert "-backup,dst=/backups" in staging_run
    assert "ALLOW_ANONYMOUS_LOGIN=yes" in zookeeper_run
    assert "--network awoooi-ch-restore-" in zookeeper_run
    assert "--network awoooi-ch-restore-" in clickhouse_run
    assert str(artifact) not in clickhouse_run
    assert "type=volume,src=awoooi-ch-restore-" in clickhouse_run
    assert "-backup,dst=/backups" in clickhouse_run
    assert "dst=/backups,readonly" not in clickhouse_run
    assert (
        "dst=/etc/clickhouse-server/config.d/awoooi-backup-disk.xml,readonly"
        in clickhouse_run
    )
    assert (
        "dst=/etc/clickhouse-server/config.d/awoooi-restore-isolation.xml,readonly"
        in clickhouse_run
    )
    hash_readback = next(
        line
        for line in docker_lines
        if line.startswith("exec ") and "sha256sum /backups/native-backup.zip" in line
    )
    assert "exec awoooi-ch-restore-" in hash_readback
    restore_line = next(line for line in docker_lines if "RESTORE DATABASE" in line)
    assert "allow_non_empty_tables" not in restore_line
    assert " SETTINGS id='restore-" in restore_line
    assert "id={operation_id:String}" not in restore_line
    macro_line = next(line for line in docker_lines if "FROM system.macros" in line)
    assert "SELECT macro,substitution" in macro_line
    assert "SELECT name,value" not in macro_line
    state_dir = Path(env["FAKE_DOCKER_STATE"])
    _assert_ephemeral_resources_absent(state_dir)
    status = _status(tmp_path)
    assert status["terminal"] == "verified"
    assert status["check_pass"] is True
    assert status["restore_terminal"] == "RESTORED"
    assert status["manifest_parity"] == 1
    assert status["check_table_count"] == status["check_table_pass_count"] == 7
    assert status["critical_nonzero_count"] == 4
    assert status["row_delta"] == 3
    expected_hash = hashlib.sha256(artifact.read_bytes()).hexdigest()
    assert status["artifact_sha256"] == expected_hash
    assert status["staged_artifact_sha256"] == expected_hash
    assert status["clickhouse_artifact_sha256"] == expected_hash
    assert status["artifact_staging_verified"] is True
    assert status["clickhouse_artifact_readback_verified"] is True
    assert status["staging_network_mode"] == "none"
    assert status["artifact_source_mount"] == "read_only_staging_only"
    assert status["backup_volume_mount"] == "writable_ephemeral"
    assert status["cleanup_verified"] is True


def test_apply_failure_still_cleans_pair_network_and_volumes(tmp_path: Path) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)
    assert _run("--check", env).returncode == 0
    env["FAKE_CHECK_TABLE_FAIL"] = "1"

    result = _run("--apply", env)

    assert result.returncode == 1
    assert "terminal=failed" in result.stdout
    state_dir = Path(env["FAKE_DOCKER_STATE"])
    _assert_ephemeral_resources_absent(state_dir)
    status = _status(tmp_path)
    assert status["terminal"] == "failed"
    assert status["cleanup_verified"] is True
    assert status["apply_pass"] is False


def test_artifact_staging_failure_cleans_exact_resources(tmp_path: Path) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)
    assert _run("--check", env).returncode == 0
    env["FAKE_STAGING_FAIL"] = "1"

    result = _run("--apply", env)

    assert result.returncode == 1
    _assert_ephemeral_resources_absent(Path(env["FAKE_DOCKER_STATE"]))
    status = _status(tmp_path)
    assert status["terminal"] == "failed"
    assert status["reason"] == "isolated_artifact_staging_failed"
    assert status["artifact_staging_verified"] is False
    assert status["cleanup_verified"] is True


def test_artifact_staging_hang_is_bounded_and_cleans_exact_resources(
    tmp_path: Path,
) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)
    env["CLICKHOUSE_RESTORE_COMMAND_TIMEOUT_SECONDS"] = "1"
    assert _run("--check", env).returncode == 0
    env["FAKE_STAGING_HANG"] = "1"

    result = subprocess.run(
        ["bash", str(SCRIPT), "--apply"],
        text=True,
        capture_output=True,
        env=env,
        timeout=12,
    )

    assert result.returncode == 1
    _assert_ephemeral_resources_absent(Path(env["FAKE_DOCKER_STATE"]))
    status = _status(tmp_path)
    assert status["terminal"] == "failed"
    assert status["reason"] == "isolated_artifact_staging_failed"
    assert status["artifact_staging_verified"] is False
    assert status["cleanup_verified"] is True


def test_staged_artifact_hash_mismatch_fails_before_isolated_services(
    tmp_path: Path,
) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)
    assert _run("--check", env).returncode == 0
    env["FAKE_STAGING_HASH_MISMATCH"] = "1"

    result = _run("--apply", env)

    assert result.returncode == 1
    docker_lines = Path(env["FAKE_DOCKER_LOG"]).read_text(encoding="utf-8").splitlines()
    assert not any(
        "signoz/zookeeper:3.7.1" in line
        for line in docker_lines
        if line.startswith("run ")
    )
    _assert_ephemeral_resources_absent(Path(env["FAKE_DOCKER_STATE"]))
    status = _status(tmp_path)
    assert status["reason"] == "staged_artifact_hash_mismatch"
    assert status["cleanup_verified"] is True


def test_clickhouse_artifact_hash_readback_mismatch_fails_and_cleans(
    tmp_path: Path,
) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)
    assert _run("--check", env).returncode == 0
    env["FAKE_CLICKHOUSE_HASH_MISMATCH"] = "1"

    result = _run("--apply", env)

    assert result.returncode == 1
    _assert_ephemeral_resources_absent(Path(env["FAKE_DOCKER_STATE"]))
    status = _status(tmp_path)
    assert status["reason"] == "isolated_clickhouse_artifact_hash_mismatch"
    assert status["artifact_staging_verified"] is True
    assert status["clickhouse_artifact_readback_verified"] is False
    assert status["clickhouse_artifact_sha256"] == "f" * 64
    assert status["cleanup_verified"] is True


def test_docker_daemon_failure_cannot_claim_cleanup_verified(tmp_path: Path) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)
    assert _run("--check", env).returncode == 0
    env["FAKE_DOCKER_INFO_FAIL"] = "1"

    result = _run("--apply", env)

    assert result.returncode == 1
    status = _status(tmp_path)
    assert status["terminal"] == "failed"
    assert status["cleanup_verified"] is False
    assert status["reason"] == "ephemeral_resource_cleanup_failed"


def test_cleanup_label_inspect_hang_is_bounded_and_fails_closed(tmp_path: Path) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)
    env["CLICKHOUSE_RESTORE_COMMAND_TIMEOUT_SECONDS"] = "1"
    assert _run("--check", env).returncode == 0
    env["FAKE_CHECK_TABLE_FAIL"] = "1"
    env["FAKE_LABEL_INSPECT_HANG"] = "1"

    result = subprocess.run(
        ["bash", str(SCRIPT), "--apply"],
        text=True,
        capture_output=True,
        env=env,
        timeout=12,
    )

    assert result.returncode == 1
    status = _status(tmp_path)
    assert status["terminal"] == "failed"
    assert status["cleanup_verified"] is False
    assert status["reason"] == "ephemeral_resource_cleanup_failed"


def test_check_fails_closed_on_local_image_id_drift(tmp_path: Path) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)
    env["FAKE_CLICKHOUSE_IMAGE_ID"] = "sha256:" + "0" * 64

    result = _run("--check", env)

    assert result.returncode == 1
    assert "clickhouse_local_image_id_mismatch" in result.stderr
    docker_log = Path(env["FAKE_DOCKER_LOG"]).read_text(encoding="utf-8")
    assert "network create" not in docker_log
    assert "volume create" not in docker_log


def test_flat_host_deploy_accepts_reviewed_absolute_helper_and_config_paths(
    tmp_path: Path,
) -> None:
    artifact, manifest, inventory = _write_backup_fixture(tmp_path)
    restored = tmp_path / "restored.tsv"
    restored.write_text(_inventory(), encoding="utf-8")
    env = _environment(tmp_path, artifact, manifest, inventory, restored)
    host_root = tmp_path / "host-backup"
    host_root.mkdir()
    helper = host_root / "clickhouse-restore-inventory.py"
    backup_config = host_root / "backup_disk.xml"
    cluster_config = host_root / "isolated-cluster.xml"
    shutil.copyfile(INVENTORY_HELPER, helper)
    shutil.copyfile(
        REPO_ROOT / "ops/signoz/clickhouse/config.d/backup_disk.xml", backup_config
    )
    shutil.copyfile(
        REPO_ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml",
        cluster_config,
    )
    env.update(
        {
            "CLICKHOUSE_RESTORE_INVENTORY_HELPER": str(helper),
            "CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG": str(backup_config),
            "CLICKHOUSE_RESTORE_CLUSTER_CONFIG": str(cluster_config),
        }
    )

    result = _run("--check", env)

    assert result.returncode == 0, result.stderr
    assert "terminal=check_pass_no_runtime_write" in result.stdout


def test_inventory_compare_allows_online_row_delta_but_not_engine_drift(
    tmp_path: Path,
) -> None:
    source = tmp_path / "source.tsv"
    restored = tmp_path / "restored.tsv"
    source.write_text(_inventory(), encoding="utf-8")
    restored.write_text(_inventory(restored_delta=9), encoding="utf-8")

    accepted = subprocess.run(
        [
            "python3",
            str(INVENTORY_HELPER),
            "compare",
            "--source-inventory",
            str(source),
            "--restored-inventory",
            str(restored),
        ],
        text=True,
        capture_output=True,
    )
    assert accepted.returncode == 0, accepted.stderr
    assert "manifest_parity=1" in accepted.stdout
    assert "row_delta=9" in accepted.stdout

    restored.write_text(_inventory(engine_mismatch=True), encoding="utf-8")
    rejected = subprocess.run(
        [
            "python3",
            str(INVENTORY_HELPER),
            "compare",
            "--source-inventory",
            str(source),
            "--restored-inventory",
            str(restored),
        ],
        text=True,
        capture_output=True,
    )
    assert rejected.returncode == 1
    assert "table_engine_mismatch_signoz_meter_meter" in rejected.stderr

    restored.write_text(_inventory(schema_mismatch=True), encoding="utf-8")
    schema_rejected = subprocess.run(
        [
            "python3",
            str(INVENTORY_HELPER),
            "compare",
            "--source-inventory",
            str(source),
            "--restored-inventory",
            str(restored),
        ],
        text=True,
        capture_output=True,
    )
    assert schema_rejected.returncode == 1
    assert "table_schema_mismatch_signoz_meter_meter" in schema_rejected.stderr