"""Isolated fixtures and loopback HTTP only; never actual host/client/payment data.""" import io,json,os,sqlite3,subprocess,sys,tempfile,threading,unittest,urllib.request,urllib.error from datetime import datetime,timezone from decimal import Decimal from pathlib import Path from unittest.mock import patch from wsgiref.simple_server import make_server,WSGIRequestHandler from licensing_controls import LicenseControl,Principal,AccessDenied,ClickwrapApp from infinbot_schedule import install_schedule,BEGIN from instrument_units import mes_contract_values,displayed_price,option_contract_value from market_sample_checks import alpaca_bars,cme_options from infinbot_common import Rejected from telemetry_notifier import ConsigliereTelemetryNotifier ROOT=Path(__file__).resolve().parents[1] class CommercialTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup) self.db=Path(self.temp.name)/'isolated-commercial.db';self.now=datetime(2026,10,13,tzinfo=timezone.utc) self.control=LicenseControl(self.db,'ISOLATED TEST TERMS, NOT AN LGC CONTRACT','TEST-V1', pepper=b'ISOLATED_TEST_PEPPER_000000000000',approved=True,approval_reference='TEST_ONLY', clock=lambda:self.now,tenant_symbols={1:['BTC/USD'],2:['BTC/USD']}) self.control.install();self.admin=Principal('TEST_ADMIN',0,'TEST_ADMIN_SESSION',admin=True) for n in (1,2):self.control.create_pending(self.admin,'TEST_TENANT_'+str(n),'2027-01-01T00:00:00Z',1500) self.people={str(n):Principal('TEST_REP_'+str(n),n,'TEST_SESSION_'+str(n),representative=True) for n in (1,2)} def authenticate(environ): cookie=environ.get('HTTP_COOKIE','') if cookie not in ('session=1','session=2'):raise AccessDenied('Unauthenticated isolated session') return self.people[cookie[-1]] self.app=ClickwrapApp(self.control,authenticate=authenticate,origin='https://client.example.invalid') def request(self,path,*,data=None,session='1',origin='https://client.example.invalid',csrf=None): env={'PATH_INFO':path,'REQUEST_METHOD':'POST' if data is not None else 'GET','HTTP_COOKIE':'session='+session,'HTTP_ORIGIN':origin} if data is not None: raw=json.dumps(data).encode();env.update(CONTENT_TYPE='application/json',CONTENT_LENGTH=str(len(raw)),**{'wsgi.input':io.BytesIO(raw),'HTTP_X_CSRF_TOKEN':csrf or self.app.csrf(self.people[session])}) info={} result=b''.join(self.app(env,lambda status,headers:info.update(status=status,headers=dict(headers)))) return info,result def accept(self,n='1'): return self.request('/api/accept',session=n,data={'version':self.control.version,'sha256':self.control.digest,'accepted':True}) def paid(self,n=1):self.control.record_payment_entitlement(self.admin,n,'2027-01-01T00:00:00Z','TEST_PAYMENT_ONLY') def frame(self):return {'symbol':'BTC/USD','side':'BUY','units':'1','price':'100','timestamp':self.now.isoformat(),'sequence':1} def test_no_key_before_click_even_when_paid(self): self.paid() with self.assertRaises(AccessDenied):self.control.activate(self.admin,1) with sqlite3.connect(self.db) as c:self.assertEqual(c.execute('SELECT COUNT(*) FROM commercial_api_credentials').fetchone()[0],0) def test_acceptance_alone_does_not_activate(self): info,raw=self.accept();self.assertEqual(info['status'],'200 OK');self.assertFalse(json.loads(raw)['license_activated']) with self.assertRaises(AccessDenied):self.control.activate(self.admin,1) def test_accepted_paid_tenant_activates(self): self.accept();self.paid();key=self.control.activate(self.admin,1) self.assertTrue(key.startswith('infinbot_'));sent=[];self.control.guarded_send(1,key,self.frame(),sent.append);self.assertEqual(len(sent),1) def test_false_checkbox_wrong_hash_and_old_version_block(self): for changed in ({'accepted':False},{'sha256':'0'*64},{'version':'OLD'}): data={'version':self.control.version,'sha256':self.control.digest,'accepted':True};data.update(changed) info,_=self.request('/api/accept',data=data);self.assertEqual(info['status'],'403 Forbidden') def test_request_cannot_select_other_tenant(self): data={'version':self.control.version,'sha256':self.control.digest,'accepted':True,'tenant_id':2} info,_=self.request('/api/accept',data=data);self.assertEqual(info['status'],'400 Bad Request') self.accept() with sqlite3.connect(self.db) as c:self.assertEqual(c.execute('SELECT DISTINCT tenant_id FROM license_acceptances').fetchall(),[(1,)]) def test_csrf_and_cross_origin_block(self): data={'version':self.control.version,'sha256':self.control.digest,'accepted':True} for kwargs in ({'origin':'https://attacker.example.invalid'},{'csrf':'bad-token'},{'csrf':self.app.csrf(self.people['2'])}): info,_=self.request('/api/accept',data=data,**kwargs);self.assertEqual(info['status'],'403 Forbidden') def test_unapproved_contract_blocks_acceptance(self): self.control.approved=False;info,_=self.accept();self.assertEqual(info['status'],'403 Forbidden') def test_direct_sql_suspension_stops_next_frame(self): self.accept();self.paid();key=self.control.activate(self.admin,1);sent=[] self.control.guarded_send(1,key,self.frame(),sent.append) with sqlite3.connect(self.db) as c:c.execute("UPDATE commercial_tenant_licenses SET license_status='SUSPENDED' WHERE tenant_id=1") with self.assertRaises(AccessDenied):self.control.guarded_send(1,key,self.frame(),sent.append) self.assertEqual(len(sent),1) def test_payment_expiry_stops_next_frame(self): self.accept();self.paid();key=self.control.activate(self.admin,1) with sqlite3.connect(self.db) as c:c.execute("UPDATE license_entitlements SET paid_until='2026-10-14T00:00:00Z' WHERE tenant_id=1") self.control.clock=lambda:datetime(2026,10,15,tzinfo=timezone.utc) frame=self.frame();frame['timestamp']=self.control.clock().isoformat() with self.assertRaisesRegex(AccessDenied,'payment entitlement elapsed'):self.control.guarded_send(1,key,frame,lambda data:None) def test_idle_stream_exits_after_suspension(self): self.accept();self.paid();key=self.control.activate(self.admin,1);outcome=[];started=threading.Event() def frames():started.set();return None def work(): try:self.control.stream(1,key,frames,lambda data:None,stop=threading.Event(),heartbeat_seconds=.01) except AccessDenied:outcome.append('STOPPED') thread=threading.Thread(target=work,daemon=True);thread.start();self.assertTrue(started.wait(1)) self.control.suspend(self.admin,1,'TEST_INCIDENT');thread.join(1) self.assertFalse(thread.is_alive());self.assertEqual(outcome,['STOPPED']) def test_real_http_client_page_and_unauthenticated_denial(self): class Quiet(WSGIRequestHandler): def log_message(self,*args):pass server=make_server('127.0.0.1',0,self.app,handler_class=Quiet) thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start() try: url='http://127.0.0.1:'+str(server.server_port)+'/client' with urllib.request.urlopen(urllib.request.Request(url,headers={'Cookie':'session=1'}),timeout=2) as response: raw=response.read();self.assertIn(b'I Accept the Terms of Service',raw) self.assertIn(b'type="checkbox" disabled required',raw);self.assertIn("sha256-",response.headers['Content-Security-Policy']) with self.assertRaises(urllib.error.HTTPError) as caught:urllib.request.urlopen(url,timeout=2) self.assertEqual(caught.exception.code,403);caught.exception.close() finally:server.shutdown();server.server_close();thread.join(2) class OperationTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup) self.root=Path(self.temp.name);self.config=self.root/'host.json' self.config.write_text(json.dumps({'receipt_dir':str(self.root/'receipts')}));self.config.chmod(0o600) self.existing='17 4 * * * /usr/bin/true\n';self.current=self.existing;self.writes=[] def backend(self,cmd,**kwargs): if cmd==['crontab','-']: self.current=kwargs['input'];self.writes.append(self.current);return subprocess.CompletedProcess(cmd,0,'','') return subprocess.CompletedProcess(cmd,0,self.current,'') def install(self,backend=None):return install_schedule(self.config,run=backend or self.backend,platform='darwin',timezone_path='/usr/share/zoneinfo/America/New_York') def test_install_readback_preserves_other_jobs_and_is_idempotent(self): result=self.install();self.assertFalse(result['first_scheduled_run_verified']);self.assertIn(self.existing,self.current) once=self.current;self.install();self.assertEqual(self.current,once);self.assertEqual(self.current.count(BEGIN),1) def test_duplicate_legacy_job_blocks_without_write(self): self.current+='59 23 * * 0 python3 archive_tax_reports.py\n' with self.assertRaises(Rejected):self.install() self.assertEqual(self.writes,[]) def test_failed_readback_restores_original_jobs(self): reads=0 def backend(cmd,**kwargs): nonlocal reads if cmd==['crontab','-l']: reads+=1 if reads==2:return subprocess.CompletedProcess(cmd,0,'UNEXPECTED_READBACK\n','') return self.backend(cmd,**kwargs) with self.assertRaisesRegex(Rejected,'previous jobs restored'):self.install(backend) self.assertEqual(self.current,self.existing) def test_wrong_timezone_and_platform_do_not_install(self): for platform,zone in (('linux','/America/New_York'),('darwin','/UTC')): with self.assertRaises(Rejected):install_schedule(self.config,run=self.backend,platform=platform,timezone_path=zone) self.assertEqual(self.writes,[]) def test_missing_notifier_credentials_no_network_attempt(self): calls=[];n=ConsigliereTelemetryNotifier(self.root/'missing.json',transport=lambda *args:calls.append(args)) self.assertFalse(n.dispatch_message('TEST'));self.assertEqual(calls,[]) def test_notifier_timeout_never_retries_or_confirms_device(self): path=self.root/'notification.json';path.write_text('{"app_token":"TEST_ONLY","user_token":"TEST_ONLY"}');path.chmod(0o600) calls=[] def transport(*args):calls.append(1);raise TimeoutError() n=ConsigliereTelemetryNotifier(path,transport=transport);self.assertFalse(n.dispatch_message('TEST')) self.assertEqual(calls,[1]);self.assertEqual(n.last_result['status'],'UNKNOWN');self.assertEqual(n.last_result['device_delivery'],'UNVERIFIED') def test_operations_cannot_reuse_old_phone_receipt(self): import contextlib from infinbot_operations import main receipt=self.root/'old-receipt.json';receipt.write_text(json.dumps({'device_delivery':'OWNER_CONFIRMED','api_status':'API_ACCEPTED','observed_at':'2020-01-01T00:00:00+00:00'}));receipt.chmod(0o600) self.config.write_text(json.dumps({'receipt_dir':str(self.root/'receipts'),'notification_receipt_file':str(receipt)})) output=io.StringIO() with patch('infinbot_operations.sys.platform','darwin'),patch('infinbot_operations.subprocess.run',return_value=subprocess.CompletedProcess([],0)),contextlib.redirect_stdout(output): self.assertEqual(main(['--config',str(self.config),'--send-test']),1) result=json.loads(output.getvalue());self.assertEqual(result['device_delivery'],'UNVERIFIED');self.assertEqual(result['status'],'HOST_OPERATIONS_UNVERIFIED') class UnitAndSchemaTests(unittest.TestCase): def test_mes_published_units_exact(self): r=mes_contract_values('2750','0.25',1);self.assertEqual(r['notional_usd'],'13750');self.assertEqual(r['price_change_value_usd'],'1.25') self.assertEqual(mes_contract_values('2750','-1',2)['price_change_value_usd'],'-10') def test_clearing_factor_explicit_and_options_multiplier_explicit(self): self.assertEqual(displayed_price('563100','0.01'),Decimal('5631.00')) self.assertEqual(option_contract_value('0.21',2,'100'),Decimal('42.00')) self.assertEqual(option_contract_value('0.21',2,'50'),Decimal('21.00')) def test_float_nonfinite_fractional_contracts_and_tick_mismatch_rejected(self): for params in ((2750.0,'0.25',1),('NaN','0.25',1),('2750','0.1',1),('2750','0.25',True),('2750','0.25',1.5)): with self.assertRaises(Rejected):mes_contract_values(*params) def test_downloaded_provider_stock_and_options_fixtures(self): for name,asset in (('alpaca-stock-provider-fixture.json','STOCKS'),('alpaca-option-provider-fixture.json','OPTIONS')): result=alpaca_bars((ROOT/'samples'/name).read_bytes(),asset);self.assertEqual(result['rows_validated'],2) self.assertFalse(result['pagination_complete']);self.assertFalse(result['live_feed_verified']) def test_wrong_profile_and_duplicate_timestamps_block(self): raw=(ROOT/'samples/alpaca-stock-provider-fixture.json').read_bytes() with self.assertRaises(Rejected):alpaca_bars(raw,'OPTIONS') payload=json.loads(raw);payload['bars']['AAPL'][1]['t']=payload['bars']['AAPL'][0]['t'] with self.assertRaises(Rejected):alpaca_bars(json.dumps(payload),'STOCKS') if __name__=='__main__':unittest.main()