"""All synthetic records are confined to TemporaryDirectory; no actual ledger touched.""" import asyncio,contextlib,hashlib,io,json,os,sqlite3,subprocess,sys,tempfile,unittest from datetime import datetime,timezone,timedelta,date from pathlib import Path from unittest.mock import patch ROOT=Path(__file__).resolve().parents[1];sys.path.insert(0,str(ROOT)) from infinbot_common import Rejected,number,strict_json,private_read,private_dir from infinbot_setup import configure from ledger_reporting import read_snapshot from risk_state import RiskState from worker_controls import WorkerControls from consigliere_server import ConsigliereWebSocketServer from infinbot_gateway_check import rehearse from feed_probe import reconcile,crypto_stream,futures_status,reconcile_balances from generated.verify_ledger_json import ProductionLedgerJsonAuditor from generated.worker_performance_ranker import MultiWorkerPerformanceRanker from rebalance_engine import QuantRiskRebalanceEngine,load_risk_limits from infinbot_backup import backup from infinbot_acceptance import execute from infinbot_schedule import merge,BEGIN,END from score_validator import initialize_score_ledger,CumulativeScoreGatekeeper from score_validator_v4 import RollingEliteGatekeeper from telemetry_notifier import ConsigliereTelemetryNotifier from licensing_controls import LicenseControl,Principal,AccessDenied from infinbot_observer_window import renew def fixture(path): with sqlite3.connect(path) as c: c.executescript('''CREATE TABLE accounts(id INTEGER PRIMARY KEY,cash TEXT,initial_equity TEXT,session TEXT,session_equity TEXT,peak_equity TEXT,halted INTEGER,started_at TEXT); CREATE TABLE positions(id INTEGER PRIMARY KEY,symbol TEXT,quantity TEXT,mark TEXT,quote_at TEXT); CREATE TABLE orders(id INTEGER PRIMARY KEY,execution_tag TEXT); CREATE TABLE dispositions(id INTEGER PRIMARY KEY,sell_order_id INTEGER,symbol TEXT,quantity TEXT,acquired_at TEXT,sold_at TEXT,proceeds TEXT,basis TEXT,economic_pnl TEXT); CREATE TABLE simulated_orders(order_id INTEGER PRIMARY KEY,execution_tag TEXT,asset_class TEXT,equity_symbol TEXT); CREATE TABLE form_8949_ledger(order_id INTEGER PRIMARY KEY,equity_symbol TEXT,date_acquired TEXT,date_sold_or_disposed TEXT,gross_proceeds TEXT,cost_basis TEXT,net_gain_or_loss TEXT); CREATE TABLE sandbox_core_metadata(value TEXT);INSERT INTO sandbox_core_metadata VALUES('SYNTHETIC_TEST_ONLY');''') c.execute('INSERT INTO accounts VALUES(1,?,?,?,?,?,?,?)',('1000','1000','2020-01-02','1000','1000',0,'2020-01-01T00:00:00Z')) rows=[(1,'crypto-fixture','CRYPTO','BTC/USD','200'),(2,'futures-fixture','FUTURES','/MES','100'),(3,'futures-fixture','FUTURES','/MES','-110')] for id,tag,asset,symbol,pnl in rows: proceeds=str(1000+int(pnl));c.execute('INSERT INTO orders VALUES(?,?)',(id,tag)) c.execute('INSERT INTO simulated_orders VALUES(?,?,?,?)',(id,tag,asset,symbol)) c.execute('INSERT INTO dispositions VALUES(?,?,?,?,?,?,?,?,?)',(id,id,symbol,'1','2020-01-01T00:00:00Z','2020-01-02T00:00:00Z',proceeds,'1000',pnl)) c.execute('INSERT INTO form_8949_ledger VALUES(?,?,?,?,?,?,?)',(id,symbol,'2020-01-01','2020-01-02',proceeds,'1000',pnl)) class HostTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory(prefix='infinbot-isolated-tests-');self.root=Path(self.temp.name) self.db=self.root/'synthetic-ledger.db';fixture(self.db) self.export=self.root/'export.json';self.export.write_text('[{"equity_symbol":"TEST_FIXTURE","currency":"USD","transaction_id":"test-1","date_acquired":"2020-01-01","date_sold_or_disposed":"2020-01-02","gross_proceeds":"1200","cost_basis":"1000","net_gain_or_loss":"200"}]') self.config_path=configure(self.root,self.db,self.export,self.root/'missing-private-paper.env','crypto-fixture') self.config=json.loads(self.config_path.read_text()) def tearDown(self):self.temp.cleanup() def test_setup_preserves_current_ledger_and_secrets(self): before=self.db.read_bytes();key=Path(self.config['gateway_token_file']).read_bytes() configure(self.root,self.db,self.export,self.root/'missing-private-paper.env','crypto-fixture') self.assertEqual(self.db.read_bytes(),before);self.assertEqual(Path(self.config['gateway_token_file']).read_bytes(),key) def test_missing_current_ledger_not_created(self): missing=self.root/'missing.db' with self.assertRaises(Rejected):read_snapshot(missing) self.assertFalse(missing.exists()) def test_float_boolean_nonfinite_rejected(self): for value in (1.0,True,'NaN','Infinity'): with self.assertRaises(Rejected):number(value) def test_duplicate_json_keys_rejected(self): with self.assertRaises(Rejected):strict_json('{"x":1,"x":2}') def test_snapshot_and_ranker_exact_recorded_math(self): snap=read_snapshot(self.db);self.assertEqual(snap['equity'],'1000');self.assertEqual(snap['data_origin'],'SYNTHETIC_FIXTURE') rank=MultiWorkerPerformanceRanker(self.db) try: r=rank.execute_worker_ranking_audit('futures-fixture');self.assertEqual(r['portfolio_net_pnl'],'-10') self.assertEqual(r['scoreboard'][0]['asset_class'],'FUTURES');self.assertFalse(r['parameters_updated']) finally:rank.close() def test_export_requires_explicit_currency(self): rows=json.loads(self.export.read_text());del rows[0]['currency'];self.export.write_text(json.dumps(rows)) self.assertEqual(ProductionLedgerJsonAuditor(self.export).validate()['status'],'FAIL') def test_json_cli_both_aliases_preserve_export(self): before=self.export.read_bytes() for flag in ('--file','--filepath'): p=subprocess.run([sys.executable,str(ROOT/'generated/verify_ledger_json.py'),flag,str(self.export)],capture_output=True,text=True) self.assertEqual(p.returncode,0,p.stdout+p.stderr) self.assertEqual(self.export.read_bytes(),before) def test_freeze_persists_after_restart(self): RiskState(self.config['risk_state']).freeze('TEST_OPERATOR_FREEZE') self.assertEqual(RiskState(self.config['risk_state']).status()['reason'],'TEST_OPERATOR_FREEZE') def test_drawdown_breaker_never_claims_liquidation(self): r=RiskState(self.config['risk_state']).observe('1000','969.816',feed_fresh=True,reconciled=True) self.assertEqual(r['reason'],'DRAWDOWN_BREAKER_TRIPPED');self.assertFalse(r['liquidation_verified']) def test_stale_feed_and_reconciliation_freeze(self): r=RiskState(self.config['risk_state']).observe('1000','1000',feed_fresh=False,reconciled=False) self.assertEqual(r['reason'],'FEED_OR_RECONCILIATION_UNVERIFIED') def test_deadline_cannot_be_extended_by_worker_restart(self): with sqlite3.connect(self.config['risk_state']) as c:c.execute('UPDATE control SET deadline=?',((datetime.now(timezone.utc)-timedelta(seconds=1)).isoformat(),)) controls=WorkerControls(state_path=self.config['risk_state']) with self.assertRaises(TimeoutError):controls.entry_allocation('200') self.assertEqual(RiskState(self.config['risk_state']).status()['reason'],'COUNTDOWN_EXPIRED') def test_owner_review_can_renew_only_observation_deadline(self): state=RiskState(self.config['risk_state']);state.freeze('PRESERVE_THIS_FREEZE_REASON') result=renew(self.config,'SYNTHETIC_OWNER_REVIEW_ONLY') self.assertTrue(result['entries_frozen']);self.assertEqual(result['reason'],'PRESERVE_THIS_FREEZE_REASON') self.assertFalse(result['live_execution_enabled']) def test_both_actual_worker_processes_zero_allocation(self): before=self.db.read_bytes() for name in ('bot-worker-crypto','bot-worker-futures'): p=subprocess.run([sys.executable,str(ROOT/'runtime/worker.py'),'--worker',name,'--db-path',str(self.db),'--state-path',self.config['risk_state'],'--once'],capture_output=True,text=True) self.assertEqual(p.returncode,0,p.stderr);self.assertEqual(json.loads(p.stdout)['allocated_units'],'0.0000') self.assertEqual(self.db.read_bytes(),before) def test_policy_output_never_enables_entries(self): output=self.root/'private/policy.json';r=QuantRiskRebalanceEngine(str(self.db),str(output)).execute_dynamic_rebalance() self.assertFalse(r['worker_allocations']['Phase1-Crypto-Sandbox']['enabled']) with self.assertRaises(Rejected):load_risk_limits(output,self.db,'Phase1-Crypto-Sandbox') def test_policy_cannot_overwrite_ledger(self): before=self.db.read_bytes() with self.assertRaises(Rejected):QuantRiskRebalanceEngine(str(self.db),str(self.db)).execute_dynamic_rebalance() self.assertEqual(self.db.read_bytes(),before) def test_gateway_real_loopback_authentication(self): r=asyncio.run(rehearse(self.config));self.assertTrue(r['invalid_token_rejected']);self.assertTrue(r['valid_owner_read_completed']) def test_gateway_remote_plaintext_rejected(self): with self.assertRaises(Rejected):ConsigliereWebSocketServer(token_file=self.config['gateway_token_file'],risk_state=self.config['risk_state'],db_filepath=self.db,host='0.0.0.0') def test_gateway_never_echoes_processed_for_unsupported_commands(self): app=ConsigliereWebSocketServer(token_file=self.config['gateway_token_file'],risk_state=self.config['risk_state'],db_filepath=self.db) self.assertEqual(app.command({'action':'BUY'})['status'],'COMMAND_REJECTED_UNSUPPORTED') self.assertFalse(app.command({'action':'freeze_entries'})['broker_liquidation_verified']) def test_position_reconciliation_mismatch_blocks(self): self.assertEqual(reconcile([],[{'symbol':'BTCUSD','asset_class':'crypto','qty':'1'}])['status'],'POSITION_RECONCILIATION_MISMATCH') self.assertEqual(reconcile([],[])['status'],'POSITION_RECONCILIATION_MATCH') def test_futures_never_inferred_from_account_auth(self): futures=futures_status() self.assertIsNone(futures['fresh_messages_observed']);self.assertIsNone(futures['adapter_enabled']) self.assertEqual(futures['status'],'EXISTING_FUTURES_CONNECTION_NOT_VERIFIED_HERE') self.assertFalse(futures['existing_configuration_changed']) def test_actual_fixture_encrypted_backup_restore_preserves_source(self): before=self.db.read_bytes();r=backup(self.config) self.assertEqual(r['status'],'SELECTED_LEDGER_BACKUP_AND_ENCRYPTED_RESTORE_VERIFIED') self.assertFalse(r['offsite_copy_verified']);self.assertEqual(self.db.read_bytes(),before) def test_offline_acceptance_stays_blocked_and_hides_secrets(self): before=self.db.read_bytes();r=execute(self.config,network=False,include_backup=False) self.assertEqual(r['production_release'],'BLOCKED');self.assertFalse(r['live_execution_enabled']) text=json.dumps(r);self.assertNotIn(Path(self.config['gateway_token_file']).read_text().strip(),text) self.assertEqual(self.db.read_bytes(),before) self.assertIn('current_ledger',r['blocking_checks']) def test_offline_cli_retains_host_configuration_on_disk(self): before=self.config_path.read_bytes() p=subprocess.run([sys.executable,str(ROOT/'infinbot_acceptance.py'),'--config',str(self.config_path),'--offline','--skip-backup'],capture_output=True,text=True) self.assertEqual(p.returncode,1) summary=json.loads(p.stdout);receipt=json.loads(Path(summary['receipt_path']).read_text()) self.assertTrue(receipt['host_config_unchanged_during_run']) self.assertEqual(self.config_path.read_bytes(),before) self.assertFalse(receipt['live_execution_enabled']) def test_cash_mismatch_never_changes_ledger_or_uses_default_funding(self): result=reconcile_balances(read_snapshot(self.db),{'currency':'USD','cash':'999','equity':'999'}) self.assertEqual(result['status'],'BALANCE_RECONCILIATION_MISMATCH');self.assertFalse(result['ledger_adjusted']) def test_cron_merge_preserves_unrelated_and_is_idempotent(self): new=BEGIN+'\n55 23 * * 5 /bin/true\n'+END+'\n' value=merge('0 8 * * * /bin/true\n',new) self.assertEqual(merge(value,new),value);self.assertIn('0 8 * * * /bin/true',value) def test_unmanaged_archive_job_blocks_duplication(self): with self.assertRaises(Rejected):merge('59 23 * * 0 python3 archive_tax_reports.py\n',BEGIN+'\n'+END+'\n') def test_notification_adapter_never_confirms_device_by_api_receipt(self): file=self.root/'private/notification-test.json';file.write_text('{"app_token":"SYNTHETIC_TOKEN","user_token":"SYNTHETIC_TOKEN"}');file.chmod(0o600) notifier=ConsigliereTelemetryNotifier(file,transport=lambda *args:'API_ACCEPTED') self.assertTrue(notifier.dispatch_message('SYNTHETIC_TEST_ONLY'));self.assertEqual(notifier.last_result['device_delivery'],'UNVERIFIED') def test_new_notifier_brand(self): file=self.root/'private/notification-brand.json';file.write_text('{"app_token":"SYNTHETIC_TOKEN","user_token":"SYNTHETIC_TOKEN"}');file.chmod(0o600) messages=[];notifier=ConsigliereTelemetryNotifier(file,transport=lambda message,*args:messages.append(message) or 'API_ACCEPTED') notifier.dispatch_emergency_triage_sms('TEST_PROJECT','TEST_VERTICAL','TEST_DIRECTIVE') self.assertTrue(messages[0].startswith('InFinBot:')) def test_completed_acceptance_receipt_never_contains_broker_secrets(self): from verify_live_feeds_standalone import authenticate_account file=Path(self.config['credentials_file']);file.write_text('ALPACA_PAPER_KEY=PKabcdefghijklmnopqrstuv\nALPACA_PAPER_SECRET=abcdefghijklmnopqrstuv1234567890\n');file.chmod(0o600) with patch('infinbot_acceptance.authenticate_account',return_value={'status':'ACCOUNT_READ_AUTHENTICATED'}), \ patch('infinbot_acceptance.read_json',return_value={}),patch('infinbot_acceptance.broker_positions',return_value=[]), \ patch('infinbot_acceptance.crypto_stream',return_value={'status':'CRYPTO_STREAM_UNVERIFIED'}): result=execute(self.config,network=True,include_backup=False) text=json.dumps(result) self.assertNotIn('PKabcdefghijklmnopqrstuv',text);self.assertNotIn('abcdefghijklmnopqrstuv1234567890',text) self.assertEqual(result['production_release'],'BLOCKED') class ScoreTests(unittest.TestCase): def test_recorded_label_without_complete_metrics_cannot_qualify(self): with tempfile.TemporaryDirectory() as folder: db=Path(folder)/'synthetic-scores.db';sqlite3.connect(db).close();initialize_score_ledger(db) gate=RollingEliteGatekeeper(db,execution_tag='SYNTHETIC_TEST_SCOPE') try: for i in range(100): day=(date(2020,1,1)+timedelta(days=i)).isoformat() gate.conn.execute('INSERT INTO expected_scoring_sessions VALUES(?,?,?)',('SYNTHETIC_TEST_SCOPE',i+1,day));gate.conn.commit() gate.process_and_log_daily_score(day,'0','0',False,input_provenance='RECORDED_SESSION_LEDGER',source_reference='ISOLATED_SYNTHETIC_TEST') r=gate.execute_rolling_window_audit();self.assertEqual(r['ELIGIBLE_RECORDED_SESSIONS'],0);self.assertFalse(r['LIVE_CLEARANCE_GRANTED']) with sqlite3.connect(db) as c: for i in range(100): day=(date(2020,1,1)+timedelta(days=i)).isoformat() c.execute('INSERT INTO daily_session_metrics VALUES(?,?,?,?,?,?,?,?,?,?)',('SYNTHETIC_TEST_SCOPE',day,'0','0',0,0,2,1,'RECORDED_SESSION_LEDGER','ISOLATED_SYNTHETIC_TEST')) r=gate.execute_rolling_window_audit();self.assertEqual(r['DECISION_TOKEN'],'RESEARCH_SCORE_THRESHOLD_MET_NO_LIVE_CLEARANCE');self.assertFalse(r['LIVE_CLEARANCE_GRANTED']) finally:gate.close() class LicensingTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory(prefix='infinbot-isolated-licensing-') self.db=Path(self.temp.name)/'synthetic-commercial-registry.db' self.now=datetime(2026,10,13,tzinfo=timezone.utc) self.admin=Principal('SYNTHETIC_ADMIN',0,'SYNTHETIC_SESSION',admin=True) self.control=LicenseControl(self.db,'SYNTHETIC TERMS ONLY; NOT AN LGC CONTRACT','SYNTHETIC_TEST_VERSION', pepper=b'SYNTHETIC_TEST_PEPPER_00000000000',approved=True, approval_reference='SYNTHETIC_TEST_ONLY_NOT_LGC_APPROVAL',clock=lambda:self.now, tenant_symbols={1:['BTC/USD'],2:['BTC/USD']}) self.control.install() for tenant in (1,2): self.control.create_pending(self.admin,'SYNTHETIC_TENANT_'+str(tenant),'2027-01-01T00:00:00Z',1500) p=Principal('SYNTHETIC_REP_'+str(tenant),tenant,'SYNTHETIC_SESSION_'+str(tenant),representative=True) self.control.accept(p,self.control.version,self.control.digest,True) self.control.record_payment_entitlement(self.admin,tenant,'2027-01-01T00:00:00Z','SYNTHETIC_PAYMENT_ONLY') self.key=self.control.activate(self.admin,1) self.frame={'symbol':'BTC/USD','side':'BUY','units':'1','price':'100','timestamp':self.now.isoformat(),'sequence':1} def tearDown(self):self.temp.cleanup() def test_unapproved_terms_cannot_activate(self): self.control.approved=False with self.assertRaises(AccessDenied):self.control.activate(self.admin,2) def test_cross_tenant_key_rejected(self): self.control.activate(self.admin,2) with self.assertRaises(AccessDenied):self.control.guarded_send(2,self.key,self.frame,lambda data:None) def test_suspended_tenant_stops_next_frame(self): sent=[];self.control.guarded_send(1,self.key,self.frame,sent.append) self.control.suspend(self.admin,1,'SYNTHETIC_SUSPENSION_TEST') with self.assertRaises(AccessDenied):self.control.guarded_send(1,self.key,self.frame,sent.append) self.assertEqual(len(sent),1) def test_expired_payment_blocks(self): self.control.clock=lambda:datetime(2028,1,1,tzinfo=timezone.utc) with self.assertRaises(AccessDenied):self.control.guarded_send(1,self.key,self.frame,lambda data:None) def test_key_brand_and_hash_storage(self): self.assertTrue(self.key.startswith('infinbot_')) with sqlite3.connect(self.db) as c: values=c.execute('SELECT key_digest FROM commercial_api_credentials').fetchall() self.assertNotIn(self.key,[x[0] for x in values]) class StreamTests(unittest.TestCase): def probe(self,events): class Socket: async def send(self,message):pass async def recv(self):return json.dumps(events.pop(0)) class Session: async def __aenter__(self):return Socket() async def __aexit__(self,*args):pass settings={'ALPACA_PAPER_KEY':'PKabcdefghijklmnopqrstuv','ALPACA_PAPER_SECRET':'abcdefghijklmnopqrstuv1234567890'} return asyncio.run(crypto_stream(settings,connect=lambda *a,**kw:Session(),timeout=1)) def events(self,at=None): at=at or datetime.now(timezone.utc).isoformat() return [[{'T':'success','msg':'authenticated'}],[{'T':'subscription','quotes':['BTC/USD','ETH/USD']}], [{'T':'q','S':'BTC/USD','bp':'100','ap':'101','t':at},{'T':'q','S':'ETH/USD','bp':'100','ap':'101','t':at}]] def test_authenticated_subscription_and_observed_messages_required(self): r=self.probe(self.events());self.assertEqual(r['status'],'FRESH_CRYPTO_STREAM_OBSERVED') self.assertTrue(r['subscription_acknowledged']);self.assertFalse(r['persistent_worker_subscription_verified']) def test_stale_messages_do_not_qualify(self): r=self.probe(self.events((datetime.now(timezone.utc)-timedelta(minutes=5)).isoformat())) self.assertEqual(r['status'],'CRYPTO_STREAM_UNVERIFIED') def test_subscription_failure_and_disconnect_do_not_qualify(self): for events in ([{'T':'success','msg':'authenticated'}],[{'T':'error','code':409}]): r=self.probe([events]);self.assertEqual(r['status'],'CRYPTO_STREAM_UNVERIFIED') if __name__=='__main__':unittest.main()