229 lines
7.9 KiB
Python
229 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import ops.check_gcp_infra_readiness as checker
|
|
|
|
|
|
def write_restore_proof(path: Path, *, generated_at: str, rows: int, status: str = "pass") -> None:
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"generated_at_utc": generated_at,
|
|
"mismatched_tables": [],
|
|
"postgres_image": "postgres:16-alpine",
|
|
"source_integrity_check": "ok",
|
|
"source_table_count": 2,
|
|
"source_total_rows": rows,
|
|
"status": status,
|
|
"target_table_count": 2,
|
|
"target_total_rows": rows,
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
def test_backup_bucket_check_uses_gcloud_storage_json(monkeypatch) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_run_json(args: list[str]) -> dict:
|
|
calls.append(args)
|
|
return {
|
|
"versioning": {"enabled": True},
|
|
"iamConfiguration": {
|
|
"uniformBucketLevelAccess": {"enabled": True},
|
|
"publicAccessPrevention": "enforced",
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(checker, "BACKUP_BUCKETS", ("teleo-test-backups",))
|
|
monkeypatch.setattr(checker, "run_json", fake_run_json)
|
|
|
|
result = checker.check_backup_buckets()
|
|
|
|
assert result.status == "pass"
|
|
assert calls == [
|
|
[
|
|
"gcloud",
|
|
"storage",
|
|
"buckets",
|
|
"describe",
|
|
"gs://teleo-test-backups",
|
|
"--format=json",
|
|
]
|
|
]
|
|
assert "versioning=True" in result.detail
|
|
assert "ubla=True" in result.detail
|
|
assert "publicAccessPrevention=True" in result.detail
|
|
|
|
|
|
def test_backup_bucket_check_rejects_weak_bucket_settings(monkeypatch) -> None:
|
|
def fake_run_json(args: list[str]) -> dict:
|
|
assert args[:4] == ["gcloud", "storage", "buckets", "describe"]
|
|
return {
|
|
"versioning": {"enabled": False},
|
|
"iamConfiguration": {
|
|
"uniformBucketLevelAccess": {"enabled": True},
|
|
"publicAccessPrevention": "inherited",
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(checker, "BACKUP_BUCKETS", ("teleo-test-backups",))
|
|
monkeypatch.setattr(checker, "run_json", fake_run_json)
|
|
|
|
result = checker.check_backup_buckets()
|
|
|
|
assert result.status == "fail"
|
|
assert "versioning=False" in result.detail
|
|
assert "ubla=True" in result.detail
|
|
assert "publicAccessPrevention=False" in result.detail
|
|
|
|
|
|
def test_readiness_checker_does_not_use_gsutil() -> None:
|
|
source = Path("ops/check_gcp_infra_readiness.py").read_text()
|
|
|
|
assert "gsutil" not in source
|
|
assert 'payload["fail_count"] or payload["blocked_count"]' in source
|
|
|
|
|
|
def test_run_json_rejects_permission_warning_even_when_gcloud_exits_zero(monkeypatch) -> None:
|
|
completed = subprocess.CompletedProcess(
|
|
["gcloud", "compute", "instances", "list"],
|
|
0,
|
|
stdout="[]\n",
|
|
stderr="WARNING: Required 'compute.instances.list' permission for 'projects/teleo-501523'\n",
|
|
)
|
|
monkeypatch.setattr(checker.subprocess, "run", lambda *args, **kwargs: completed)
|
|
|
|
try:
|
|
checker.run_json(completed.args)
|
|
except checker.GcloudAccessError as exc:
|
|
assert "compute.instances.list" in str(exc)
|
|
else: # pragma: no cover - assertion branch
|
|
raise AssertionError("permission warning was treated as an empty live inventory")
|
|
|
|
|
|
def test_safe_check_classifies_access_error_as_blocked() -> None:
|
|
def denied() -> checker.Check:
|
|
raise checker.GcloudAccessError("PERMISSION_DENIED")
|
|
|
|
result = checker.safe_check("compute_runtime_service_accounts", denied)
|
|
|
|
assert result.status == "blocked"
|
|
assert result.current_tier == "T0_no_live_readback"
|
|
assert "PERMISSION_DENIED" in result.detail
|
|
|
|
|
|
def test_restore_status_does_not_assert_unread_target_exists() -> None:
|
|
result = checker.check_kb_restore_or_replication()
|
|
|
|
assert result.status == "blocked"
|
|
assert "target existence must be established separately" in result.detail
|
|
assert "standby target exists" not in result.detail
|
|
|
|
|
|
def test_canonical_postgres_restore_contract_is_complete() -> None:
|
|
result = checker.check_canonical_postgres_restore_contract()
|
|
|
|
assert result.status == "pass"
|
|
assert result.current_tier == "T2_canonical_restore_contract"
|
|
|
|
|
|
def test_local_canonical_postgres_restore_preflight_is_retained() -> None:
|
|
result = checker.check_local_canonical_postgres_restore_preflight()
|
|
|
|
assert result.status == "pass"
|
|
assert "exported_snapshot_tables=39" in result.detail
|
|
assert "rows=52164" in result.detail
|
|
|
|
|
|
def test_restore_canary_check_uses_newest_generated_at_not_filename(tmp_path, monkeypatch) -> None:
|
|
write_restore_proof(
|
|
tmp_path / "sqlite-postgres-restore-canary-zzzz.json",
|
|
generated_at="2026-07-07T10:00:00+00:00",
|
|
rows=10,
|
|
)
|
|
write_restore_proof(
|
|
tmp_path / "sqlite-postgres-restore-canary-aaaa.json",
|
|
generated_at="2026-07-07T12:00:00+00:00",
|
|
rows=20,
|
|
)
|
|
monkeypatch.setattr(checker, "PROOF_ROOT", tmp_path)
|
|
|
|
result = checker.check_local_sqlite_postgres_restore_canary()
|
|
|
|
assert result.status == "pass"
|
|
assert "sqlite-postgres-restore-canary-aaaa.json" in result.detail
|
|
assert "rows=20" in result.detail
|
|
|
|
|
|
def test_restore_canary_check_ignores_invalid_json_when_valid_proof_exists(tmp_path, monkeypatch) -> None:
|
|
(tmp_path / "sqlite-postgres-restore-canary-newer-name.json").write_text("{")
|
|
write_restore_proof(
|
|
tmp_path / "sqlite-postgres-restore-canary-valid.json",
|
|
generated_at="2026-07-07T12:00:00+00:00",
|
|
rows=20,
|
|
)
|
|
monkeypatch.setattr(checker, "PROOF_ROOT", tmp_path)
|
|
|
|
result = checker.check_local_sqlite_postgres_restore_canary()
|
|
|
|
assert result.status == "pass"
|
|
assert "sqlite-postgres-restore-canary-valid.json" in result.detail
|
|
|
|
|
|
def test_restore_canary_check_accepts_redacted_capsule(tmp_path, monkeypatch) -> None:
|
|
capsule = tmp_path / "restore-canary-capsule.json"
|
|
capsule.write_text(
|
|
json.dumps(
|
|
{
|
|
"artifact": "sqlite_postgres_restore_canary_capsule",
|
|
"source_proof_generated_at_utc": "2026-07-07T12:00:00+00:00",
|
|
"status": "pass",
|
|
"source_integrity_check": "ok",
|
|
"source_table_count": 41,
|
|
"target_table_count": 41,
|
|
"source_total_rows": 805745,
|
|
"target_total_rows": 805745,
|
|
"mismatched_table_count": 0,
|
|
"postgres_image": "postgres:16-alpine",
|
|
}
|
|
)
|
|
)
|
|
monkeypatch.setattr(checker, "PROOF_ROOT", tmp_path / "empty")
|
|
monkeypatch.setattr(checker, "RESTORE_CANARY_CAPSULE", str(capsule))
|
|
|
|
result = checker.check_local_sqlite_postgres_restore_canary()
|
|
|
|
assert result.status == "pass"
|
|
assert "evidence=capsule" in result.detail
|
|
assert "rows=805745" in result.detail
|
|
|
|
|
|
def test_restore_canary_check_labels_capsule_files_under_proof_root(tmp_path, monkeypatch) -> None:
|
|
capsule = tmp_path / "sqlite-postgres-restore-canary-capsule-20260707.json"
|
|
capsule.write_text(
|
|
json.dumps(
|
|
{
|
|
"artifact": "sqlite_postgres_restore_canary_capsule",
|
|
"source_proof_generated_at_utc": "2026-07-07T12:00:00+00:00",
|
|
"status": "pass",
|
|
"source_integrity_check": "ok",
|
|
"source_table_count": 41,
|
|
"target_table_count": 41,
|
|
"source_total_rows": 805745,
|
|
"target_total_rows": 805745,
|
|
"mismatched_table_count": 0,
|
|
"postgres_image": "postgres:16-alpine",
|
|
}
|
|
)
|
|
)
|
|
monkeypatch.setattr(checker, "PROOF_ROOT", tmp_path)
|
|
monkeypatch.setattr(checker, "RESTORE_CANARY_CAPSULE", None)
|
|
|
|
result = checker.check_local_sqlite_postgres_restore_canary()
|
|
|
|
assert result.status == "pass"
|
|
assert "evidence=capsule" in result.detail
|