tegwick
c49c6b39ce
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 6m35s
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
94 lines
4.3 KiB
Python
94 lines
4.3 KiB
Python
"""No network or provider sends: verify the proof's no-replay and approval boundaries."""
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"backup_email_proof", Path(__file__).parents[1] / "scripts/verify-backup-alert-email.py"
|
|
)
|
|
proof = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(proof)
|
|
|
|
|
|
def test_changed_digest_and_nonterminal_refuse_before_ssh(monkeypatch):
|
|
def forbidden(*args, **kwargs):
|
|
pytest.fail("SSH must not run")
|
|
monkeypatch.setattr(proof.subprocess, "run", forbidden)
|
|
with pytest.raises(SystemExit, match="changed"):
|
|
proof.main(["--execute", "--expect-digest", "sha256:wrong"])
|
|
monkeypatch.setattr(proof.sys, "stdin", io.StringIO())
|
|
with pytest.raises(SystemExit, match="Founder terminal"):
|
|
proof.main(["--execute", "--expect-digest", proof.prepare()[2]])
|
|
|
|
|
|
def run_node(monkeypatch, capsys, mode, existing=False, wait_failed=False):
|
|
plan, job, digest = proof.prepare()
|
|
calls = []
|
|
def run(args, **kwargs):
|
|
calls.append(args)
|
|
code = 0
|
|
value = {}
|
|
if args[1:3] == ["get", "namespace"]:
|
|
value = {"metadata": {"uid": plan["cluster_uid"] if args[3] == "kube-system"
|
|
else plan["namespace_uid"]}}
|
|
elif args[1:3] == ["get", "networkpolicy"]:
|
|
value = {"spec": plan["smtp_policy_spec"]}
|
|
elif args[1:3] == ["get", "externalsecret"]:
|
|
value = {"spec": {"target": {"name": plan["secret_reference"]}},
|
|
"status": {"conditions": [{"type": "Ready", "status": "True"}]}}
|
|
elif "--ignore-not-found" in args:
|
|
return SimpleNamespace(returncode=0, stdout=b'{"metadata":{}}' if existing else b'')
|
|
elif args[1] == "create":
|
|
value = {"metadata": {"uid": "synthetic-job-uid"}}
|
|
elif args[1] == "wait" and wait_failed:
|
|
code = 1
|
|
elif args[1:3] == ["get", "job"]:
|
|
value = {"metadata": {"uid": "synthetic-job-uid"}, "status": {"succeeded": 1}}
|
|
elif args[1] == "logs":
|
|
value = {"status": "smtp-accepted", "message_id": plan["message_id"],
|
|
"recipient_delivery_confirmed": False, "alert_acknowledged": False}
|
|
return SimpleNamespace(returncode=code, stdout=json.dumps(value).encode())
|
|
monkeypatch.setattr(subprocess, "run", run)
|
|
import socket
|
|
monkeypatch.setattr(socket, "getaddrinfo", lambda *a: [
|
|
(None, None, None, None, (ip, 587)) for ip in plan["smtp_ipv4"]])
|
|
monkeypatch.setattr(proof.sys, "stdin", io.StringIO(json.dumps(
|
|
{"plan": plan, "job": job, "digest": digest, "mode": mode})))
|
|
try:
|
|
exec(compile(proof.NODE, "node-proof", "exec"), {})
|
|
except SystemExit as error:
|
|
assert error.code == 1
|
|
return json.loads(capsys.readouterr().out), calls
|
|
|
|
|
|
def test_preflight_never_creates_or_sends(monkeypatch, capsys):
|
|
result, calls = run_node(monkeypatch, capsys, "preflight")
|
|
assert result["status"] == "preflight_verified" and result["email_sent"] is False
|
|
assert all("--dry-run=server" in c for c in calls if c[1] == "create")
|
|
assert not any(c[1] in ("wait", "logs") for c in calls)
|
|
|
|
|
|
def test_existing_job_refuses_replay(monkeypatch, capsys):
|
|
result, calls = run_node(monkeypatch, capsys, "execute", existing=True)
|
|
assert result["status"] == "execution_requires_inspection"
|
|
assert not any(c[1] == "create" for c in calls)
|
|
|
|
|
|
def test_uncertain_send_never_retries_or_claims_no_send(monkeypatch, capsys):
|
|
result, calls = run_node(monkeypatch, capsys, "execute", wait_failed=True)
|
|
assert result["email_sent"] is None and result["recipient_confirmed"] is False
|
|
assert sum(c[1] == "create" and "--dry-run=server" not in c for c in calls) == 1
|
|
assert not any(c[1] in ("delete", "replace", "apply") for c in calls)
|
|
|
|
|
|
def test_provider_acceptance_is_not_recipient_confirmation(monkeypatch, capsys):
|
|
result, calls = run_node(monkeypatch, capsys, "execute")
|
|
assert result["status"] == "smtp_provider_accepted" and result["email_sent"] is True
|
|
assert result["recipient_confirmed"] is False
|
|
assert result["automatic_monitoring_active"] is False
|
|
assert sum(c[1] == "create" and "--dry-run=server" not in c for c in calls) == 1
|