import hashlib import importlib.util import json import os from pathlib import Path import sqlite3 import subprocess import sys import tempfile import unittest from datetime import datetime,timezone from unittest.mock import Mock,patch ROOT=Path(__file__).parent sys.path.insert(0,str(ROOT/'generated')) from verify_ledger_json import ProductionLedgerJsonAuditor from portfolio_performance_ranker import PortfolioPerformanceRanker from telemetry_notifier import ConsigliereTelemetryNotifier,push class Tests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name).resolve();self.ledger=self.root/'ledger.json' def tearDown(self):self.tmp.cleanup() def row(self,**changes): r={'transaction_id':'unique-1','equity_symbol':'TEST','date_acquired':'2026-10-01','date_sold_or_disposed':'2026-10-02','gross_proceeds':'110.00','cost_basis':'100.00','net_gain_or_loss':'10.00'};r.update(changes);return r def audit(self,rows): self.ledger.write_text(json.dumps(rows));before=self.ledger.read_bytes();r=ProductionLedgerJsonAuditor(self.ledger).validate();self.assertEqual(self.ledger.read_bytes(),before);return r def test_valid_ledger_unchanged(self):self.assertEqual(self.audit([self.row()])['status'],'PASS') def test_loss_and_breakeven_separate(self): r=self.audit([self.row(),self.row(transaction_id='2',gross_proceeds='90.00',net_gain_or_loss='-10.00'),self.row(transaction_id='3',gross_proceeds='100.00',net_gain_or_loss='0.00')]);self.assertEqual(r['dispositions']['losses'],1);self.assertEqual(r['dispositions']['breakeven'],1) def test_float_and_subprecision_rejected(self): for value in (110.0,'110.0000001',True,'NaN','Infinity'): with self.subTest(value=value):self.assertEqual(self.audit([self.row(gross_proceeds=value)])['status'],'FAIL') def test_exact_six_decimal_math(self): r=self.audit([self.row(gross_proceeds='1.000001',cost_basis='1.000000',net_gain_or_loss='0.000001')]);self.assertEqual(r['status'],'PASS');self.assertEqual(r['totals']['net_gain_or_loss'],'0.000001') def test_math_mismatch(self):self.assertEqual(self.audit([self.row(net_gain_or_loss='9.00')])['status'],'FAIL') def test_invalid_and_reverse_dates(self): for changes in ({'date_sold_or_disposed':'2026-09-30'},{'date_acquired':'bad'},{'date_acquired':'2026-10-01T10:00:00','date_sold_or_disposed':'2026-10-02T10:00:00'}):self.assertEqual(self.audit([self.row(**changes)])['status'],'FAIL') def test_timezone_timestamps(self): r=self.audit([self.row(date_acquired='2026-10-01T10:00:00-04:00',date_sold_or_disposed='2026-10-01T15:00:00Z')]);self.assertEqual(r['status'],'PASS') def test_duplicate_record_and_id(self): self.assertEqual(self.audit([self.row(),self.row()])['status'],'FAIL') self.assertEqual(self.audit([self.row(),self.row(gross_proceeds='120.00',net_gain_or_loss='20.00')])['status'],'FAIL') def test_empty_and_wrong_root(self): for rows in ([],{},[1]):self.assertEqual(self.audit(rows)['status'],'FAIL') def test_adjustment_explicit(self): self.assertEqual(self.audit([self.row(adjustment_amount='2.00',adjustment_reason='source correction',net_gain_or_loss='12.00')])['status'],'PASS') self.assertEqual(self.audit([self.row(adjustment_amount='2.00',net_gain_or_loss='12.00')])['status'],'FAIL') def test_currency_and_negative_basis(self): for changes in ({'currency':'EUR'},{'cost_basis':'-1.00'}):self.assertEqual(self.audit([self.row(**changes)])['status'],'FAIL') def test_missing_file_not_created(self): self.assertEqual(ProductionLedgerJsonAuditor(self.ledger).validate()['status'],'FAIL');self.assertFalse(self.ledger.exists()) def test_parse_and_duplicate_json_keys(self): for raw in ('bad','[{"gross_proceeds":"1","gross_proceeds":"2"}]'): self.ledger.write_text(raw);self.assertEqual(ProductionLedgerJsonAuditor(self.ledger).validate()['status'],'FAIL') def test_validator_cli_never_seeds(self): r=subprocess.run([sys.executable,str(ROOT/'generated/verify_ledger_json.py'),'--file',str(self.ledger)],capture_output=True,text=True);self.assertEqual(r.returncode,2);self.assertFalse(self.ledger.exists()) def db(self): p=self.root/'portfolio.db';c=sqlite3.connect(p) c.execute('CREATE TABLE enterprise_portfolio(project_id INTEGER PRIMARY KEY,project_name TEXT,industry_vertical TEXT,lifecycle_state TEXT,supervisor_deadline TEXT,allocated_budget_cents INTEGER)') c.execute('CREATE TABLE remediation_lifecycle(remediation_id INTEGER PRIMARY KEY,associated_project TEXT,execution_status TEXT,execution_evidence_log TEXT,migration_audit_notes TEXT)') c.executemany('INSERT INTO enterprise_portfolio VALUES(?,?,?,?,?,?)',[(1,'BPA','ACCOUNTING','ACTIVE','2026-11-15',4500000),(2,'TEST_VERIFICATION_PASS_NODE','ACCOUNTING','ACTIVE','2026-12-31',100000),(3,'FASHION','FASHION','HOLD','2026-11-15T00:00:00Z',12500000)]) c.execute("INSERT INTO remediation_lifecycle VALUES(1,'BPA','DEPLOYED',NULL,NULL)");c.commit();c.close();return p def test_ranker_current_schema_readonly_and_test_exclusion(self): p=self.db();before=p.read_bytes();r=PortfolioPerformanceRanker(p) try:report=r.execute_portfolio_ranking_audit(datetime(2026,10,13,tzinfo=timezone.utc)) finally:r.close() self.assertEqual(report['business_units'],2);self.assertEqual(report['active_units'],1);self.assertEqual(report['budget_total_usd'],'170000.00') self.assertEqual(report['portfolio_summary'][0]['deadline_status'],'NEEDS_DEADLINE_REVIEW');self.assertEqual(report['portfolio_summary'][0]['evidence_status'],'MISSING_EXECUTION_EVIDENCE');self.assertEqual(p.read_bytes(),before) def test_ranker_rejects_legacy_shape(self): p=self.root/'old.db';c=sqlite3.connect(p);c.execute('CREATE TABLE enterprise_portfolio(allocated_budget REAL)');c.close();r=PortfolioPerformanceRanker(p) try: with self.assertRaises(ValueError):r.execute_portfolio_ranking_audit() finally:r.close() def test_ranker_missing_database_not_created(self): with self.assertRaises(ValueError):PortfolioPerformanceRanker(self.root/'missing.db') self.assertFalse((self.root/'missing.db').exists()) def creds(self): p=self.root/'creds.json';p.write_text(json.dumps({'app_token':'test-only','user_token':'test-only'}));p.chmod(0o600);return p def test_notifier_missing_tokens_no_call(self): fn=Mock();n=ConsigliereTelemetryNotifier(self.root/'missing',transport=fn);self.assertFalse(n.dispatch_message('test'));fn.assert_not_called() def test_notifier_acceptance_not_delivery(self): n=ConsigliereTelemetryNotifier(self.creds(),transport=lambda m,c:'API_ACCEPTED');self.assertTrue(n.dispatch_message('test'));self.assertEqual(n.last_result['device_delivery'],'UNVERIFIED') def test_notifier_failures_and_unknown(self): for state in ('RETRY','UNKNOWN','PERMANENT_REJECTION'): n=ConsigliereTelemetryNotifier(self.creds(),transport=lambda m,c:state);self.assertFalse(n.dispatch_message('test'));self.assertEqual(n.last_result['status'],state) def test_notifier_private_credentials_required(self): p=self.creds();p.chmod(0o644) with self.assertRaises(ValueError):ConsigliereTelemetryNotifier(p) def test_legacy_env_credentials(self): p=self.root/'credentials.env';p.write_text("export NOTIFIER_APP_TOKEN='test-only'\nNOTIFIER_USER_TOKEN=\"test-only\"\n");p.chmod(0o600) n=ConsigliereTelemetryNotifier(p,transport=lambda m,c:'API_ACCEPTED');self.assertTrue(n.dispatch_message('test')) def test_notifier_cli_defaults_dry(self): r=subprocess.run([sys.executable,str(ROOT/'generated/telemetry_notifier.py')],capture_output=True,text=True);self.assertEqual(r.returncode,0);self.assertFalse(json.loads(r.stdout)['sent']) def test_push_mock_protocol(self): 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('telemetry_notifier.urllib.request.build_opener',return_value=opener): self.assertEqual(push('test',{'app_token':'test-only','user_token':'test-only'}),'API_ACCEPTED') response.read.return_value=b'bad';self.assertEqual(push('test',{'app_token':'test-only','user_token':'test-only'}),'UNKNOWN') if __name__=='__main__':unittest.main(verbosity=2)