Prepare source-bound one-use catalogue maintenance exception

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

tests/test_catalogue_bootstrap.py

@@ -0,0 +1,106 @@
+"""Real isolated PostgreSQL tests; no native estate connection/identity claims."""
+import importlib.util
+from pathlib import Path
+import unittest,json,uuid,os,hashlib
+from datetime import datetime,timedelta,timezone
+from dataclasses import replace
+import psycopg
+from user_engine.adapters.postgres import PostgresUserEngineStore
+from user_engine.domain import AuthorizationDecision,AuthorizationEffect,Application
+spec=importlib.util.spec_from_file_location('catalogue_bootstrap',Path(__file__).resolve().parents[1]/'tools'/'catalogue_bootstrap.py')
+m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
+class Deny:
+ def check(self,request):return AuthorizationDecision(AuthorizationEffect.DENY,decision_id='fixture-pdp-denial',reason='no_matching_role_or_context')
+class Unavailable:
+ def check(self,request):return AuthorizationDecision(AuthorizationEffect.DENY,decision_id=None,reason='authorization service unavailable')
+class Allow:
+ def check(self,request):return AuthorizationDecision.for_standalone(AuthorizationEffect.ALLOW,reason='fixture-ordinary-allow')
+@unittest.skipUnless(os.environ.get('USER_ENGINE_ISOLATED_CATALOGUE_PG_TEST')=='network-none-fixture', 'requires explicitly isolated disposable PostgreSQL fixture; never uses runtime credentials')
+class Tests(unittest.TestCase):
+ def setUp(self):
+  self.conn=psycopg.connect('host=127.0.0.1 user=postgres dbname=postgres')
+  with self.conn.cursor() as c:c.execute('DROP SCHEMA public CASCADE');c.execute('CREATE SCHEMA public')
+  self.conn.commit();self.store=PostgresUserEngineStore(self.conn);self.store.migrate()
+  self.g={'schema':m.SCHEMA,'proposal_id':m.PROPOSAL,'approval_id':str(uuid.uuid4()),'memo_id':'infd-20261011-user-catalogue-bootstrap-r1','memo_version':1,'approval_receipt_sha256':'a'*64,'executor_sha256':'b'*64,'records_sha256':m.digest(m.records()),'expires_at':(datetime.now(timezone.utc)+timedelta(minutes=10)).isoformat(),'operator_id':'fixture-supervised-maintenance','authority':'founder-approved-one-time-administrative-exception'}
+ def tearDown(self):self.conn.close()
+ def run_op(self,mode='apply',auth=None,grant=None):
+  g=grant or self.g
+  return m.execute(self.conn,auth or Deny(),g,expected_grant_sha256=m.digest(g),actual_executor_sha256='b'*64,mode=mode)
+ def counts(self):
+  with self.conn.cursor() as c:
+   counts=[]
+   for table in ['user_engine_records','user_engine_audit_records','user_engine_outbox_events']:c.execute('SELECT count(*) FROM '+table);counts.append(c.fetchone()[0])
+  self.conn.rollback();return counts
+ def test_absent_registration_preserves_real_denial_and_explicit_maintenance_attribution(self):
+  result=self.run_op();self.assertEqual(result['result'],'created');self.assertEqual(self.counts(),[2,3,1]);self.assertFalse(result['ordinary_pdp_allow_claimed'])
+  audits=self.store.audit_log();self.assertTrue(any(x.summary=='authorization denied' and x.decision_id=='fixture-pdp-denial' for x in audits));self.assertTrue(all(x.actor.issuer=='urn:supervised-maintenance' and not x.actor.roles and not x.actor.assurance for x in audits));event=self.store.outbox_history()[0];self.assertTrue(event.payload['maintenance_exception']);self.assertEqual(event.payload['approval_id'],self.g['approval_id'])
+ def test_dry_run_rolls_back_application_binding_all_audit_and_outbox(self):
+  result=self.run_op('dry-run');self.assertTrue(result['dry_run_rolled_back']);self.assertEqual(self.counts(),[0,0,0]);self.assertEqual(self.run_op()['result'],'created')
+ def test_grant_is_closed_after_one_use_across_fresh_connection(self):
+  self.run_op();self.conn.close();self.conn=psycopg.connect('host=127.0.0.1 user=postgres dbname=postgres')
+  with self.assertRaises(m.Refused):self.run_op()
+  self.assertEqual(self.counts(),[2,3,1])
+ def test_exact_preexisting_match_is_nochange_and_has_no_undo(self):
+  with self.store.transaction():self.store.save_application(m.APP);self.store.save_binding(m.BINDING)
+  self.assertEqual(self.run_op()['result'],'no-change');self.assertEqual(self.counts(),[2,1,0])
+  with self.assertRaises(m.Refused):self.run_op('rollback')
+ def test_conflicting_or_partial_metadata_refuses_without_mutation(self):
+  with self.store.transaction():self.store.save_application(Application('informed-decision','Other','someone'))
+  with self.assertRaises(m.Refused):self.run_op()
+  self.assertEqual(self.counts(),[1,0,0])
+ def test_unavailable_or_unexpected_ordinary_allow_cannot_trigger_exception(self):
+  for auth in [Unavailable(),Allow()]:
+   with self.assertRaises(m.Refused):self.run_op(auth=auth)
+   self.assertEqual(self.counts(),[0,0,0])
+ def test_expired_or_extra_or_wrong_source_grant_refuses(self):
+  for g in [{**self.g,'expires_at':'2026-01-01T00:00:00Z'},{**self.g,'extra':'grant'},{**self.g,'executor_sha256':'c'*64}]:
+   with self.assertRaises(m.Refused):self.run_op(grant=g)
+  self.assertEqual(self.counts(),[0,0,0])
+ def test_created_only_rollback_preserves_history_and_deadletters_unsent_event(self):
+  self.run_op();result=self.run_op('rollback');self.assertEqual(result['result'],'rolled-back');self.assertEqual(self.counts(),[0,4,1]);self.assertEqual(self.store.pending_outbox(),());self.assertIsNotNone(self.store.outbox_history()[0].dead_lettered_at)
+  with self.assertRaises(m.Refused):self.run_op('rollback')
+ def test_claimed_event_stops_rollback_without_deleting_entities(self):
+  self.run_op();event=self.store.outbox_history()[0]
+  with self.store.transaction():self.store.save_outbox(replace(event,claimed_by='other-worker',claimed_at=datetime.now(timezone.utc)))
+  with self.assertRaises(m.Refused):self.run_op('rollback')
+  self.assertEqual(self.counts(),[2,3,1])
+ def test_changed_created_record_stops_rollback(self):
+  self.run_op()
+  with self.store.transaction():self.store.save_application(replace(m.APP,owner='new-owner'))
+  with self.assertRaises(m.Refused):self.run_op('rollback')
+  self.assertEqual(self.counts(),[2,3,1])
+ def test_transaction_failure_after_first_row_leaves_no_partial_data_or_grant(self):
+  with self.conn.cursor() as c:c.execute("CREATE FUNCTION reject_binding() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF NEW.record_type = 'application_bindings' THEN RAISE EXCEPTION 'test interruption'; END IF; RETURN NEW; END $$");c.execute('CREATE TRIGGER deny_binding BEFORE INSERT ON user_engine_records FOR EACH ROW EXECUTE FUNCTION reject_binding()')
+  self.conn.commit()
+  with self.assertRaises(Exception):self.run_op()
+  self.assertEqual(self.counts(),[0,0,0])
+ def test_delivered_event_stops_recovery_and_retains_entities(self):
+  self.run_op();event=self.store.outbox_history()[0]
+  with self.store.transaction():self.store.save_outbox(replace(event,delivered_at=datetime.now(timezone.utc)))
+  with self.assertRaises(m.Refused):self.run_op('rollback')
+  self.assertEqual(self.counts(),[2,3,1])
+ def test_recovery_requires_original_consumed_grant(self):
+  self.run_op()
+  with self.assertRaises(m.Refused):self.run_op('rollback',grant={**self.g,'operator_id':'different-maintenance-operator'})
+  self.assertEqual(self.counts(),[2,3,1])
+ def test_native_clock_refuses_expired_grant_even_with_stale_caller_clock(self):
+  g={**self.g,'expires_at':(datetime.now(timezone.utc)-timedelta(seconds=1)).isoformat()}
+  with self.assertRaises(m.Refused):m.execute(self.conn,Deny(),g,expected_grant_sha256=m.digest(g),actual_executor_sha256='b'*64,mode='apply',now=datetime.now(timezone.utc)-timedelta(minutes=1))
+  self.assertEqual(self.counts(),[0,0,0])
+ def test_delivery_worker_lock_blocks_recovery_without_partial_delete(self):
+  self.run_op();event=self.store.outbox_history()[0]
+  other=psycopg.connect('host=127.0.0.1 user=postgres dbname=postgres')
+  try:
+   with other.cursor() as c:c.execute('SELECT pg_advisory_lock(hashtextextended(%s,0))',('user-engine:tenant-lifecycle:outbox-event:'+event.event_id,))
+   other.commit()
+   self.conn.close();self.conn=psycopg.connect('host=127.0.0.1 user=postgres dbname=postgres',options='-c lock_timeout=100ms')
+   with self.assertRaises(psycopg.errors.LockNotAvailable):self.run_op('rollback')
+   self.assertEqual(self.counts(),[2,3,1])
+  finally:other.close()
+ def test_later_dependent_record_stops_created_only_recovery(self):
+  self.run_op()
+  with self.conn.cursor() as c:c.execute("INSERT INTO user_engine_records (record_type,record_key,application_id,payload) VALUES ('fixture_dependencies','fixture-key','informed-decision','{}'::jsonb)")
+  self.conn.commit()
+  with self.assertRaises(m.Refused):self.run_op('rollback')
+  self.assertEqual(self.counts(),[3,3,1])
+if __name__=='__main__':unittest.main()

