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

Implement authenticated custody observations and durable root-grant baseline
Some checks failed
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details
Build and Publish Container Image / build-and-push (push) Successful in 27s
Details
Account journey acceptance / journeys (push) Failing after 6s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29
This commit is contained in:
tegwick 2026-10-03 19:50:15 +02:00
parent 1999a144c9
commit 0d0751d39d
18 changed files with 1129 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

17
tests/test_custody_adapter.py
View file

@ -8,12 +8,7 @@ from datetime import datetime, timedelta, timezone
from user_engine.custody import CustodyBaseline, CustodyObservationAdapter, CustodyRead
CANON = Path(__file__).resolve().parents[2] / "net-kingdom/tools/custody-record"
try:
spec = importlib.util.spec_from_file_location("custody_canon", CANON / "validate.py")
policy = importlib.util.module_from_spec(spec)
spec.loader.exec_module(policy)
except (FileNotFoundError, ModuleNotFoundError):
policy = None
from user_engine.custody_canon import policy
class Baselines:
@ -31,7 +26,7 @@ class Baselines:
return True
@unittest.skipIf(policy is None, "optional sibling NetKingdom validator/jsonschema absent")
@unittest.skipUnless(CANON.exists(), "optional sibling NetKingdom fixtures absent")
class CustodyAdapterTests(unittest.TestCase):
def setUp(self):
self.document = json.loads((CANON / "fixtures/valid/grant.json").read_text())
@ -106,6 +101,14 @@ class CustodyAdapterTests(unittest.TestCase):
self.assertIsNone(self.observe())
self.assertIsNone(self.baselines.value)
def test_check_clock_sampled_after_owner_read(self):
self.adapter.clock = lambda: self.now + timedelta(seconds=2)
self.read_at = self.now + timedelta(seconds=1)
result = self.adapter.observe(**self.subject, tenant="platform")
self.assertIsNotNone(result)
self.assertEqual(result["observed_at"], self.read_at.isoformat().replace("+00:00", "Z"))
def test_bootstrap_and_persistence_conflict_deny(self):
self.adapter.allow_initial_baseline = False
self.assertIsNone(self.observe())