coulomb/activity-core
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: select retained Custodian snapshot through pinned descriptor

Browse source 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181
This commit is contained in:
tegwick 2026-10-19 15:22:22 +02:00
parent fed119a17c
commit 3b25aea7b0
6 changed files with 150 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

25
tests/test_custodian_inputs.py
View file

@ -91,3 +91,28 @@ def test_fifo_refused_without_open(evidence):
os.mkfifo(path)
with pytest.raises(ValueError, match="unsafe"):
check(evidence)
def test_descriptor_pins_snapshot_and_receipts(evidence, tmp_path):
from activity_core.custodian_inputs import select_snapshot
fleet, policy, receipts = evidence
root = tmp_path / "snapshots"
root.mkdir()
fleet.rename(root / "capture-1")
state = tmp_path / "state"
(state / "inputs").mkdir(parents=True)
retained = state / "inputs/capture-1.receipts.json"
retained.write_bytes(receipts.read_bytes())
descriptor = state / "inputs/current.json"
data = {"schema": "activity-core.custodian-snapshot-selection.v1", "snapshot_id": "capture-1",
"manifest_sha256": hashlib.sha256((root / "capture-1/manifest.json").read_bytes()).hexdigest(),
"receipts_sha256": hashlib.sha256(retained.read_bytes()).hexdigest()}
descriptor.write_text(json.dumps(data))
assert select_snapshot(root, state, descriptor) == (root / "capture-1", retained)
retained.write_text("[]")
with pytest.raises(ValueError, match="digest mismatch"):
select_snapshot(root, state, descriptor)
data["snapshot_id"] = "../escape"
descriptor.write_text(json.dumps(data))
with pytest.raises(ValueError, match="invalid snapshot"):
select_snapshot(root, state, descriptor)