awoooi/scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py

from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path


ROOT = Path(__file__).resolve().parents[3]
DEPLOYER = ROOT / "scripts" / "ops" / "deploy-signoz-metadata-toolchain.sh"


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


def fake_environment(tmp_path: Path, *, mode: str) -> dict[str, str]:
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    event_log = tmp_path / "events.log"
    ssh_count = tmp_path / "ssh.count"
    event_log.touch()
    ssh_count.write_text("0\n", encoding="utf-8")

    _write_executable(
        fake_bin / "timeout",
        """#!/bin/bash
set -eu
while [[ "${1:-}" == --kill-after=* ]]; do shift; done
shift
exec "$@"
""",
    )
    _write_executable(
        fake_bin / "ssh",
        """#!/bin/bash
set -eu
count="$(cat "${TEST_SSH_COUNT:?}")"
count=$((count + 1))
printf '%s\n' "$count" > "${TEST_SSH_COUNT:?}"
printf 'ssh %s\n' "$*" >> "${TEST_EVENT_LOG:?}"
cat >/dev/null || true
case "${FAKE_SSH_MODE:?}" in
  check_absent)
    printf 'REMOTE_CHECK terminal=check_pass_no_write state=absent drift=0 candidate_count=0 api=200 active_operations=0 container_state_sha256=%064d\n' 0
    ;;
  check_drift)
    exit 3
    ;;
  apply)
    case "$count" in
      1)
        printf 'REMOTE_CHECK terminal=check_pass_no_write state=absent drift=0 candidate_count=0 api=200 active_operations=0 container_state_sha256=%064d\n' 0
        ;;
      2) ;;
      3)
        printf '{"phase":"terminal","terminal":"pass_source_deployed_inactive"}\n'
        ;;
      4)
        printf 'REMOTE_CHECK terminal=check_pass_no_write state=exact drift=0 candidate_count=0 api=200 active_operations=0 container_state_sha256=%064d\n' 0
        ;;
      *) exit 90 ;;
    esac
    ;;
esac
""",
    )
    _write_executable(
        fake_bin / "scp",
        """#!/bin/bash
set -eu
printf 'scp %s\n' "$*" >> "${TEST_EVENT_LOG:?}"
""",
    )
    return {
        **os.environ,
        "PATH": f"{fake_bin}:{os.environ['PATH']}",
        "TEST_EVENT_LOG": str(event_log),
        "TEST_SSH_COUNT": str(ssh_count),
        "FAKE_SSH_MODE": mode,
        "TRACE_ID": "trace-P0-OBS-002-metadata-deploy",
        "RUN_ID": f"run-{mode}",
        "WORK_ITEM_ID": "P0-OBS-002",
    }


def run_deployer(
    tmp_path: Path, *, mode: str, apply: bool
) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["bash", str(DEPLOYER), "--apply" if apply else "--check"],
        env=fake_environment(tmp_path, mode=mode),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=20,
        check=False,
    )


def receipt_rows(stdout: str) -> list[dict[str, object]]:
    return [json.loads(line) for line in stdout.splitlines() if line.startswith("{")]


def test_help_documents_inactive_immutable_contract() -> None:
    result = subprocess.run(
        ["bash", str(DEPLOYER), "--help"],
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=10,
        check=False,
    )
    assert result.returncode == 0
    assert "immutable exact-hash directory" in result.stdout
    assert "does not create or" in result.stdout
    assert "active pointer" in result.stdout


def test_check_mode_is_no_write_and_ready_when_bundle_absent(tmp_path: Path) -> None:
    result = run_deployer(tmp_path, mode="check_absent", apply=False)
    assert result.returncode == 0, result.stderr
    rows = receipt_rows(result.stdout)
    assert rows[-1]["terminal"] == "check_pass_no_write"
    assert rows[-1]["detail"].endswith("remote_state_absent")
    events = (tmp_path / "events.log").read_text(encoding="utf-8").splitlines()
    assert len([line for line in events if line.startswith("ssh ")]) == 1
    assert not [line for line in events if line.startswith("scp ")]


def test_check_mode_fails_closed_on_remote_drift(tmp_path: Path) -> None:
    result = run_deployer(tmp_path, mode="check_drift", apply=False)
    assert result.returncode == 1
    rows = receipt_rows(result.stdout)
    assert rows[-1]["terminal"] == "failed_no_write"


def test_apply_uses_five_transfers_and_exact_postcheck(tmp_path: Path) -> None:
    result = run_deployer(tmp_path, mode="apply", apply=True)
    assert result.returncode == 0, result.stderr
    rows = receipt_rows(result.stdout)
    assert rows[-1]["terminal"] == "pass_source_deployed_inactive"
    assert rows[-1]["detail"] == "runtime_export_not_executed"
    events = (tmp_path / "events.log").read_text(encoding="utf-8").splitlines()
    assert len([line for line in events if line.startswith("scp ")]) == 5
    assert len([line for line in events if line.startswith("ssh ")]) == 4


def test_deployer_has_no_active_pointer_or_destructive_fallback() -> None:
    source = DEPLOYER.read_text(encoding="utf-8")
    assert '"${REMOTE_ROOT}/current"' in source
    assert "active_pointer_0" in source
    assert "pass_source_deployed_inactive" in source
    assert "quarantine-candidate" in source
    assert "quarantine-target" in source
    assert "docker stop" not in source
    assert "docker start" not in source
    assert "docker restart" not in source
    assert "sqlite3" not in source
    assert "github.com" not in source.casefold()
    assert "curl |" not in source
    assert "curl -fsSL" not in source
    assert "\nrm " not in source
    assert "ln -s" not in source


def test_shared_operation_lock_is_existing_read_only_and_fail_closed() -> None:
    source = DEPLOYER.read_text(encoding="utf-8")
    assert "OPERATION_LOCK=/tmp/awoooi-signoz-backup-operation.lock" in source
    assert '[ -f "${OPERATION_LOCK}" ] && [ ! -L "${OPERATION_LOCK}" ]' in source
    assert 'exec 8<"${OPERATION_LOCK}"' in source
    assert "flock -n 8" in source
    assert "exec 8>>/tmp/awoooi-signoz-backup-operation.lock" not in source


def test_exact_five_file_supply_chain_and_modes_are_fixed() -> None:
    source = DEPLOYER.read_text(encoding="utf-8")
    for path in (
        "config/signoz/metadata-export-policy.json",
        "scripts/backup/signoz_metadata_contract.py",
        "scripts/backup/signoz-metadata-export.py",
        "scripts/backup/verify-signoz-metadata-export.py",
        "scripts/backup/signoz-metadata-restore-drill.py",
    ):
        assert path in source
    assert "MODES=(0644 0644 0755 0755 0755)" in source
    assert '[ "${#EXPECTED_HASHES[@]}" -eq 5 ]' in source