"""Isolated regression suite. Never touches /Users/bpa or sends real messages.""" import hashlib import hmac import json import os from pathlib import Path import sqlite3 import subprocess import sys import tempfile import unittest import uuid from datetime import timedelta from unittest.mock import patch from consigliere_supervisor import ConsigliereProductionCore as Core, canonical, cents, deadline, now, push class Tests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) self.db = self.root/'db.sqlite3' self.logs = self.root/'logs' self.key = os.urandom(32) self.core = None def tearDown(self): if self.core: self.core.close() self.tmp.cleanup() def open(self, **kwargs): self.core = Core(self.db,self.logs,sender_keys={'manager':self.key},**kwargs) return self.core def event(self, **overrides): payload={'security_headers':{'sender_id':'manager','event_uuid':str(uuid.uuid4()),'issued_at':now().isoformat()},'metadata':{'project_name':'TEST','vertical':'ACCOUNTING','lifecycle_state':'ACTIVE','deadline':(now()-timedelta(days=1)).isoformat(),'budget_value':'45250.75'}} payload['metadata'].update(overrides) payload['security_headers']['signature']=hmac.new(self.key,canonical(payload),hashlib.sha256).hexdigest() return payload def write(self,payload): p=self.logs/(str(uuid.uuid4())+'.json');p.write_text(json.dumps(payload));return p def queue(self): self.core.conn.execute("INSERT INTO notification_outbox(associated_project,alert_payload_text,delivery_status,queued_at) VALUES('TEST','Test message','PENDING',?)",(now().isoformat(),)) def legacy(self, *, duplicate=False, invalid=False): c=sqlite3.connect(self.db) c.execute('CREATE TABLE enterprise_portfolio(project_id INTEGER PRIMARY KEY AUTOINCREMENT,project_name TEXT UNIQUE,industry_vertical TEXT,lifecycle_state TEXT,supervisor_deadline TEXT,allocated_budget REAL,days_lagging_count INT,logged_at TEXT)') c.execute("INSERT INTO enterprise_portfolio VALUES(17,'TEST','ACCOUNTING',?, '2026-11-15',45250.75,3,'original')",('INVALID' if invalid else 'ACTIVE',)) c.execute('CREATE TABLE remediation_lifecycle(remediation_id INTEGER PRIMARY KEY,associated_project TEXT REFERENCES enterprise_portfolio(project_name),deployed_directive TEXT,execution_status TEXT,triggered_at TEXT)') c.execute("INSERT INTO remediation_lifecycle VALUES(23,'TEST','DIRECTIVE','PENDING_EXECUTION','historical')") if duplicate:c.execute("INSERT INTO remediation_lifecycle VALUES(24,'TEST','DIRECTIVE','PENDING_EXECUTION','historical')") c.commit();c.close();self.db.chmod(0o600) def test_fresh_startup_and_foreign_keys(self): c=self.open();self.assertEqual(c.conn.execute('PRAGMA foreign_keys').fetchone()[0],1) self.assertEqual(c.conn.execute('PRAGMA user_version').fetchone()[0],1) with self.assertRaises(sqlite3.IntegrityError):c.conn.execute("INSERT INTO remediation_lifecycle(associated_project,trigger_condition,deployed_directive,execution_status,triggered_at,incident_id) VALUES('missing','x','x','FAILED','x','x')") def test_monetary_boundaries(self): self.assertEqual(cents('45250.75'),4525075) self.assertEqual(cents('9000000000000000.00'),900000000000000000) self.assertEqual(cents('1.005',legacy=True),101) for value in (True,1.2,'NaN','Infinity','-1','1.001','1e3','9000000000000000.01','1'*100): with self.subTest(value=value),self.assertRaises(ValueError):cents(value) def test_deadlines(self): self.assertEqual(deadline('2026-10-12T10:00:00-04:00').hour,14) for value in ('2026-10-12','2026-10-12T10:00:00','bad'): with self.assertRaises(ValueError):deadline(value) def test_ingestion_and_archival(self): c=self.open();p=self.write(self.event());self.assertEqual(c.autonomous_directory_sync_scan(),['INGESTED']) self.assertFalse(p.exists());self.assertEqual(c.conn.execute('SELECT allocated_budget_cents FROM enterprise_portfolio').fetchone()[0],4525075) self.assertEqual(len(list((self.logs/'processed').glob('*'))),1) def test_replay_and_conflicting_event(self): c=self.open();e=self.event();self.write(e);c.autonomous_directory_sync_scan() self.write(e);self.assertEqual(c.autonomous_directory_sync_scan(),['DUPLICATE']) e['metadata']['budget_value']='1.00';e['security_headers']['signature']=hmac.new(self.key,canonical(e),hashlib.sha256).hexdigest() self.write(e);self.assertEqual(c.autonomous_directory_sync_scan(),['CONFLICT']) self.assertEqual(c.conn.execute('SELECT allocated_budget_cents FROM enterprise_portfolio').fetchone()[0],4525075) def test_invalid_signature(self): c=self.open();e=self.event();e['metadata']['project_name']='FORGED';self.write(e) self.assertEqual(c.autonomous_directory_sync_scan(),['REJECTED']);self.assertEqual(c.conn.execute('SELECT COUNT(*) FROM enterprise_portfolio').fetchone()[0],0) def test_unknown_sender(self): c=self.open();e=self.event();e['security_headers']['sender_id']='unknown';self.write(e) self.assertEqual(c.autonomous_directory_sync_scan(),['REJECTED']) def test_expired_event(self): c=self.open();e=self.event();e['security_headers']['issued_at']=(now()-timedelta(days=2)).isoformat() e['security_headers']['signature']=hmac.new(self.key,canonical(e),hashlib.sha256).hexdigest();self.write(e) self.assertEqual(c.autonomous_directory_sync_scan(),['REJECTED']) def test_duplicate_json_keys(self): c=self.open();(self.logs/'bad.json').write_text('{"metadata":{},"metadata":{}}') self.assertEqual(c.autonomous_directory_sync_scan(),['REJECTED']) def test_invalid_money_rolls_back_event(self): c=self.open();self.write(self.event(budget_value='1.001')) self.assertEqual(c.autonomous_directory_sync_scan(),['REJECTED']) self.assertEqual(c.conn.execute('SELECT COUNT(*) FROM ingestion_journal').fetchone()[0],0) def test_archive_failure_is_replay_safe(self): c=self.open();e=self.event();self.write(e) with patch.object(c,'_archive',side_effect=OSError):self.assertEqual(c.autonomous_directory_sync_scan(),['INGESTED']) self.assertEqual(c.autonomous_directory_sync_scan(),['DUPLICATE']) self.assertEqual(c.conn.execute('SELECT COUNT(*) FROM enterprise_portfolio').fetchone()[0],1) def test_symlink_input_rejected(self): c=self.open();target=self.root/'external';target.write_text('private');(self.logs/'link.json').symlink_to(target) self.assertEqual(c.autonomous_directory_sync_scan(),['REJECTED']);self.assertEqual(target.read_text(),'private') def test_runtime_lock(self): self.open() with self.assertRaises(BlockingIOError):Core(self.db,self.logs) def test_legacy_preserves_ids_and_status(self): self.legacy();c=self.open() self.assertEqual(c.conn.execute('SELECT project_id,allocated_budget_cents,logged_at FROM enterprise_portfolio').fetchone()[:],(17,4525075,'original')) r=c.conn.execute('SELECT remediation_id,execution_status,execution_evidence_log,triggered_at FROM remediation_lifecycle').fetchone() self.assertEqual(r[:],(23,'PENDING_EXECUTION',None,'historical')) self.assertEqual(c.conn.execute('PRAGMA foreign_key_check').fetchall(),[]) self.assertTrue(list(self.root.glob('*.backup-*.sqlite3'))) raw=json.loads(c.conn.execute("SELECT source_rows_json FROM migration_audit WHERE source_table='enterprise_portfolio'").fetchone()[0]);self.assertEqual(raw[0]['days_lagging_count'],3) def test_duplicate_legacy_rolls_back(self): self.legacy(duplicate=True) with self.assertRaises(sqlite3.IntegrityError):self.open() c=sqlite3.connect(self.db) self.assertIn('allocated_budget',[r[1] for r in c.execute('PRAGMA table_info(enterprise_portfolio)')]) self.assertEqual(c.execute('SELECT COUNT(*) FROM remediation_lifecycle').fetchone()[0],2);c.close() def test_invalid_legacy_rolls_back(self): self.legacy(invalid=True) with self.assertRaises(sqlite3.IntegrityError):self.open() c=sqlite3.connect(self.db);self.assertEqual(c.execute('SELECT lifecycle_state FROM enterprise_portfolio').fetchone()[0],'INVALID');c.close() def test_unknown_tables_preserved(self): c=sqlite3.connect(self.db);c.execute('CREATE TABLE custom(data TEXT)');c.close();self.db.chmod(0o600) with self.assertRaises(ValueError):self.open() c=sqlite3.connect(self.db);self.assertTrue(c.execute("SELECT name FROM sqlite_master WHERE name='custom'").fetchone());c.close() def test_custom_views_not_dropped(self): self.legacy();c=sqlite3.connect(self.db);c.execute('CREATE VIEW custom AS SELECT * FROM enterprise_portfolio');c.close() with self.assertRaises(ValueError):self.open() def test_restart_keeps_data(self): c=self.open();self.write(self.event());c.autonomous_directory_sync_scan();c.close();c=self.open() self.assertEqual(c.conn.execute('SELECT COUNT(*) FROM enterprise_portfolio').fetchone()[0],1) def test_rules_preserved_and_disabled(self): self.legacy();c=sqlite3.connect(self.db);c.execute('CREATE TABLE autonomous_resolution_rules(rule_id INTEGER PRIMARY KEY,target_vertical TEXT,trigger_condition TEXT,pre_approved_action TEXT)');c.execute("INSERT INTO autonomous_resolution_rules VALUES(8,'ACCOUNTING','DEADLINE_LAG','CUSTOM')");c.commit();c.close() c=self.open();r=c.conn.execute('SELECT rule_id,pre_approved_action,enabled FROM autonomous_resolution_rules').fetchone();self.assertEqual(r[:],(8,'CUSTOM',0)) def test_action_dry_run_and_missing_handler(self): c=self.open();self.write(self.event());c.autonomous_directory_sync_scan() c.conn.execute("INSERT INTO autonomous_resolution_rules(target_vertical,trigger_condition,pre_approved_action,enabled) VALUES('ACCOUNTING','DEADLINE_LAG','TEST_ACTION',1)") self.assertEqual(c.execute_autonomous_triage_loop(),['DRY_RUN']) self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['BLOCKED']) self.assertEqual(c.conn.execute('SELECT COUNT(*) FROM remediation_lifecycle').fetchone()[0],0) def handler(self,script): folder=self.root/'actions';folder.mkdir(mode=0o700) p=folder/'test.sh';p.write_text(script);p.chmod(0o700) return {'TEST_ACTION':{'path':str(p),'sha256':hashlib.sha256(p.read_bytes()).hexdigest()}} def prepare_action(self,handlers): c=self.open(handlers=handlers);self.write(self.event());c.autonomous_directory_sync_scan() c.conn.execute("INSERT INTO autonomous_resolution_rules(target_vertical,trigger_condition,pre_approved_action,enabled) VALUES('ACCOUNTING','DEADLINE_LAG','TEST_ACTION',1)") return c def test_real_local_action_and_duplicate_block(self): c=self.prepare_action(self.handler('#!/bin/sh\nexit 0\n')) self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['DEPLOYED']) self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['ALREADY_OPEN']) self.assertNotIn('secured',c.conn.execute('SELECT alert_payload_text FROM notification_outbox').fetchone()[0]) def test_handler_hash_mismatch(self): handlers=self.handler('#!/bin/sh\nexit 0\n');handlers['TEST_ACTION']['sha256']='0'*64 c=self.prepare_action(handlers);self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['BLOCKED']) def test_action_failure(self): c=self.prepare_action(self.handler('#!/bin/sh\nexit 3\n')) self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['FAILED']) self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['ALREADY_OPEN']) def test_crash_recovery_unknown_blocks_repeat(self): c=self.prepare_action(self.handler('#!/bin/sh\nexit 0\n')) c.conn.execute("INSERT INTO remediation_lifecycle(associated_project,trigger_condition,deployed_directive,execution_status,triggered_at,incident_id) VALUES('TEST','DEADLINE_LAG','TEST_ACTION','STAGED',?,'test-incident')",(now().isoformat(),));c.close() c=self.open(handlers=c.handlers) self.assertEqual(c.conn.execute('SELECT execution_status FROM remediation_lifecycle').fetchone()[0],'UNKNOWN') self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['ALREADY_OPEN']) def test_missing_notification_config_does_not_consume_attempt(self): c=self.open();self.queue();self.assertEqual(c.process_notification_retry_outbox(send=True),['NOT_CONFIGURED']) self.assertEqual(c.conn.execute('SELECT retry_count,delivery_status FROM notification_outbox').fetchone()[:],(0,'PENDING')) def test_notification_acceptance_no_resend(self): calls=[];c=self.open(notifier=lambda m:(calls.append(m) or 'API_ACCEPTED'),notification_configured=True);self.queue() self.assertEqual(c.process_notification_retry_outbox(send=True),['GATEWAY_ACCEPTED']) self.assertEqual(c.process_notification_retry_outbox(send=True),[]);self.assertEqual(len(calls),1) def test_notification_retry_backoff(self): c=self.open(notifier=lambda m:'RETRY',notification_configured=True);self.queue() self.assertEqual(c.process_notification_retry_outbox(send=True),['PENDING']) self.assertEqual(c.process_notification_retry_outbox(send=True),[]) c.conn.execute("UPDATE notification_outbox SET retry_count=4,retry_backoff_expiry=NULL") self.assertEqual(c.process_notification_retry_outbox(send=True),['FAILED']) def test_notification_ambiguous_no_resend(self): c=self.open(notifier=lambda m:'UNKNOWN',notification_configured=True);self.queue() self.assertEqual(c.process_notification_retry_outbox(send=True),['UNKNOWN']) self.assertEqual(c.process_notification_retry_outbox(send=True),[]) def test_notification_claim_crash_recovery(self): c=self.open();self.queue();c.conn.execute("UPDATE notification_outbox SET delivery_status='CLAIMED'");c.close();c=self.open() self.assertEqual(c.conn.execute('SELECT delivery_status FROM notification_outbox').fetchone()[0],'UNKNOWN') def test_hold_active_counts(self): c=self.open();self.write(self.event(lifecycle_state='HOLD'));c.autonomous_directory_sync_scan() r=c.render_executive_control_scoreboard();self.assertEqual(r['active_count'],0);self.assertEqual(r['total_units'],1) self.assertEqual(r['budget_total_usd'],'45250.75') def test_notifier_without_credentials(self): self.assertEqual(push('Test',None),'NOT_CONFIGURED') def test_cli_safe_fresh_run(self): result=subprocess.run([sys.executable,str(Path(__file__).with_name('consigliere_supervisor.py')),'--root',str(self.root)],capture_output=True,text=True) self.assertEqual(result.returncode,0,result.stderr) self.assertEqual(json.loads(result.stdout)['scoreboard']['total_units'],0) def test_out_of_order_signed_update_rejected(self): c=self.open();self.write(self.event());c.autonomous_directory_sync_scan() e=self.event(budget_value='1.00');e['security_headers']['issued_at']=(now()-timedelta(hours=1)).isoformat() e['security_headers']['signature']=hmac.new(self.key,canonical(e),hashlib.sha256).hexdigest();self.write(e) self.assertEqual(c.autonomous_directory_sync_scan(),['REJECTED']) self.assertEqual(c.conn.execute('SELECT allocated_budget_cents FROM enterprise_portfolio').fetchone()[0],4525075) def test_wal_backup_restore_contents(self): self.legacy();external=sqlite3.connect(self.db);external.execute('PRAGMA journal_mode=WAL') external.execute("UPDATE enterprise_portfolio SET allocated_budget=123.45");external.commit() c=self.open();self.assertEqual(c.conn.execute('SELECT allocated_budget_cents FROM enterprise_portfolio').fetchone()[0],12345) backup=list(self.root.glob('*.backup-*.sqlite3'))[0];restored=sqlite3.connect(backup) self.assertEqual(restored.execute('SELECT allocated_budget FROM enterprise_portfolio').fetchone()[0],123.45) self.assertEqual(restored.execute('PRAGMA integrity_check').fetchone()[0],'ok');restored.close();external.close() def test_schema_mismatch_rejected(self): c=self.open();c.conn.execute('DROP INDEX idx_open_remediations');c.close() with self.assertRaises(ValueError):self.open() def test_legacy_outbox_fields_upgraded(self): c=sqlite3.connect(self.db);c.execute("CREATE TABLE notification_outbox(notification_id INTEGER PRIMARY KEY,associated_project TEXT,alert_payload_text TEXT,retry_count INTEGER,delivery_status TEXT,queued_at TEXT)") c.execute("INSERT INTO notification_outbox VALUES(9,'TEST','old',0,'SENT_VERIFIED','old-time')");c.commit();c.close();self.db.chmod(0o600) c=self.open();r=c.conn.execute('SELECT * FROM notification_outbox').fetchone() self.assertEqual(r['delivery_status'],'LEGACY_UNVERIFIED');self.assertIsNone(r['retry_backoff_expiry']);self.assertEqual(r['notification_id'],9) def test_action_timeout_unknown(self): from unittest.mock import Mock c=self.prepare_action(self.handler('#!/bin/sh\nexit 0\n')) proc=Mock(pid=123456);proc.wait.side_effect=[subprocess.TimeoutExpired('test',10),0] with patch('consigliere_supervisor.subprocess.Popen',return_value=proc),patch('consigliere_supervisor.os.killpg'): self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['UNKNOWN']) self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['ALREADY_OPEN']) def test_notifier_protocol_responses(self): from unittest.mock import Mock creds={'app_token':'test-only','user_token':'test-only'} response=Mock(status=200);response.read.return_value=b'{"status":1}' response.__enter__=Mock(return_value=response);response.__exit__=Mock(return_value=False) opener=Mock();opener.open.return_value=response with patch('consigliere_supervisor.urllib.request.build_opener',return_value=opener): self.assertEqual(push('test',creds),'API_ACCEPTED') response.read.return_value=b'{"status":0}';self.assertEqual(push('test',creds),'PERMANENT_REJECTION') response.read.return_value=b'bad-json';self.assertEqual(push('test',creds),'UNKNOWN') opener.open.side_effect=TimeoutError();self.assertEqual(push('test',creds),'UNKNOWN') def test_handler_symlink_and_permissions(self): handlers=self.handler('#!/bin/sh\nexit 0\n');c=self.prepare_action(handlers) p=Path(handlers['TEST_ACTION']['path']);p.chmod(0o777) self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['BLOCKED']) p.chmod(0o700);link=p.parent/'link.sh';link.symlink_to(p);handlers['TEST_ACTION']['path']=str(link) self.assertEqual(c.execute_autonomous_triage_loop(execute=True),['BLOCKED']) def test_autoincrement_high_water_preserved(self): self.legacy();c=sqlite3.connect(self.db);c.execute("UPDATE sqlite_sequence SET seq=99 WHERE name='enterprise_portfolio'");c.commit();c.close() c=self.open();self.assertEqual(c.conn.execute("SELECT seq FROM sqlite_sequence WHERE name='enterprise_portfolio'").fetchone()[0],99) if __name__=='__main__': unittest.main(verbosity=2)