69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
from pathlib import Path
|
|
|
|
from ops.capture_vps_canonical_postgres_snapshot import (
|
|
SAFE_SSH_TARGET_RE,
|
|
cleanup_failed_output,
|
|
remote_capture_script,
|
|
toc_counts,
|
|
)
|
|
|
|
|
|
def test_remote_capture_uses_one_exported_read_only_snapshot() -> None:
|
|
script = remote_capture_script(
|
|
run_id="canonical-20260711",
|
|
container="teleo-pg",
|
|
database="teleo",
|
|
service="leoclean-gateway.service",
|
|
)
|
|
|
|
assert "begin transaction isolation level repeatable read read only" in script
|
|
assert "select pg_export_snapshot()" in script
|
|
assert "[0-9A-F]+-[0-9A-F]+-[0-9]+" in script
|
|
assert '--snapshot="$snapshot_id"' in script
|
|
assert '-v "snapshot_id=$snapshot_id"' in script
|
|
assert "--no-owner --no-acl" in script
|
|
assert "cmp -s" in script
|
|
assert "systemctl restart" not in script
|
|
assert "docker restart" not in script
|
|
|
|
|
|
def test_manifest_hash_order_is_collation_independent() -> None:
|
|
manifest = Path("ops/postgres_parity_manifest.sql").read_text()
|
|
|
|
assert 'collate "C"' in manifest
|
|
assert "'server_address', host(inet_server_addr())" in manifest
|
|
|
|
|
|
def test_toc_counts_database_objects(tmp_path: Path) -> None:
|
|
toc = tmp_path / "dump.toc"
|
|
toc.write_text(
|
|
"; header\n"
|
|
"1; 0 0 TABLE DATA public claims postgres\n"
|
|
"2; 0 0 CONSTRAINT public claims claims_pkey postgres\n"
|
|
"3; 0 0 FK CONSTRAINT public evidence evidence_claim_fkey postgres\n"
|
|
"4; 0 0 INDEX public claims claims_text_idx postgres\n"
|
|
)
|
|
|
|
assert toc_counts(toc) == {"entries": 4, "table_data": 1, "constraints": 2, "indexes": 1}
|
|
|
|
|
|
def test_failure_cleanup_never_removes_preexisting_output(tmp_path: Path) -> None:
|
|
existing = tmp_path / "existing"
|
|
existing.mkdir()
|
|
marker = existing / "user-file"
|
|
marker.write_text("keep")
|
|
cleanup_failed_output(existing, preexisting=True)
|
|
|
|
created_by_run = tmp_path / "created-by-run"
|
|
created_by_run.mkdir()
|
|
cleanup_failed_output(created_by_run, preexisting=False)
|
|
|
|
assert marker.read_text() == "keep"
|
|
assert not created_by_run.exists()
|
|
|
|
|
|
def test_ssh_target_rejects_open_ssh_option_injection() -> None:
|
|
assert SAFE_SSH_TARGET_RE.fullmatch("root@77.42.65.182")
|
|
assert SAFE_SSH_TARGET_RE.fullmatch("teleo-staging-1.internal")
|
|
assert not SAFE_SSH_TARGET_RE.fullmatch("-oProxyCommand=malicious")
|
|
assert not SAFE_SSH_TARGET_RE.fullmatch("root@host command")
|