609 lines
21 KiB
Python
609 lines
21 KiB
Python
import argparse
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from ops import detect_vps_gcp_source_drift as detector
|
|
from ops import private_receipt_io
|
|
|
|
|
|
def endpoint_target(label: str) -> dict:
|
|
if label == "vps":
|
|
return {
|
|
"route_kind": "vps_docker",
|
|
"transport": "root@192.0.2.10",
|
|
"runtime_service": "leoclean-gateway.service",
|
|
"database_endpoint": "docker:teleo-pg",
|
|
"database_address": None,
|
|
"database_port": None,
|
|
"database": "teleo",
|
|
"database_user": "postgres",
|
|
"identity_source": "explicit_cli_target",
|
|
}
|
|
return {
|
|
"route_kind": "gcp_private_cloudsql",
|
|
"transport": "teleo-gcp-staging",
|
|
"runtime_service": "leoclean-gcp-prod-parallel.service",
|
|
"database_endpoint": "10.61.0.3:5432",
|
|
"database_address": "10.61.0.3",
|
|
"database_port": 5432,
|
|
"database": "teleo",
|
|
"database_user": "postgres",
|
|
"project": "teleo-501523",
|
|
"credential_source": "secret-manager:gcp-teleo-pgvector-standby-postgres-password",
|
|
"identity_source": "explicit_cli_target",
|
|
}
|
|
|
|
|
|
def manifest(*, label: str, row_count: int = 3, rowset_md5: str = "1" * 32) -> dict:
|
|
singleton = {kind: {"kind": kind, "items": []} for kind in detector.SINGLETON_KINDS if kind != "identity"}
|
|
is_gcp = label == "gcp"
|
|
singleton["identity"] = {
|
|
"kind": "identity",
|
|
"database": "teleo",
|
|
"current_user": "postgres",
|
|
"transaction_read_only": "on",
|
|
"server_version_num": 160000,
|
|
"server_address": "10.61.0.3" if is_gcp else None,
|
|
"server_port": 5432 if is_gcp else None,
|
|
"ssl": is_gcp,
|
|
"ssl_version": "TLSv1.3" if is_gcp else None,
|
|
}
|
|
return {
|
|
"singleton": singleton,
|
|
"tables": {
|
|
"public.claims": {
|
|
"kind": "table",
|
|
"schema": "public",
|
|
"table": "claims",
|
|
"row_count": row_count,
|
|
"rowset_md5": rowset_md5,
|
|
}
|
|
},
|
|
"performance": {},
|
|
}
|
|
|
|
|
|
def route_observation(label: str, *, runtime_commit: str = "a" * 40) -> dict:
|
|
target = endpoint_target(label)
|
|
process_cwd = f"/srv/{label}"
|
|
database_endpoint_identity = None
|
|
if label == "vps":
|
|
database_endpoint_identity = {
|
|
"kind": "docker_container",
|
|
"container_id": "c" * 64,
|
|
"container_name": "teleo-pg",
|
|
"running": True,
|
|
}
|
|
return {
|
|
"schema": detector.REMOTE_ROUTE_SCHEMA,
|
|
"route_kind": target["route_kind"],
|
|
"transport": target["transport"],
|
|
"hostname": f"{label}-host",
|
|
"service": target["runtime_service"],
|
|
"service_state": {
|
|
"ActiveState": "active",
|
|
"SubState": "running",
|
|
"MainPID": "123",
|
|
},
|
|
"process_cwd": process_cwd,
|
|
"expected_database": target["database"],
|
|
"database_endpoint": target["database_endpoint"],
|
|
"database_endpoint_identity": database_endpoint_identity,
|
|
"database_user": target["database_user"],
|
|
"runtime_commit": runtime_commit,
|
|
"commit_candidates": {f"git:{process_cwd}": runtime_commit},
|
|
"commit_identity_ambiguous": False,
|
|
"secret_values_emitted": False,
|
|
}
|
|
|
|
|
|
def observation(
|
|
label: str,
|
|
value: dict | None = None,
|
|
*,
|
|
runtime_commit: str = "a" * 40,
|
|
) -> dict:
|
|
selected = copy.deepcopy(value or manifest(label=label))
|
|
target = endpoint_target(label)
|
|
route = route_observation(label, runtime_commit=runtime_commit)
|
|
endpoint_identity = detector.validate_endpoint_binding(
|
|
label,
|
|
route,
|
|
selected["singleton"]["identity"],
|
|
target,
|
|
)
|
|
return {
|
|
"schema": detector.OBSERVATION_SCHEMA,
|
|
"label": label,
|
|
"status": "observed_current",
|
|
"observed_at_utc": "2026-07-15T20:00:00Z",
|
|
"expected_target": target,
|
|
"route": route,
|
|
"database_identity": selected["singleton"]["identity"],
|
|
"endpoint_identity": endpoint_identity,
|
|
"endpoint_identity_sha256": detector.canonical_hash(endpoint_identity),
|
|
"fingerprint": detector.manifest_fingerprint(selected),
|
|
"manifest": selected,
|
|
"cleanup": {"remote_files_created": False, "local_temp_files_created": False},
|
|
}
|
|
|
|
|
|
def live_args(tmp_path: Path, manifest_bytes: bytes) -> argparse.Namespace:
|
|
manifest_path = tmp_path / "manifest.sql"
|
|
manifest_path.write_bytes(manifest_bytes)
|
|
key = tmp_path / "key"
|
|
key.write_text("not-a-real-key")
|
|
return argparse.Namespace(
|
|
output=tmp_path / "output.json",
|
|
manifest=manifest_path,
|
|
connect_timeout=3,
|
|
probe_timeout=30,
|
|
vps_ssh_target="root@192.0.2.10",
|
|
vps_ssh_key=key,
|
|
vps_container="teleo-pg",
|
|
vps_database="teleo",
|
|
vps_database_user="postgres",
|
|
vps_service="leoclean-gateway.service",
|
|
gcp_ssh_target="teleo-gcp-staging",
|
|
gcp_host="10.61.0.3",
|
|
gcp_database="teleo",
|
|
gcp_database_user="postgres",
|
|
gcp_project="teleo-501523",
|
|
gcp_secret="gcp-teleo-pgvector-standby-postgres-password",
|
|
gcp_service="leoclean-gcp-prod-parallel.service",
|
|
)
|
|
|
|
|
|
def test_identical_current_observations_pass() -> None:
|
|
receipt = detector.evaluate_observations(
|
|
observation("vps"),
|
|
observation("gcp"),
|
|
generated_at_utc="2026-07-15T20:01:00Z",
|
|
)
|
|
|
|
assert receipt["status"] == "identical"
|
|
assert receipt["comparison"]["state"] == "identical"
|
|
assert receipt["fallback_used"] is False
|
|
assert receipt["historical_receipt_substituted"] is False
|
|
assert receipt["database_mutated"] is False
|
|
assert "manifest" not in receipt["observations"]["vps"]
|
|
|
|
|
|
def test_remote_route_inventory_source_compiles() -> None:
|
|
compile(detector.REMOTE_ROUTE_INVENTORY, "<remote-route-inventory>", "exec")
|
|
|
|
|
|
def test_vps_remote_script_executes_manifest_against_inspected_container_id() -> None:
|
|
script = detector.remote_script(
|
|
route_kind="vps_docker",
|
|
transport="root@192.0.2.10",
|
|
service="leoclean-gateway.service",
|
|
database="teleo",
|
|
container="teleo-pg",
|
|
database_user="postgres",
|
|
gcp_host=None,
|
|
gcp_project=None,
|
|
gcp_secret=None,
|
|
)
|
|
|
|
syntax = detector.subprocess.run(["bash", "-n"], input=script, text=True, capture_output=True, check=False)
|
|
|
|
assert syntax.returncode == 0, syntax.stderr
|
|
assert "docker inspect --format" in script
|
|
assert ' -i "$container_id" psql ' in script
|
|
assert "exec docker exec -e PGOPTIONS=-c default_transaction_read_only=on -i teleo-pg" not in script
|
|
|
|
|
|
def test_content_drift_fails_and_names_exact_table_mismatch() -> None:
|
|
gcp_manifest = manifest(label="gcp", row_count=4, rowset_md5="2" * 32)
|
|
receipt = detector.evaluate_observations(observation("vps"), observation("gcp", gcp_manifest))
|
|
|
|
assert receipt["status"] == "drift"
|
|
assert receipt["comparison"]["content_identical"] is False
|
|
assert "table_content_mismatches=1" in receipt["comparison"]["problems"]
|
|
assert receipt["comparison"]["details"]["table_mismatches"] == [
|
|
{
|
|
"table": "public.claims",
|
|
"mismatch": {
|
|
"row_count": {"source": 3, "target": 4},
|
|
"rowset_md5": {"source": "1" * 32, "target": "2" * 32},
|
|
},
|
|
}
|
|
]
|
|
|
|
|
|
def test_unreachable_gcp_is_incomplete_without_vps_substitution() -> None:
|
|
gcp = {
|
|
"schema": detector.OBSERVATION_SCHEMA,
|
|
"label": "gcp",
|
|
"status": "unreachable",
|
|
"failure_class": "probe_timeout",
|
|
"error": "gcp probe exceeded 30s",
|
|
"expected_target": endpoint_target("gcp"),
|
|
"manifest": None,
|
|
"route": None,
|
|
}
|
|
|
|
receipt = detector.evaluate_observations(observation("vps"), gcp)
|
|
|
|
assert receipt["status"] == "incomplete"
|
|
assert receipt["unavailable_targets"] == ["gcp"]
|
|
assert receipt["comparison"] is None
|
|
assert receipt["observations"]["gcp"]["route"] is None
|
|
assert receipt["observations"]["gcp"] != receipt["observations"]["vps"]
|
|
assert receipt["fallback_used"] is False
|
|
assert receipt["historical_receipt_substituted"] is False
|
|
|
|
|
|
def test_declared_invalid_observation_takes_precedence_over_unreachable() -> None:
|
|
vps = {
|
|
"schema": detector.OBSERVATION_SCHEMA,
|
|
"label": "vps",
|
|
"status": "invalid",
|
|
"failure_class": "invalid_probe_receipt",
|
|
"error": "manifest row is malformed",
|
|
"expected_target": endpoint_target("vps"),
|
|
"manifest": None,
|
|
"route": None,
|
|
}
|
|
gcp = {
|
|
"schema": detector.OBSERVATION_SCHEMA,
|
|
"label": "gcp",
|
|
"status": "unreachable",
|
|
"failure_class": "probe_timeout",
|
|
"error": "target unavailable",
|
|
"expected_target": endpoint_target("gcp"),
|
|
"manifest": None,
|
|
"route": None,
|
|
}
|
|
|
|
receipt = detector.evaluate_observations(vps, gcp)
|
|
|
|
assert receipt["status"] == "invalid"
|
|
assert receipt["invalid_targets"] == ["vps"]
|
|
assert receipt["unavailable_targets"] == ["gcp"]
|
|
assert receipt["comparison"] is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "mutated_value"),
|
|
[
|
|
("schema", "livingip.sourceFingerprintObservation.v0"),
|
|
("label", "vps"),
|
|
("status", "unknown"),
|
|
("endpoint_identity_sha256", "0" * 64),
|
|
],
|
|
)
|
|
def test_mutated_observation_envelope_is_invalid(field: str, mutated_value: str) -> None:
|
|
gcp = observation("gcp")
|
|
gcp[field] = mutated_value
|
|
|
|
receipt = detector.evaluate_observations(observation("vps"), gcp)
|
|
|
|
assert receipt["status"] == "invalid"
|
|
assert receipt["invalid_targets"] == ["gcp"]
|
|
assert "gcp" in receipt["validation_errors"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("route_field", "mutated_value"),
|
|
[
|
|
("transport", "other-gcp-host"),
|
|
("service", "other.service"),
|
|
("database_endpoint", "10.61.0.4:5432"),
|
|
],
|
|
)
|
|
def test_mutated_route_binding_is_invalid(route_field: str, mutated_value: str) -> None:
|
|
gcp = observation("gcp")
|
|
gcp["route"][route_field] = mutated_value
|
|
|
|
receipt = detector.evaluate_observations(observation("vps"), gcp)
|
|
|
|
assert receipt["status"] == "invalid"
|
|
assert receipt["invalid_targets"] == ["gcp"]
|
|
|
|
|
|
def test_swapped_vps_container_identity_is_invalid() -> None:
|
|
vps = observation("vps")
|
|
vps["route"]["database_endpoint_identity"]["container_name"] = "other-postgres"
|
|
|
|
receipt = detector.evaluate_observations(vps, observation("gcp"))
|
|
|
|
assert receipt["status"] == "invalid"
|
|
assert "Docker endpoint identity is invalid" in receipt["validation_errors"]["vps"]
|
|
|
|
|
|
def test_mutated_database_address_is_invalid_even_when_manifest_and_hash_are_recomputed() -> None:
|
|
gcp = observation("gcp")
|
|
gcp["database_identity"]["server_address"] = "10.61.0.4"
|
|
gcp["fingerprint"] = detector.manifest_fingerprint(gcp["manifest"])
|
|
gcp["endpoint_identity"]["server_address"] = "10.61.0.4"
|
|
gcp["endpoint_identity_sha256"] = detector.canonical_hash(gcp["endpoint_identity"])
|
|
|
|
receipt = detector.evaluate_observations(observation("vps"), gcp)
|
|
|
|
assert receipt["status"] == "invalid"
|
|
assert "address does not match" in receipt["validation_errors"]["gcp"]
|
|
|
|
|
|
def test_malformed_current_manifest_is_invalid_instead_of_incomplete() -> None:
|
|
gcp = observation("gcp")
|
|
gcp["manifest"]["tables"]["public.claims"].pop("row_count")
|
|
|
|
receipt = detector.evaluate_observations(observation("vps"), gcp)
|
|
|
|
assert receipt["status"] == "invalid"
|
|
assert receipt["invalid_targets"] == ["gcp"]
|
|
assert receipt["unavailable_targets"] == []
|
|
|
|
|
|
def test_identically_malformed_manifests_cannot_compare_as_identical() -> None:
|
|
vps = observation("vps")
|
|
gcp = observation("gcp")
|
|
for item in (vps, gcp):
|
|
item["manifest"]["singleton"]["schemas"]["items"] = {}
|
|
item["fingerprint"] = detector.manifest_fingerprint(item["manifest"])
|
|
|
|
receipt = detector.evaluate_observations(vps, gcp)
|
|
|
|
assert receipt["status"] == "invalid"
|
|
assert receipt["invalid_targets"] == ["vps", "gcp"]
|
|
|
|
|
|
def test_coherent_same_host_endpoint_aliases_are_invalid() -> None:
|
|
vps = observation("vps")
|
|
gcp = observation("gcp")
|
|
gcp["route"]["hostname"] = vps["route"]["hostname"]
|
|
gcp["endpoint_identity"]["hostname"] = vps["endpoint_identity"]["hostname"]
|
|
gcp["endpoint_identity_sha256"] = detector.canonical_hash(gcp["endpoint_identity"])
|
|
|
|
receipt = detector.evaluate_observations(vps, gcp)
|
|
|
|
assert receipt["status"] == "invalid"
|
|
assert receipt["invalid_targets"] == ["vps", "gcp"]
|
|
assert "same route hostname" in receipt["validation_errors"]["endpoint_pair"]
|
|
|
|
|
|
def test_runtime_revision_mismatch_is_drift() -> None:
|
|
receipt = detector.evaluate_observations(
|
|
observation("vps", runtime_commit="a" * 40),
|
|
observation("gcp", runtime_commit="b" * 40),
|
|
)
|
|
|
|
assert receipt["status"] == "drift"
|
|
assert receipt["comparison"]["runtime_revision_identical"] is False
|
|
assert "runtime_commit_mismatch" in receipt["comparison"]["problems"]
|
|
|
|
|
|
def test_probe_timeout_preserves_exact_expected_target(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
expected = endpoint_target("gcp")
|
|
expected["database"] = "teleo_canonical"
|
|
|
|
def timeout(*args, **kwargs):
|
|
raise detector.subprocess.TimeoutExpired(args[0], kwargs["timeout"])
|
|
|
|
monkeypatch.setattr(detector.subprocess, "run", timeout)
|
|
result = detector.collect_endpoint(
|
|
label="gcp",
|
|
command=["ssh", "teleo-gcp-staging", "--", "true"],
|
|
manifest_sql=b"select 1",
|
|
timeout=30,
|
|
expected_target=expected,
|
|
)
|
|
|
|
assert result["status"] == "unreachable"
|
|
assert result["failure_class"] == "probe_timeout"
|
|
assert result["expected_target"] == expected
|
|
assert result["route"] is None
|
|
|
|
|
|
def test_parser_rejects_non_readonly_or_wrong_database(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
selected = manifest(label="gcp")
|
|
lines = [json.dumps(row) for row in selected["singleton"].values()]
|
|
lines.extend(json.dumps(row) for row in selected["tables"].values())
|
|
expected = endpoint_target("gcp")
|
|
expected["database"] = "different"
|
|
route = route_observation("gcp")
|
|
route["expected_database"] = "different"
|
|
stdout = (detector.ROUTE_PREFIX + json.dumps(route) + "\n" + "\n".join(lines)).encode()
|
|
|
|
monkeypatch.setattr(
|
|
detector.subprocess,
|
|
"run",
|
|
lambda *args, **kwargs: detector.subprocess.CompletedProcess(args[0], 0, stdout, b""),
|
|
)
|
|
result = detector.collect_endpoint(
|
|
label="gcp",
|
|
command=["ssh", "target", "--", "true"],
|
|
manifest_sql=b"select 1",
|
|
timeout=30,
|
|
expected_target=expected,
|
|
)
|
|
|
|
assert result["status"] == "invalid"
|
|
assert result["failure_class"] == "invalid_probe_receipt"
|
|
assert "does not match the explicit target" in result["error"]
|
|
|
|
|
|
def test_collector_rejects_route_that_does_not_match_requested_endpoint(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
selected = manifest(label="gcp")
|
|
lines = [json.dumps(row) for row in selected["singleton"].values()]
|
|
lines.extend(json.dumps(row) for row in selected["tables"].values())
|
|
route = route_observation("gcp")
|
|
route["transport"] = "attacker-controlled-alias"
|
|
stdout = (detector.ROUTE_PREFIX + json.dumps(route) + "\n" + "\n".join(lines)).encode()
|
|
monkeypatch.setattr(
|
|
detector.subprocess,
|
|
"run",
|
|
lambda *args, **kwargs: detector.subprocess.CompletedProcess(args[0], 0, stdout, b""),
|
|
)
|
|
|
|
result = detector.collect_endpoint(
|
|
label="gcp",
|
|
command=["ssh", "teleo-gcp-staging", "--", "true"],
|
|
manifest_sql=b"select 1",
|
|
timeout=30,
|
|
expected_target=endpoint_target("gcp"),
|
|
)
|
|
|
|
assert result["status"] == "invalid"
|
|
assert "route transport does not match" in result["error"]
|
|
|
|
|
|
def test_live_runs_both_independent_collectors_and_returns_incomplete(tmp_path: Path) -> None:
|
|
manifest_bytes = b"reviewed manifest"
|
|
called = []
|
|
|
|
def fake_collector(**kwargs):
|
|
called.append(kwargs["label"])
|
|
if kwargs["label"] == "vps":
|
|
return observation("vps")
|
|
return {
|
|
"schema": detector.OBSERVATION_SCHEMA,
|
|
"label": "gcp",
|
|
"status": "unreachable",
|
|
"failure_class": "transport_or_remote_probe_failed",
|
|
"error": "target unavailable",
|
|
"expected_target": endpoint_target("gcp"),
|
|
"manifest": None,
|
|
"route": None,
|
|
}
|
|
|
|
monkeypatch = pytest.MonkeyPatch()
|
|
monkeypatch.setattr(detector, "REVIEWED_MANIFEST_SQL_SHA256", hashlib.sha256(manifest_bytes).hexdigest())
|
|
args = live_args(tmp_path, manifest_bytes)
|
|
try:
|
|
receipt, exit_code = detector.run_live(args, collector=fake_collector)
|
|
finally:
|
|
monkeypatch.undo()
|
|
|
|
assert sorted(called) == ["gcp", "vps"]
|
|
assert receipt["status"] == "incomplete"
|
|
assert receipt["manifest_sql_sha256"] == hashlib.sha256(manifest_bytes).hexdigest()
|
|
assert exit_code == 2
|
|
|
|
|
|
def test_live_invalid_collector_payload_returns_exit_3(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
manifest_bytes = b"reviewed manifest"
|
|
args = live_args(tmp_path, manifest_bytes)
|
|
monkeypatch.setattr(detector, "REVIEWED_MANIFEST_SQL_SHA256", hashlib.sha256(manifest_bytes).hexdigest())
|
|
|
|
def fake_collector(**kwargs):
|
|
result = observation(kwargs["label"])
|
|
if kwargs["label"] == "gcp":
|
|
result.pop("endpoint_identity_sha256")
|
|
return result
|
|
|
|
receipt, exit_code = detector.run_live(args, collector=fake_collector)
|
|
|
|
assert receipt["status"] == "invalid"
|
|
assert receipt["invalid_targets"] == ["gcp"]
|
|
assert exit_code == 3
|
|
|
|
|
|
def test_live_rejects_collector_that_coherently_retargets_an_endpoint(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
manifest_bytes = b"reviewed manifest"
|
|
args = live_args(tmp_path, manifest_bytes)
|
|
monkeypatch.setattr(detector, "REVIEWED_MANIFEST_SQL_SHA256", hashlib.sha256(manifest_bytes).hexdigest())
|
|
|
|
def fake_collector(**kwargs):
|
|
result = observation(kwargs["label"])
|
|
if kwargs["label"] == "gcp":
|
|
result["expected_target"]["transport"] = "other-gcp-host"
|
|
result["route"]["transport"] = "other-gcp-host"
|
|
result["endpoint_identity"]["transport"] = "other-gcp-host"
|
|
result["endpoint_identity_sha256"] = detector.canonical_hash(result["endpoint_identity"])
|
|
return result
|
|
|
|
receipt, exit_code = detector.run_live(args, collector=fake_collector)
|
|
|
|
assert receipt["status"] == "invalid"
|
|
assert exit_code == 3
|
|
assert "requested collector target" in receipt["validation_errors"]["gcp"]
|
|
|
|
|
|
def test_interruption_removes_stale_receipt_before_probe(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
output = tmp_path / "current.json"
|
|
output.write_text('{"status": "identical", "run": "stale"}\n')
|
|
args = argparse.Namespace(output=output)
|
|
monkeypatch.setattr(detector, "parse_args", lambda: args)
|
|
|
|
def interrupt(_args):
|
|
assert not output.exists()
|
|
raise KeyboardInterrupt
|
|
|
|
monkeypatch.setattr(detector, "run_live", interrupt)
|
|
|
|
with pytest.raises(KeyboardInterrupt):
|
|
detector.main()
|
|
|
|
assert not output.exists()
|
|
|
|
|
|
def test_interruption_during_atomic_publication_cannot_restore_stale_receipt(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
output = tmp_path / "current.json"
|
|
output.write_text('{"status": "identical", "run": "stale"}\n')
|
|
args = argparse.Namespace(output=output)
|
|
receipt = {"schema": detector.RECEIPT_SCHEMA, "status": "identical"}
|
|
monkeypatch.setattr(detector, "parse_args", lambda: args)
|
|
monkeypatch.setattr(detector, "run_live", lambda _args: (receipt, 0))
|
|
monkeypatch.setattr(
|
|
private_receipt_io.os,
|
|
"replace",
|
|
lambda _source, _destination: (_ for _ in ()).throw(KeyboardInterrupt()),
|
|
)
|
|
|
|
with pytest.raises(KeyboardInterrupt):
|
|
detector.main()
|
|
|
|
assert not output.exists()
|
|
assert list(tmp_path.glob(".*.tmp")) == []
|
|
|
|
|
|
def test_successful_atomic_publication_replaces_stale_receipt(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
output = tmp_path / "private" / "current.json"
|
|
output.parent.mkdir()
|
|
output.write_text('{"status": "identical", "run": "stale"}\n')
|
|
args = argparse.Namespace(output=output)
|
|
receipt = {"schema": detector.RECEIPT_SCHEMA, "status": "invalid", "run": "current"}
|
|
monkeypatch.setattr(detector, "parse_args", lambda: args)
|
|
monkeypatch.setattr(detector, "run_live", lambda _args: (receipt, 3))
|
|
|
|
assert detector.main() == 3
|
|
|
|
assert json.loads(output.read_text()) == receipt
|
|
assert output.stat().st_mode & 0o777 == 0o600
|
|
assert str(output.parent) not in capsys.readouterr().out
|
|
|
|
|
|
def test_sanitizer_redacts_connection_secrets() -> None:
|
|
error = detector.sanitize_error(
|
|
"fatal password=hunter2 postgresql://user:pw@db.internal/teleo",
|
|
target="db.internal",
|
|
)
|
|
|
|
assert "hunter2" not in error
|
|
assert "user:pw" not in error
|
|
assert "[REDACTED]" in error
|