tegwick
fed119a17c
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181
93 lines
3.6 KiB
Python
93 lines
3.6 KiB
Python
import hashlib
|
|
import json
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from activity_core.custodian_inputs import admit
|
|
|
|
|
|
@pytest.fixture
|
|
def evidence(tmp_path):
|
|
fleet = tmp_path / "fleet"
|
|
file = fleet / "example/workplans/task.md"
|
|
file.parent.mkdir(parents=True)
|
|
file.write_text("task")
|
|
manifest = {"schema": "repo-manager.fleet-input.v1", "freshness_verified": False,
|
|
"repositories": [{"slug": "example", "commit": "a" * 40, "files": [
|
|
{"path": "workplans/task.md", "bytes": 4, "sha256": hashlib.sha256(b"task").hexdigest()}]}]}
|
|
raw = json.dumps(manifest).encode()
|
|
(fleet / "manifest.json").write_bytes(raw)
|
|
(fleet / "COMPLETE").write_text(hashlib.sha256(raw).hexdigest())
|
|
policy = tmp_path / "policy.json"
|
|
policy.write_text(json.dumps({"schema": "activity-core.custodian-input-policy.v1", "repositories": ["example"],
|
|
"max_observation_age_seconds": 7200, "max_future_skew_seconds": 300,
|
|
"max_snapshot_bytes": 1024, "expected_hub_instance": "railiance01"}))
|
|
receipts = tmp_path / "receipts.json"
|
|
receipt = {"schema": "repo-manager.freshness-receipt.v1", "repo_slug": "example", "forge_commit": "a" * 40,
|
|
"outcome": "noop", "reason": "forge_head_matches_last_success", "finished_at": "2026-10-19T10:00:00Z",
|
|
"hub": {"verified_primary": True, "instance_role": "primary", "instance_label": "railiance01"}}
|
|
prior = {**receipt, "outcome": "applied", "reason": None,
|
|
"finished_at": "2026-10-01T00:00:00Z",
|
|
**{key: "a" * 40 for key in ("expected_commit", "derived_commit", "projected_commit")}}
|
|
receipts.write_text(json.dumps([receipt, prior]))
|
|
return fleet, policy, receipts
|
|
|
|
|
|
def check(evidence):
|
|
return admit(*evidence, ["example"], now=datetime(2026, 10, 19, 11, tzinfo=UTC))
|
|
|
|
|
|
def test_accepts_independent_recent_receipt(evidence):
|
|
assert len(check(evidence)) == 64
|
|
|
|
|
|
@pytest.mark.parametrize("case", ["missing", "stale", "future", "wrong_commit", "wrong_hub", "failed"])
|
|
def test_rejects_invalid_freshness(evidence, case):
|
|
path = evidence[2]
|
|
rows = json.loads(path.read_text())
|
|
if case == "missing":
|
|
rows = []
|
|
elif case == "stale":
|
|
rows[0]["finished_at"] = "2026-10-18T10:00:00Z"
|
|
elif case == "future":
|
|
rows[0]["finished_at"] = "2026-10-20T10:00:00Z"
|
|
elif case == "wrong_commit":
|
|
rows[0]["forge_commit"] = "b" * 40
|
|
elif case == "wrong_hub":
|
|
rows[0]["hub"]["instance_label"] = "workstation"
|
|
else:
|
|
rows[0]["outcome"] = "failed"
|
|
path.write_text(json.dumps(rows))
|
|
with pytest.raises(ValueError, match="fresh Forge receipt"):
|
|
check(evidence)
|
|
|
|
|
|
@pytest.mark.parametrize("case", ["tamper", "extra", "marker", "extra_repo"])
|
|
def test_rejects_changed_snapshot(evidence, case):
|
|
fleet = evidence[0]
|
|
if case == "tamper":
|
|
(fleet / "example/workplans/task.md").write_text("changed")
|
|
elif case == "extra":
|
|
(fleet / "example/workplans/extra.md").write_text("extra")
|
|
elif case == "extra_repo":
|
|
(fleet / "outside").mkdir()
|
|
else:
|
|
(fleet / "COMPLETE").write_text("incorrect")
|
|
with pytest.raises(ValueError):
|
|
check(evidence)
|
|
|
|
|
|
def test_native_noop_uses_protected_issuer_contract(evidence):
|
|
path = evidence[2]
|
|
path.write_text(json.dumps(json.loads(path.read_text())[:1]))
|
|
assert len(check(evidence)) == 64
|
|
|
|
|
|
def test_fifo_refused_without_open(evidence):
|
|
import os
|
|
path = evidence[2]
|
|
path.unlink()
|
|
os.mkfifo(path)
|
|
with pytest.raises(ValueError, match="unsafe"):
|
|
check(evidence)
|