tests/test_catalogue_bootstrap_owner_negatives.py

@@ -0,0 +1,52 @@
+"""Independent source-only negatives; no database/runtime/approval claims."""
+import importlib.util
+from pathlib import Path
+from dataclasses import replace
+from datetime import datetime, timedelta, timezone
+import unittest
+from user_engine.domain import AuthorizationRequest
+spec = importlib.util.spec_from_file_location("review_catalogue_executor", Path(__file__).resolve().parents[1] / "tools/catalogue_bootstrap.py")
+m = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(m)
+NOW = datetime(2026, 10, 11, 12, tzinfo=timezone.utc)
+
+def grant():
+    return {"schema": m.SCHEMA, "proposal_id": m.PROPOSAL,
+            "approval_id": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
+            "memo_id": "infd-20261011-user-catalogue-bootstrap-r1", "memo_version": 1,
+            "approval_receipt_sha256": "a" * 64, "executor_sha256": "b" * 64,
+            "records_sha256": m.digest(m.records()), "expires_at": "2026-10-12T12:00:00Z",
+            "operator_id": "fixture-maintenance-service", "authority": "founder-approved-one-time-administrative-exception"}
+
+class OwnerNegatives(unittest.TestCase):
+    def validate(self, value):
+        return m.validate_grant(value, expected_grant_sha256=m.digest(value), actual_executor_sha256="b"*64, now=NOW)
+
+    def test_json_boolean_is_not_an_authenticated_memo_revision(self):
+        with self.assertRaises(m.Refused): self.validate({**grant(), "memo_version": True})
+
+    def test_other_memo_or_added_scope_cannot_expand_fixed_exception(self):
+        for value in ({**grant(), "memo_id": "another-approved-memo"}, {**grant(), "allowed_apps": ["other"]}):
+            with self.subTest(value=value), self.assertRaises(m.Refused): self.validate(value)
+
+    def test_deadline_hard_cap_and_no_timezone_are_refused(self):
+        for expires in ("2026-10-13T00:00:00Z", "2026-10-12T12:00:00"):
+            with self.subTest(expires=expires), self.assertRaises(m.Refused): self.validate({**grant(), "expires_at": expires})
+
+    def test_exact_request_adapter_cannot_switch_app_tenant_or_identity(self):
+        actor=m._actor(grant())
+        request=AuthorizationRequest(actor, "user-engine:application", "informed-decision", "application.register", "tenant:platform", "fixture:correlation", "informed-decision")
+        changes=[{"resource_id":"other"}, {"application_id":"other"}, {"tenant":"tenant:other"}, {"actor":replace(actor,subject="different-maintenance")}, {"action":"application.delete"}, {"correlation_id":"other:correlation"}]
+        for fields in changes:
+            adapter=m._ExceptionAuthorization(actor,"fixture:correlation",datetime.now(timezone.utc)+timedelta(minutes=1))
+            with self.subTest(fields=fields), self.assertRaises(m.Refused): adapter.check(replace(request,**fields))
+
+    def test_exception_is_single_request_with_no_pdp_decision_id(self):
+        actor=m._actor(grant());adapter=m._ExceptionAuthorization(actor,"fixture:correlation",datetime.now(timezone.utc)+timedelta(minutes=1))
+        request=AuthorizationRequest(actor,"user-engine:application","informed-decision","application.register","tenant:platform","fixture:correlation","informed-decision")
+        decision=adapter.check(request)
+        self.assertTrue(decision.allowed);self.assertIsNone(decision.decision_id)
+        self.assertEqual(decision.binding,"supervised-maintenance")
+        with self.assertRaises(m.Refused): adapter.check(request)
+
+if __name__ == "__main__": unittest.main()