from __future__ import annotations import importlib.util import json import stat import subprocess import sys from types import SimpleNamespace from pathlib import Path from typing import Any import pytest ROOT = Path(__file__).resolve().parents[3] CONTROLLER = ROOT / "scripts" / "backup" / "signoz-metadata-isolated-restore.py" POLICY = ROOT / "config" / "signoz" / "metadata-export-policy.json" TRACE_ID = "trace-P0-OBS-002-isolated" RUN_ID = "run-P0-OBS-002-isolated" WORK_ITEM_ID = "P0-OBS-002" def load_controller() -> Any: backup_dir = str(ROOT / "scripts" / "backup") if backup_dir not in sys.path: sys.path.insert(0, backup_dir) spec = importlib.util.spec_from_file_location("signoz_isolated_restore", CONTROLLER) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module @pytest.fixture() def controller() -> Any: return load_controller() class RecordingRunner: def __init__(self, controller: Any) -> None: self.controller = controller self.commands: list[list[str]] = [] self.resources: dict[tuple[str, str], dict[str, Any]] = {} self.production = { name: { "Id": (name.encode().hex() + "0" * 64)[:64], "Image": image_id, "Config": {"Image": ref}, "State": { "Running": True, "StartedAt": "2026-07-22T00:00:00Z", }, "RestartCount": 0, } for name, (ref, image_id) in controller.EXPECTED_PRODUCTION.items() } self.image_rows = ["sha256:test|test:fixture"] def __call__(self, argv: list[str], _timeout: int) -> Any: self.commands.append(argv[:]) c = self.controller if argv[1:3] == ["image", "inspect"]: ref = argv[-1] for expected_ref, image_id in c.EXPECTED_IMAGES.values(): if ref == expected_ref: return c.CommandResult(0, f"{image_id}\n") return c.CommandResult(1, "") if argv[1:3] == ["image", "ls"]: return c.CommandResult(0, "\n".join(self.image_rows) + "\n") if argv[1:2] == ["info"]: return c.CommandResult(0, "28.0.0\n") if len(argv) >= 4 and argv[2] == "inspect": kind, name = argv[1], argv[3] if kind == "container" and name in self.production: return c.CommandResult(0, json.dumps([self.production[name]])) value = self.resources.get((kind, name)) if value is None: return c.CommandResult(1, "") return c.CommandResult(0, json.dumps([value])) if argv[1:3] == ["network", "create"]: name = argv[-1] labels = labels_from_command(argv) self.resources[("network", name)] = {"Name": name, "Labels": labels} return c.CommandResult(0, f"{name}\n") if argv[1:3] == ["volume", "create"]: name = argv[-1] labels = labels_from_command(argv) self.resources[("volume", name)] = {"Name": name, "Labels": labels} return c.CommandResult(0, f"{name}\n") if argv[1:2] == ["run"]: name = argv[argv.index("--name") + 1] ref = argv[-1] expected_id = next( image_id for expected_ref, image_id in c.EXPECTED_IMAGES.values() if expected_ref == ref ) self.resources[("container", name)] = { "Id": (name.encode().hex() + "0" * 64)[:64], "Image": expected_id, "Config": {"Image": ref, "Labels": labels_from_command(argv)}, "HostConfig": {"RestartPolicy": {"Name": "no"}}, "State": {"Running": True}, "NetworkSettings": { "Ports": { "8080/tcp": [ {"HostIp": "127.0.0.1", "HostPort": "49173"} ] } if name.endswith("-server") else {} }, "RestartCount": 0, } return c.CommandResult(0, f"{name}\n") if argv[1:2] == ["port"]: return c.CommandResult(0, "127.0.0.1:49173\n") if argv[1:2] == ["exec"] and argv[-2:] == ["--query", "SELECT 1"]: return c.CommandResult(0, "1\n") if argv[1:3] == ["rm", "--force"]: self.resources.pop(("container", argv[-1]), None) return c.CommandResult(0, "") if argv[1:3] == ["volume", "rm"]: self.resources.pop(("volume", argv[-1]), None) return c.CommandResult(0, "") if argv[1:3] == ["network", "rm"]: self.resources.pop(("network", argv[-1]), None) return c.CommandResult(0, "") if argv[1:2] == ["ps"] or ( len(argv) >= 3 and argv[1] in {"volume", "network"} and argv[2] == "ls" ): label = argv[-1].removeprefix("label=") key, expected = label.split("=", 1) matches = [] for (kind, name), value in self.resources.items(): if argv[1] == "ps" and kind != "container": continue if argv[1] in {"volume", "network"} and kind != argv[1]: continue labels = ( value.get("Config", {}).get("Labels", {}) if kind == "container" else value.get("Labels", {}) ) if labels.get(key) == expected: matches.append(name) return c.CommandResult(0, "\n".join(matches)) return c.CommandResult(1, "") def labels_from_command(argv: list[str]) -> dict[str, str]: labels: dict[str, str] = {} for index, value in enumerate(argv): if value == "--label": key, label_value = argv[index + 1].split("=", 1) labels[key] = label_value return labels def test_identity_and_exact_resource_names(controller: Any) -> None: plan = controller.build_resource_plan(TRACE_ID, RUN_ID, WORK_ITEM_ID) assert plan.identity == controller.derive_identity( TRACE_ID, RUN_ID, WORK_ITEM_ID ) assert len(plan.identity) == 64 assert plan.prefix == f"awoooi-signoz-md-restore-{plan.identity[:16]}" assert plan.canonical_id == ( f"ephemeral-signoz-metadata-restore-{plan.identity[:16]}" ) assert plan.containers == ( f"{plan.prefix}-server", f"{plan.prefix}-clickhouse", f"{plan.prefix}-zookeeper", ) assert len(plan.volumes) == 4 assert plan.workspace == Path(f"/tmp/{plan.prefix}-private") def test_policy_requires_reviewed_exact_startup_and_images(controller: Any) -> None: policy = json.loads(POLICY.read_text(encoding="utf-8")) assert controller.validate_restore_policy(policy)["startup_contract"][ "reviewed" ] is True policy["restore_isolation"]["startup_contract"]["reviewed"] = False with pytest.raises( controller.ContractError, match="restore_startup_contract_not_reviewed" ): controller.validate_restore_policy(policy) policy = json.loads(POLICY.read_text(encoding="utf-8")) policy["restore_isolation"]["images"]["clickhouse"]["id"] = ( f"sha256:{'a' * 64}" ) with pytest.raises( controller.ContractError, match="restore_image_policy_not_exact" ): controller.validate_restore_policy(policy) def test_resource_creation_is_internal_no_pull_no_restart_and_loopback( controller: Any, ) -> None: runner = RecordingRunner(controller) plan = controller.build_resource_plan(TRACE_ID, RUN_ID, WORK_ITEM_ID) cluster = ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml" controller.create_resources(runner, plan, RUN_ID, cluster) commands = runner.commands network = next(command for command in commands if command[1:3] == ["network", "create"]) assert "--internal" in network run_commands = [command for command in commands if command[1:2] == ["run"]] assert len(run_commands) == 3 for command in run_commands: assert "--pull=never" in command assert "--restart=no" in command assert f"{controller.PRIMARY_LABEL}={RUN_ID}" in command assert f"{controller.IDENTITY_LABEL}={plan.identity}" in command assert not any("signoz-clickhouse:" in value for value in command) server = next(command for command in run_commands if plan.server_container in command) assert "127.0.0.1::8080" in server assert f"{plan.server_data_volume}:/var/lib/signoz" in server assert "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN=tcp://clickhouse:9000" in server clickhouse_run_index = commands.index( next(command for command in run_commands if plan.clickhouse_container in command) ) readiness_index = next( index for index, command in enumerate(commands) if command[1:2] == ["exec"] and command[-2:] == ["--query", "SELECT 1"] ) server_run_index = commands.index(server) assert clickhouse_run_index < readiness_index < server_run_index def test_exact_image_drift_fails_before_resource_creation(controller: Any) -> None: runner = RecordingRunner(controller) def drift(argv: list[str], timeout: int) -> Any: result = runner(argv, timeout) if argv[-1] == controller.EXPECTED_IMAGES["clickhouse"][0]: return controller.CommandResult(0, f"sha256:{'b' * 64}\n") return result with pytest.raises( controller.ContractError, match="restore_image_id_mismatch_clickhouse" ): controller.verify_exact_images(drift) assert not runner.resources def test_cleanup_refuses_same_name_with_wrong_ownership(controller: Any) -> None: runner = RecordingRunner(controller) plan = controller.build_resource_plan(TRACE_ID, RUN_ID, WORK_ITEM_ID) runner.resources[("container", plan.server_container)] = { "Config": { "Labels": { controller.PRIMARY_LABEL: "some-other-run", controller.IDENTITY_LABEL: plan.identity, } } } with pytest.raises( controller.ContractError, match="cleanup_container_ownership_mismatch" ): controller.remove_owned_resource( runner, plan, RUN_ID, "container", plan.server_container, ) assert ("container", plan.server_container) in runner.resources assert not any(command[1:3] == ["rm", "--force"] for command in runner.commands) def test_cleanup_removes_exact_resources_and_proves_continuity( controller: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: runner = RecordingRunner(controller) plan = controller.build_resource_plan( TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path ) cluster = ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml" controller.create_workspace(plan) controller.create_resources(runner, plan, RUN_ID, cluster) state = { "production_before": controller.production_snapshot(runner), "image_inventory_before": controller.image_inventory(runner), "listener": {"host": "127.0.0.1", "port": 49173}, } monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False) monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None) errors = controller.cleanup_and_verify(runner, plan, RUN_ID, state) assert errors == [] assert not runner.resources assert not plan.workspace.exists() @pytest.mark.parametrize("server_created", [False, True]) def test_cleanup_reconciles_without_listener_state_after_interruption( controller: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, server_created: bool, ) -> None: runner = RecordingRunner(controller) plan = controller.build_resource_plan( TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path ) controller.create_workspace(plan) if server_created: cluster = ( ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml" ) controller.create_resources(runner, plan, RUN_ID, cluster) state = { "production_before": controller.production_snapshot(runner), "image_inventory_before": controller.image_inventory(runner), "listener": None, } monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False) monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None) errors = controller.cleanup_and_verify(runner, plan, RUN_ID, state) assert errors == [] assert not runner.resources assert not plan.workspace.exists() def test_cleanup_already_clean_without_state_uses_fresh_continuity( controller: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: runner = RecordingRunner(controller) plan = controller.build_resource_plan( TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path ) monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False) monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None) errors = controller.cleanup_and_verify(runner, plan, RUN_ID, None) assert errors == [] assert not plan.workspace.exists() def test_cleanup_reconciles_resource_materialized_after_first_inventory( controller: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: runner = RecordingRunner(controller) plan = controller.build_resource_plan( TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path ) sleeps = 0 def materialize_after_first_pass(_seconds: float) -> None: nonlocal sleeps sleeps += 1 if sleeps == 1: runner.resources[("volume", plan.server_data_volume)] = { "Name": plan.server_data_volume, "Labels": { controller.PRIMARY_LABEL: RUN_ID, controller.IDENTITY_LABEL: plan.identity, }, } monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False) monkeypatch.setattr(controller.time, "sleep", materialize_after_first_pass) errors = controller.cleanup_and_verify(runner, plan, RUN_ID, None) assert errors == [] assert not runner.resources removals = [ command for command in runner.commands if command[1:3] == ["volume", "rm"] and command[-1] == plan.server_data_volume ] assert len(removals) == 1 def test_apply_cli_requires_durable_receipt_before_runtime_dispatch() -> None: result = subprocess.run( [ sys.executable, str(CONTROLLER), "--apply", "--trace-id", TRACE_ID, "--run-id", RUN_ID, "--work-item-id", WORK_ITEM_ID, ], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False, ) assert result.returncode == 2 assert "receipt_file_required_for_restore_apply" in result.stderr assert result.stdout == "" def test_distinct_verifier_requires_immutable_apply_receipt( controller: Any, tmp_path: Path, ) -> None: args = SimpleNamespace( trace_id=TRACE_ID, run_id=RUN_ID, work_item_id=WORK_ITEM_ID, ) plan = controller.build_resource_plan( TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path ) executor_receipt = controller.executor_success( args, plan, { "restoreSemanticDigest": "a" * 64, "restoredAssetCounts": { "dashboards": 1, "alert_rules": 2, "explorer_views": 3, }, }, ) path = tmp_path / "apply.json" controller.write_terminal(path, executor_receipt) validated = controller.validate_apply_receipt(path, args, plan) verified = controller.independent_success(args, plan, validated) assert executor_receipt["completionClaim"] is False assert executor_receipt["executorZeroResidueVerified"] is True assert verified["phase"] == "independent_restore_verifier" assert verified["completionClaim"] is True assert verified["independentZeroResidueVerified"] is True assert verified["restoreSemanticDigest"] == "a" * 64 assert verified["applyReceiptValidated"] is True assert len(verified["applyReceiptSha256"]) == 64 def test_interrupted_verify_cleanup_reconciles_again_before_zero_residue_receipt( controller: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: plan = controller.build_resource_plan( TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path ) calls = 0 emitted: list[dict[str, Any]] = [] def interrupt_once( _runner: Any, _plan: Any, _run_id: str, _state: dict[str, Any] | None, ) -> list[str]: nonlocal calls calls += 1 if calls == 1: raise controller.SignalAbort("signal_15_cleanup_required") return [] monkeypatch.setattr( sys, "argv", [ str(CONTROLLER), "--verify-cleanup", "--trace-id", TRACE_ID, "--run-id", RUN_ID, "--work-item-id", WORK_ITEM_ID, ], ) monkeypatch.setattr(controller, "build_resource_plan", lambda *_args: plan) monkeypatch.setattr(controller, "load_policy", lambda _path: {}) monkeypatch.setattr(controller, "load_state_if_safe", lambda _plan: None) monkeypatch.setattr(controller, "cleanup_and_verify", interrupt_once) monkeypatch.setattr(controller, "emit", emitted.append) exit_code = controller.main() assert exit_code == 1 assert calls == 2 assert emitted[-1]["terminal"] == "failed_zero_residue_verified_no_completion" assert emitted[-1]["zeroResidueVerified"] is True assert emitted[-1]["cleanupErrorCodes"] == [] assert "signal_15_cleanup_required" in emitted[-1]["detail"] def test_bootstrap_secrets_stay_in_mode_0400_files( controller: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: calls: list[tuple[str, dict[str, Any] | None]] = [] def fake_request( url: str, *, method: str = "GET", payload: dict[str, Any] | None = None, authorization: str | None = None, timeout: int = 10, ) -> dict[str, Any]: assert authorization is None calls.append((url, payload)) if url.endswith("/api/v1/register"): return {"data": {"orgId": "isolated-org"}, "status": "success"} return { "data": { "accessToken": "a" * 64, "refreshToken": "r" * 64, }, "status": "success", } monkeypatch.setattr(controller, "request_json", fake_request) token = controller.bootstrap_isolated_session("http://127.0.0.1:49173", tmp_path) assert token == tmp_path / "session-access-token" assert stat.S_IMODE(token.stat().st_mode) == 0o400 assert sorted(path.name for path in tmp_path.iterdir()) == [ "session-access-token" ] assert calls[0][0].endswith("/api/v1/register") assert calls[1][0].endswith("/api/v2/sessions/email_password") assert capsys.readouterr().out == "" def test_bootstrap_accepts_live_root_response_envelope( controller: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: def fake_request( url: str, **_kwargs: Any, ) -> dict[str, Any]: if url.endswith("/api/v1/register"): return {"orgId": "isolated-org"} return {"accessToken": "a" * 64, "refreshToken": "r" * 64} monkeypatch.setattr(controller, "request_json", fake_request) token = controller.bootstrap_isolated_session( "http://127.0.0.1:49173", tmp_path ) assert token.read_text(encoding="ascii").strip() == "a" * 64 assert stat.S_IMODE(token.stat().st_mode) == 0o400 def test_controller_source_has_signal_cleanup_and_no_secret_argv() -> None: source = CONTROLLER.read_text(encoding="utf-8") assert "signal.SIGHUP" in source assert "signal.SIGINT" in source assert "signal.SIGTERM" in source assert "finally:" in source assert "cleanup_and_verify(" in source assert '"--pull=never"' in source assert '"--restart=no"' in source assert '"--auth-mode"' in source assert "access_token" not in source[source.index("argv = [") : source.index("output = run_required", source.index("argv = ["))] assert "password" not in source[source.index("argv = [") : source.index("output = run_required", source.index("argv = ["))] durable_write = source.rindex("write_terminal(Path(args.receipt_file), result)") final_handler_restore = source.rindex("restore_signal_handlers()") assert durable_write < final_handler_restore assert "if not args.apply and not signals_held_until_terminal:" in source