119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
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:
|
|
assert "gsutil" not in Path("ops/check_gcp_infra_readiness.py").read_text()
|
|
|
|
|
|
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
|