"""Native accounting / observer regression tests; all balances are synthetic.""" import asyncio from datetime import datetime, timezone from decimal import Decimal as D import json import hashlib from pathlib import Path import sqlite3 import sys import tempfile import unittest sys.path.insert(0,str(Path(__file__).resolve().parents[1])) import aiohttp import gauntlet_v2 as g import audit_viewer as v from audit_dashboard import Observer def observation(cash='500',qty='0',price='100'): cash,qty,price=map(D,(cash,qty,price)) return {'at':g.now(),'cash':cash,'quantity':qty,'equity':cash+qty*price,'gross_asset_value':qty*price} class AccountingTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory();self.root=Path(self.temp.name) g.initialize(self.root,'alpaca','SPY');self.cfg=g.configuration(self.root) self.j=g.Journal(self.root,self.cfg,'synthetic') def tearDown(self):self.j.close();self.temp.cleanup() def test_opening_and_buy_sell_trial_balance(self): self.j.observe('PREFLIGHT',observation()) cid=self.j.intent('buy',D('1'),D('100'),'2026-10') self.j.receipt(cid,{'broker_id':'B1','status':'filled','quantity':D('1'),'price':D('100'), 'fee':D('.1'),'fee_basis':'synthetic_authoritative','at':g.now()}) self.j.observe('MARKET_TICK',observation('399.9','1','110')) cid=self.j.intent('sell',D('1'),D('110'),emergency=True) self.j.receipt(cid,{'broker_id':'S1','status':'filled','quantity':D('1'),'price':D('110'), 'fee':D('.1'),'fee_basis':'synthetic_authoritative','at':g.now()}) self.j.observe('LIQUIDATION_CONFIRMED',observation('509.8')) proof=self.j.verify() self.assertEqual(D(proof['trial_balance']['Assets:Cash']),D('509.8')) self.assertEqual(D(proof['trial_balance']['Assets:Inventory']),0) self.assertEqual(D(proof['trial_balance']['Income:RealizedPnL']),D('-9.8')) self.assertEqual(D(proof['trial_balance']['Assets:ValuationAdjustment']),0) self.assertGreater(proof['postings'],10) def test_immutable_audit_tables(self): self.j.observe('PREFLIGHT',observation()) for table in ('events','journal_lines'): for operation in ('DELETE','UPDATE'): sql=f'DELETE FROM {table}' if operation=='DELETE' else f'UPDATE {table} SET '+('body=body' if table=='events' else 'debit=debit') with self.subTest(table=table,operation=operation),self.assertRaises(sqlite3.IntegrityError): with self.j.db:self.j.db.execute(sql) def test_suspense_lesson_and_mandatory_halt(self): self.j.observe('PREFLIGHT',observation()) self.j.observe('RISK_SAMPLE',observation('501')) self.assertTrue(self.j.halted.is_set()) self.assertTrue(self.j.get('accounting_unresolved')) self.assertEqual(self.j.lesson_count(),1) proof=self.j.verify() self.assertEqual(D(proof['trial_balance']['Equity:UnclassifiedSuspense']),D('-1')) def test_lesson_ancestry_and_physical_rows(self): with self.j.mutex,self.j.db: self.j._lesson('synthetic.a',{'expected':'1','actual':'2'}) self.j._lesson('synthetic.a',{'expected':'1','actual':'3'}) lessons=[json.loads(r[0]) for r in self.j.db.execute('SELECT body FROM lessons ORDER BY id')] self.assertEqual(lessons[1]['parent_lesson_hash'],lessons[0]['lesson_hash']) self.assertEqual(lessons[1]['multiplier_after'],'0.9025') self.j.verify() def test_independent_verifier_detects_posting_tamper(self): self.j.observe('PREFLIGHT',observation()) with self.j.db: self.j.db.execute('DROP TRIGGER native_immutable_journal_lines_UPDATE') self.j.db.execute("UPDATE journal_lines SET debit='501' WHERE seq=0") with self.assertRaises(g.Fault):self.j.verify() def test_independent_verifier_detects_state_tamper(self): self.j.observe('PREFLIGHT',observation()) with self.j.db:self.j.set('book',{'cash':'999','cost':'0','value':'0','quantity':'0'}) with self.assertRaises(g.Fault):self.j.verify() def test_pending_receipt_defers_reconciliation(self): self.j.observe('PREFLIGHT',observation()) self.j.intent('buy',D('1'),D('100')) self.j.observe('RISK_SAMPLE',observation('400','1')) self.assertFalse(self.j.get('accounting_unresolved',False)) self.assertEqual(self.j.get('book')['cash'],'500') self.j.verify() def test_diagnostic_is_exclusive_and_read_only(self): self.j.observe('PREFLIGHT',observation()) output=self.root/'report.html' proof=v.diagnostic(self.root/'native-live.sqlite3',output,{'synthetic':True}) self.assertTrue(proof['verified']);self.assertIn('pre-entry diagnostic',output.read_text()) with self.assertRaises(FileExistsError):v.diagnostic(self.root/'native-live.sqlite3',output,{}) self.assertEqual(self.j.db.execute('SELECT count(*) FROM events').fetchone()[0],1) def test_milestone_selects_first_crossing(self): self.j.observe('PREFLIGHT',observation('500')) self.j.observe('RISK_SAMPLE',observation('1000')) crossing=self.j.db.execute("SELECT max(id) FROM events WHERE kind='RISK_SAMPLE'").fetchone()[0] self.j.observe('RISK_SAMPLE',observation('1100')) data=v.snapshot(self.root/'native-live.sqlite3',milestone='1000') self.assertEqual(data['selected']['id'],crossing) def test_date_before_history_is_empty(self): self.j.observe('PREFLIGHT',observation()) data=v.snapshot(self.root/'native-live.sqlite3',date='1900-01-01') self.assertIsNone(data['selected']);self.assertEqual(data['events'],[]) def test_historical_halt_is_not_current_halt(self): self.j.observe('PREFLIGHT',observation()) before=self.j.db.execute('SELECT max(id) FROM events').fetchone()[0] self.j.halt('later') self.assertIsNone(v.snapshot(self.root/'native-live.sqlite3',event_id=before)['halt']) self.assertEqual(v.snapshot(self.root/'native-live.sqlite3')['halt']['reason'],'later') def test_rc1_upgrade_preserves_legacy_risk_lessons_and_pending_state(self): self.j.close() path=self.root/'native-live.sqlite3' path.unlink() db=sqlite3.connect(path) db.executescript("""CREATE TABLE state(k TEXT PRIMARY KEY,v TEXT NOT NULL); CREATE TABLE events(id INTEGER PRIMARY KEY,at TEXT NOT NULL,kind TEXT NOT NULL,body TEXT NOT NULL,parent TEXT NOT NULL,hash TEXT NOT NULL); CREATE TABLE intents(client_id TEXT PRIMARY KEY,body TEXT NOT NULL,status TEXT NOT NULL,broker_id TEXT); CREATE TABLE receipts(broker_id TEXT PRIMARY KEY,body TEXT NOT NULL); CREATE TABLE lessons(id INTEGER PRIMARY KEY,event_id INTEGER,body TEXT NOT NULL);""") binding=hashlib.sha256(g.encode({'config':self.cfg,'key_fingerprint':'synthetic'}).encode()).hexdigest() states={'binding':binding,'peak':'1000','ratchet':'500','quantity':'0','asset_cost':'0', 'halt':{'at':g.now(),'reason':'legacy_operator_stop'}} parent='0'*64 lesson={'failure_path':'legacy.fill','details':{},'cumulative_count':1, 'multiplier_before':'1','multiplier_after':'0.95'} event_bodies=[('PREFLIGHT',{'snapshot':g.fixed(observation('1000')),'peak':'1000','milestone_floor':'500'}),('LESSON',lesson)] with db: for key,value in states.items():db.execute('INSERT INTO state VALUES(?,?)',(key,g.encode(value))) for kind,body in event_bodies: at=g.now();body=g.encode(body) digest=hashlib.sha256(g.encode([at,kind,body,parent]).encode()).hexdigest() db.execute('INSERT INTO events(at,kind,body,parent,hash) VALUES(?,?,?,?,?)',(at,kind,body,parent,digest));parent=digest db.execute('INSERT INTO lessons(event_id,body) VALUES(?,?)',(2,g.encode(lesson))) db.execute('INSERT INTO intents VALUES(?,?,?,NULL)',('legacy-pending',g.encode({'side':'buy','quantity':'1','reference':'100','month':'2026-10','symbol':'SPY','emergency':False}),'pending')) db.close() self.j=g.Journal(self.root,self.cfg,'synthetic') self.assertTrue(self.j.halted.is_set()) self.assertEqual(self.j.get('halt')['reason'],'legacy_operator_stop') self.assertEqual(self.j.get('peak'),'1000');self.assertEqual(self.j.get('ratchet'),'500') self.assertEqual(self.j.lesson_count(),1) self.assertEqual(self.j.db.execute('SELECT count(*) FROM events').fetchone()[0],2) self.assertEqual(self.j.db.execute("SELECT count(*) FROM intents WHERE status='pending'").fetchone()[0],1) self.assertEqual(self.j.verify()['tip_hash'],parent) def test_observer_connections_are_query_only(self): self.j.observe('PREFLIGHT',observation()) db=v.open_readonly(self.root/'native-live.sqlite3') try: with self.assertRaises(sqlite3.OperationalError):db.execute('DELETE FROM state') finally:db.close() class DashboardTests(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.temp=tempfile.TemporaryDirectory();self.root=Path(self.temp.name) g.initialize(self.root,'alpaca','SPY');self.cfg=g.configuration(self.root) self.j=g.Journal(self.root,self.cfg,'synthetic') self.j.observe('PREFLIGHT',observation()) self.observer=await Observer(self.root,0).start();self.client=aiohttp.ClientSession() async def asyncTearDown(self): await self.client.close();await self.observer.close();self.j.close();self.temp.cleanup() async def test_http_snapshot_and_security_headers(self): async with self.client.get(self.observer.endpoint+'/api/window') as r: self.assertEqual(r.status,200);self.assertIn("default-src 'none'",r.headers['Content-Security-Policy']) body=await r.json();self.assertEqual(body['selected']['snapshot']['cash'],'500') async with self.client.post(self.observer.endpoint+'/api/window') as r:self.assertEqual(r.status,405) async def test_host_and_origin_checks(self): async with self.client.get(self.observer.endpoint+'/api/window',headers={'Host':'evil.invalid'}) as r:self.assertEqual(r.status,403) async with self.client.get(self.observer.endpoint+'/api/window',headers={'Origin':'https://evil.invalid'}) as r:self.assertEqual(r.status,403) async def test_socket_attestation_and_cursor(self): self.assertFalse(self.observer.attested(self.root)) async with self.client.ws_connect(self.observer.endpoint+'/ws') as ws: await ws.send_json({'type':'heartbeat'});await asyncio.sleep(.02) self.assertTrue(self.observer.attested(self.root)) data=await ws.receive_json(timeout=2) self.assertEqual(data['database'],'native-live.sqlite3') self.assertEqual(data['stream'][0]['id'],1) self.j.append('SYNTHETIC_CHANGE',{'value':'1'}) data=await ws.receive_json(timeout=2) self.assertEqual(data['stream'][-1]['kind'],'SYNTHETIC_CHANGE') self.assertFalse(self.observer.attested(self.root/'other')) async def test_observer_cannot_place_orders(self): for path in ('/order','/run','/api/order','/auth'): async with self.client.get(self.observer.endpoint+path) as r:self.assertEqual(r.status,404) async def test_ui_contains_required_views_and_no_external_assets(self): async with self.client.get(self.observer.endpoint+'/') as r: text=await r.text() for label in ('Transaction ledger','Lesson memory','Positions & context','type="range"','datetime-local'): self.assertIn(label,text) self.assertNotIn('https://',text) async def test_missing_observer_blocks_diagnostics(self): class Gate: generation=1 def require(self):pass self.j.gate=Gate() with self.assertRaises(g.Fault):await g.prepare_diagnostics(self.j)