fa958cc051
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29
140 lines
6.3 KiB
Python
140 lines
6.3 KiB
Python
import copy
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
import root_grant_attended as attended
|
|
|
|
spec = importlib.util.spec_from_file_location("custody_validator", ROOT.parent / "net-kingdom/tools/custody-record/validate.py")
|
|
validator = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(validator)
|
|
|
|
|
|
class FakeBao:
|
|
def __init__(self, document=None, version=1):
|
|
self.actual = None if document is None else {"data": {"document": copy.deepcopy(document)}, "metadata": {"version": version}}
|
|
self.metadata = {"max_versions": 123, "cas_required": False}
|
|
self.config = {"auth/kubernetes/config": {}, "sys/audit": {"file/": {"type": "file"}}, attended.META: self.metadata}
|
|
self.writes = []
|
|
self.fail_after_write = False
|
|
|
|
def read(self, path, optional=False):
|
|
return copy.deepcopy(self.actual if path == attended.PATH else self.config.get(path))
|
|
|
|
def write(self, path, payload):
|
|
self.writes.append((path, copy.deepcopy(payload)))
|
|
if path == attended.PATH:
|
|
version = 0 if self.actual is None else self.actual["metadata"]["version"]
|
|
if payload["options"]["cas"] != version:
|
|
raise attended.Refused("cas_conflict")
|
|
self.actual = {"data": payload["data"], "metadata": {"version": version + 1}}
|
|
if self.fail_after_write:
|
|
raise attended.Refused("ambiguous_timeout")
|
|
else:
|
|
self.config[path] = payload
|
|
if path == attended.META:
|
|
self.metadata.update(payload)
|
|
self.config[path] = self.metadata
|
|
|
|
|
|
class AttendedTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.state = Path(self.temp.name) / "baseline.json"
|
|
self.first = json.loads((ROOT.parent / "net-kingdom/tools/custody-record/fixtures/valid/grant.json").read_text())
|
|
self.first["record_id"] = attended.RECORD_ID
|
|
self.next = dict(copy.deepcopy(self.first), revision=self.first["revision"] + 1)
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def test_first_publication_and_update_keep_history(self):
|
|
bao = FakeBao()
|
|
receipt = attended.publish(bao, self.first, self.state, validator, "human-receipt")
|
|
self.assertNotIn("document", receipt)
|
|
attended.publish(bao, self.next, self.state, validator, "human-receipt-2")
|
|
self.assertEqual(len(list((self.state.parent / "history").glob("*.json"))), 2)
|
|
self.assertEqual(bao.writes[1][1]["options"], {"cas": 1})
|
|
|
|
def test_exact_replay_creates_no_kv_version_or_history(self):
|
|
bao = FakeBao()
|
|
attended.publish(bao, self.first, self.state, validator, "human-receipt")
|
|
result = attended.publish(bao, self.first, self.state, validator, "human-receipt")
|
|
self.assertTrue(result["unchanged"])
|
|
self.assertEqual(len(bao.writes), 1)
|
|
self.assertEqual(len(list((self.state.parent / "history").glob("*.json"))), 1)
|
|
|
|
def test_existing_receipt_prevents_any_live_call(self):
|
|
from unittest.mock import patch
|
|
receipt = Path(self.temp.name) / "receipt.json"
|
|
receipt.write_text('{"ok":true}')
|
|
with patch.object(sys, "argv", ["executor", "provision", "--receipt", str(receipt)]), patch.object(attended, "require_attended") as check:
|
|
self.assertEqual(attended.main(), 1)
|
|
check.assert_not_called()
|
|
self.assertEqual(receipt.read_text(), '{"ok":true}')
|
|
|
|
def test_existing_source_without_baseline_is_not_bootstrap(self):
|
|
bao = FakeBao(self.first)
|
|
with self.assertRaises(attended.Refused):
|
|
attended.publish(bao, self.next, self.state, validator, "human-receipt")
|
|
self.assertEqual(bao.writes, [])
|
|
|
|
def test_rollback_restore_denied(self):
|
|
bao = FakeBao()
|
|
attended.publish(bao, self.first, self.state, validator, "human-receipt")
|
|
attended.publish(bao, self.next, self.state, validator, "human-receipt")
|
|
state = json.loads(self.state.read_text())
|
|
with self.assertRaises(attended.Refused):
|
|
attended.check_baseline(FakeBao(self.first).actual, state, validator)
|
|
|
|
def test_ambiguous_network_failure_leaves_durable_pending(self):
|
|
bao = FakeBao()
|
|
bao.fail_after_write = True
|
|
with self.assertRaises(attended.Refused):
|
|
attended.publish(bao, self.first, self.state, validator, "human-receipt")
|
|
self.assertIn("pending", json.loads(self.state.read_text()))
|
|
bao.fail_after_write = False
|
|
attended.publish(bao, self.next, self.state, validator, "human-receipt-2")
|
|
self.assertNotIn("pending", json.loads(self.state.read_text()))
|
|
self.assertEqual(len(bao.writes), 2)
|
|
self.assertEqual(len(list((self.state.parent / "history").glob("*.json"))), 2)
|
|
|
|
def test_pending_mismatch_requires_review_and_no_retry(self):
|
|
bao = FakeBao()
|
|
bao.fail_after_write = True
|
|
with self.assertRaises(attended.Refused):
|
|
attended.publish(bao, self.first, self.state, validator, "human-receipt")
|
|
bao.actual = None
|
|
with self.assertRaises(attended.Refused):
|
|
attended.publish(bao, self.next, self.state, validator, "human-receipt")
|
|
self.assertEqual(len(bao.writes), 1)
|
|
|
|
def test_exact_admission_is_idempotent_and_retention_unchanged(self):
|
|
bao = FakeBao()
|
|
attended.provision(bao)
|
|
self.assertEqual(bao.config[attended.ROLE_PATH], attended.ROLE)
|
|
self.assertTrue(bao.metadata["cas_required"])
|
|
self.assertEqual(bao.metadata["max_versions"], 123)
|
|
writes = len(bao.writes)
|
|
attended.provision(bao)
|
|
self.assertEqual(writes, len(bao.writes))
|
|
|
|
def test_different_existing_role_and_missing_audit_are_refused_before_write(self):
|
|
for mode in ["role", "audit"]:
|
|
bao = FakeBao()
|
|
if mode == "role":
|
|
bao.config[attended.ROLE_PATH] = dict(attended.ROLE, bound_service_account_names=["other"])
|
|
else:
|
|
bao.config["sys/audit"] = {}
|
|
with self.assertRaises(attended.Refused):
|
|
attended.provision(bao)
|
|
self.assertEqual(bao.writes, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|