user-engine/tests/test_catalogue_bootstrap.py

"""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()