171 lines
6.5 KiB
Python
171 lines
6.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scripts import release_provenance_gate
|
|
|
|
|
|
def test_documented_python_invocation_loads_project_without_pythonpath():
|
|
script = Path(release_provenance_gate.__file__).resolve()
|
|
env = os.environ.copy()
|
|
env.pop("PYTHONPATH", None)
|
|
|
|
completed = subprocess.run(
|
|
[sys.executable, str(script), "--help"],
|
|
cwd=script.parent,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert "--observed-evidence-file" in completed.stdout
|
|
assert script.stat().st_mode & 0o111 != 0
|
|
|
|
|
|
def _init_repo(tmp_path: Path) -> tuple[Path, str]:
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
|
|
subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=repo, check=True)
|
|
subprocess.run(["git", "config", "user.name", "Test"], cwd=repo, check=True)
|
|
(repo / "runtime.py").write_text("print('ok')\n", encoding="utf-8")
|
|
subprocess.run(["git", "add", "runtime.py"], cwd=repo, check=True)
|
|
subprocess.run(["git", "commit", "-m", "init"], cwd=repo, check=True, capture_output=True)
|
|
head = subprocess.run(
|
|
["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True
|
|
).stdout.strip()
|
|
subprocess.run(["git", "checkout", "--detach", head], cwd=repo, check=True, capture_output=True)
|
|
return repo, head
|
|
|
|
|
|
def _fake_read_only_commands(tmp_path: Path) -> tuple[Path, Path]:
|
|
bindir = tmp_path / "bin"
|
|
bindir.mkdir(parents=True)
|
|
trex = bindir / "trex-core"
|
|
rabbit = bindir / "rabbitmqctl"
|
|
trex.write_text("#!/bin/sh\nprintf '{}\\n'\n", encoding="utf-8")
|
|
rabbit.write_text("#!/bin/sh\nprintf 'consumer\\n'\n", encoding="utf-8")
|
|
trex.chmod(0o755)
|
|
rabbit.chmod(0o755)
|
|
return trex, rabbit
|
|
|
|
|
|
def _argv(tmp_path: Path, repo: Path, head: str) -> list[str]:
|
|
trex, rabbit = _fake_read_only_commands(tmp_path)
|
|
broker_log = tmp_path / "broker.log"
|
|
broker_log.write_text("event=order_submit account=must-not-leak\n", encoding="utf-8")
|
|
evidence = tmp_path / "observed-evidence.json"
|
|
evidence.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"target_id": "service-a",
|
|
"target_kind": "compose-service",
|
|
"image_id": "sha256:" + ("b" * 64),
|
|
"image_digest": "sha256:" + ("a" * 64),
|
|
"oci_revision": head,
|
|
"observed_at": "2026-09-20T02:00:00Z",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return [
|
|
"--actor", "operator-a",
|
|
"--reason", "deploy release",
|
|
"--instance", "inst-a",
|
|
"--operation-id", "op-123",
|
|
"--phase", "pre",
|
|
"--command-class", "deploy",
|
|
"--checkout-dir", str(repo),
|
|
"--observed-evidence-file", str(evidence),
|
|
"--runtime-file", "runtime.py",
|
|
"--broker-log-path", str(broker_log),
|
|
"--capture-command", f"pending_slots={trex} slots --status pending --instance inst-a --json",
|
|
"--capture-command", f"reconciliation_slots={trex} slots --status all --instance inst-a --json",
|
|
"--capture-command", f"queue_consumers={rabbit} list_consumers",
|
|
]
|
|
|
|
|
|
def test_cli_writes_unique_record_outside_checkout_and_keeps_checkout_clean(tmp_path, capsys):
|
|
repo, head = _init_repo(tmp_path)
|
|
argv = _argv(tmp_path, repo, head)
|
|
|
|
assert release_provenance_gate.main(argv) == 0
|
|
first_receipt = json.loads(capsys.readouterr().out)
|
|
first = Path(first_receipt["record_path"])
|
|
assert release_provenance_gate.main(argv) == 0
|
|
second_receipt = json.loads(capsys.readouterr().out)
|
|
second = Path(second_receipt["record_path"])
|
|
|
|
assert first != second
|
|
assert first.parent == tmp_path / "repo-release-provenance"
|
|
assert first.stat().st_mode & 0o222 == 0
|
|
assert second.stat().st_mode & 0o222 == 0
|
|
payload = json.loads(first.read_text(encoding="utf-8"))
|
|
assert payload["schema_version"] == 1
|
|
assert payload["observed_evidence"]["source"] == "operator_observed"
|
|
assert first_receipt["record_sha256"] == __import__("hashlib").sha256(first.read_bytes()).hexdigest()
|
|
assert payload["gate"]["passed"] is True
|
|
assert "must-not-leak" not in first.read_text(encoding="utf-8")
|
|
status = subprocess.run(
|
|
["git", "status", "--porcelain"], cwd=repo, check=True, capture_output=True, text=True
|
|
).stdout
|
|
assert status == ""
|
|
|
|
|
|
def test_cli_dirty_checkout_is_fail_closed_and_isolated(tmp_path, capsys):
|
|
repo, head = _init_repo(tmp_path)
|
|
argv = _argv(tmp_path, repo, head)
|
|
(repo / "runtime.py").write_text("dirty\n", encoding="utf-8")
|
|
|
|
assert release_provenance_gate.main(argv) == 2
|
|
receipt = json.loads(capsys.readouterr().out)
|
|
output = Path(receipt["record_path"])
|
|
payload = json.loads(output.read_text(encoding="utf-8"))
|
|
assert output.stat().st_mode & 0o222 == 0
|
|
assert payload["gate"]["passed"] is False
|
|
assert payload["gate"]["deployment_blocked"] is True
|
|
assert payload["state_capture"]["command_outputs"] == {}
|
|
snapshot = Path(payload["dirty_checkout_isolation"]["isolate_dir"])
|
|
assert snapshot.parent == tmp_path / "repo-release-provenance-isolation"
|
|
assert (snapshot / "runtime.py").read_text() == "dirty\n"
|
|
|
|
|
|
def test_cli_rejects_unsafe_command_and_checkout_local_output(tmp_path):
|
|
repo, head = _init_repo(tmp_path)
|
|
argv = _argv(tmp_path, repo, head)
|
|
unsafe = argv.index(next(x for x in argv if x.startswith("pending_slots=")))
|
|
argv[unsafe] = "pending_slots=trex-core cancel slot-1 --yes"
|
|
with pytest.raises(SystemExit):
|
|
release_provenance_gate.main(argv)
|
|
|
|
argv = _argv(tmp_path / "other", repo, head)
|
|
argv.extend(["--output-dir", str(repo / "artifacts")])
|
|
with pytest.raises(SystemExit):
|
|
release_provenance_gate.main(argv)
|
|
|
|
|
|
def test_invalid_observed_evidence_creates_no_record_or_receipt(tmp_path, capsys):
|
|
repo, head = _init_repo(tmp_path)
|
|
argv = _argv(tmp_path, repo, head)
|
|
evidence_index = argv.index("--observed-evidence-file") + 1
|
|
evidence = Path(argv[evidence_index])
|
|
evidence.write_text(
|
|
'{"schema_version":1,"schema_version":1,"target_id":"service-a"}',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(SystemExit):
|
|
release_provenance_gate.main(argv)
|
|
|
|
assert capsys.readouterr().out == ""
|
|
assert not (tmp_path / "repo-release-provenance").exists()
|