147 lines
4.4 KiB
Python
147 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
RUNNER = "ops/run_gcp_infra_execute_canary.py"
|
|
|
|
|
|
def run_runner(args: list[str], *, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]:
|
|
merged_env = os.environ.copy()
|
|
if env:
|
|
merged_env.update(env)
|
|
return subprocess.run(
|
|
["python3", RUNNER, *args],
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
env=merged_env,
|
|
)
|
|
|
|
|
|
def write_capsule(path: Path) -> None:
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"artifact": "sqlite_postgres_restore_canary_capsule",
|
|
"status": "pass",
|
|
"source_total_rows": 805745,
|
|
"target_total_rows": 805745,
|
|
"secret_marker": "capsule-secret-marker",
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
def load_json(path: Path) -> dict:
|
|
return json.loads(path.read_text())
|
|
|
|
|
|
def test_dry_run_writes_full_plan_and_redacts_capsule(tmp_path: Path) -> None:
|
|
capsule = tmp_path / "capsule.json"
|
|
output = tmp_path / "execute-canary.json"
|
|
write_capsule(capsule)
|
|
|
|
completed = run_runner(
|
|
[
|
|
"--restore-canary-capsule",
|
|
str(capsule),
|
|
"--output",
|
|
str(output),
|
|
],
|
|
env={"TELEO_CLOUDSQL_POSTGRES_PASSWORD": "do-not-record-this"},
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
payload = load_json(output)
|
|
rendered = output.read_text()
|
|
assert payload["mode"] == "dry_run"
|
|
assert payload["status"] == "planned"
|
|
assert [operation["name"] for operation in payload["operations"]] == [
|
|
"gcloud_token_preflight",
|
|
"apply_gcp_iam_split",
|
|
"apply_gcp_runtime_baseline",
|
|
"trigger_gcp_readiness",
|
|
]
|
|
assert all(operation["exitcode"] is None for operation in payload["operations"])
|
|
assert "restore_canary_capsule_b64=<restore_canary_capsule_b64>" in rendered
|
|
assert "capsule-secret-marker" not in rendered
|
|
assert "do-not-record-this" not in rendered
|
|
|
|
|
|
def test_validation_failure_also_redacts_capsule(tmp_path: Path) -> None:
|
|
capsule = tmp_path / "capsule.json"
|
|
output = tmp_path / "failed-validation.json"
|
|
write_capsule(capsule)
|
|
|
|
completed = run_runner(
|
|
[
|
|
"--execute",
|
|
"--restore-canary-capsule",
|
|
str(capsule),
|
|
"--output",
|
|
str(output),
|
|
],
|
|
)
|
|
|
|
assert completed.returncode == 1
|
|
payload = load_json(output)
|
|
rendered = output.read_text()
|
|
assert payload["status"] == "failed_validation"
|
|
assert "--admin-ssh-cidr is required in execute mode" in payload["problems"]
|
|
assert "restore_canary_capsule_b64=<restore_canary_capsule_b64>" in rendered
|
|
assert "capsule-secret-marker" not in rendered
|
|
|
|
|
|
def test_execute_restore_requires_execute_and_sqlite_backup(tmp_path: Path) -> None:
|
|
output = tmp_path / "failed-validation.json"
|
|
|
|
completed = run_runner(["--execute-restore", "--output", str(output)])
|
|
|
|
assert completed.returncode == 1
|
|
payload = load_json(output)
|
|
assert payload["status"] == "failed_validation"
|
|
assert "--execute-restore requires --execute" in payload["problems"]
|
|
assert "--execute-restore requires --sqlite-backup" in payload["problems"]
|
|
|
|
|
|
def test_execute_requires_single_host_admin_cidr(tmp_path: Path) -> None:
|
|
output = tmp_path / "failed-validation.json"
|
|
|
|
completed = run_runner(
|
|
[
|
|
"--execute",
|
|
"--admin-ssh-cidr",
|
|
"10.0.0.0/24",
|
|
"--output",
|
|
str(output),
|
|
],
|
|
)
|
|
|
|
assert completed.returncode == 1
|
|
payload = load_json(output)
|
|
assert payload["status"] == "failed_validation"
|
|
assert "--admin-ssh-cidr must be a single trusted IPv4 /32" in payload["problems"]
|
|
|
|
|
|
def test_invalid_capsule_path_fails_validation(tmp_path: Path) -> None:
|
|
output = tmp_path / "failed-validation.json"
|
|
missing = tmp_path / "missing-capsule.json"
|
|
|
|
completed = run_runner(
|
|
[
|
|
"--restore-canary-capsule",
|
|
str(missing),
|
|
"--output",
|
|
str(output),
|
|
],
|
|
)
|
|
|
|
assert completed.returncode == 1
|
|
payload = load_json(output)
|
|
assert payload["status"] == "failed_validation"
|
|
assert f"restore canary capsule not found: {missing}" in payload["problems"]
|