"""Isolated synthetic DataMine protocol cases, never actual ledger/provider observations.""" import contextlib,gzip,hashlib,io,json,tempfile,unittest import subprocess from pathlib import Path from unittest.mock import patch from datamine_adapter import FIELDS,SCHEMA,HOST,parse_timesales,manifest_entry,inspect_saved,api_read,NoRedirect from infinbot_secret_audit import audit,forbidden_value from infinbot_common import Rejected def row(seq='1',t='2025-03-12T23:10:25.833993999Z'): return ['2025-03-12',t,seq,'MES','MESH5','752819','FUT','202503','','','TRD','5600.25','5600.25','2','505','B','','H','XCME'] def raw(rows):return ('\n'.join(','.join(r) for r in rows)+'\n').encode() def listing(size,fid='20250312-TS_XCME_MES_FUT_0'): return {'data':[{'dataset_code':'TS_XCME_MES_FUT_0','period_date':'20250312','files':[ {'file_id':fid,'file_name':'synthetic-datamine-case.csv.gz','size':size, 'api_download_link':'https://'+HOST+'/cme/api/v2/download?fid='+fid}]}], 'paging':{'next':'','previous':''}} class DataMineTests(unittest.TestCase): def test_v2_decimal_and_nanoseconds_preserved(self): r=parse_timesales(raw([row()]),'MES') self.assertEqual(r['latest_source_timestamp'],'2025-03-12T23:10:25.833993999Z') self.assertEqual(r['trade_records'],1);self.assertFalse(r['live_stream_verified']) def test_sequence_order_not_timestamp_order(self): r=parse_timesales(raw([row(),row('2','2025-03-12T23:10:24.833993999Z')]),'MES') self.assertEqual(r['rows_validated'],2) def test_decreasing_sequence_rejected(self): with self.assertRaises(Rejected):parse_timesales(raw([row('2'),row('1')]),'MES') def test_duplicate_sequence_rejected(self): with self.assertRaises(Rejected):parse_timesales(raw([row(),row()]),'MES') def test_wrong_product_never_guessed(self): with self.assertRaises(Rejected):parse_timesales(raw([row()]),'ES') def test_option_cannot_become_futures(self): r=row();r[6]='OPT' with self.assertRaises(Rejected):parse_timesales(raw([r]),'MES') def test_nonfinite_price_rejected(self): r=row();r[12]='NaN' with self.assertRaises(Rejected):parse_timesales(raw([r]),'MES') def test_volume_row_not_reported_as_trade(self): r=row();r[10]='VOL';r[11]=r[12]=r[13]='' result=parse_timesales(raw([r]),'MES');self.assertEqual(result['trade_records'],0) self.assertEqual(result['volume_records'],1) def test_unknown_row_schema_and_empty_rejected(self): for value in (b'',raw([row()[:-1]]),b'price,time,symbol\n1,now,MES\n'): with self.assertRaises(Rejected):parse_timesales(value,'MES') def test_explicit_header(self): self.assertEqual(parse_timesales(raw([list(FIELDS),row()]),'MES',header=True)['rows_validated'],1) with self.assertRaises(Rejected):parse_timesales(raw([list(reversed(FIELDS)),row()]),'MES',header=True) def test_saved_gzip_and_metadata_match_without_live_claim(self): with tempfile.TemporaryDirectory() as tmp: p=Path(tmp);data=gzip.compress(raw([row()]));f=p/'synthetic-datamine-case.csv.gz';f.write_bytes(data) m=p/'list.json';m.write_text(json.dumps(listing(len(data)))) config={'provider':'CME DataMine','schema':SCHEMA,'list_file':str(m),'data_file':str(f), 'file_id':'20250312-TS_XCME_MES_FUT_0','product':'MES'} before=f.read_bytes();r=inspect_saved(config) self.assertEqual(r['source_sha256'],hashlib.sha256(before).hexdigest()) self.assertFalse(r['current_entitlement_verified']);self.assertFalse(r['network_used']) self.assertEqual(f.read_bytes(),before) def test_manifest_rejects_secret_redirect_route(self): m=listing(100);m['data'][0]['files'][0]['api_download_link']='https://example.invalid/steal' with self.assertRaises(Rejected):manifest_entry(m,'20250312-TS_XCME_MES_FUT_0') def test_non_datamine_api_cannot_load_credentials(self): with patch('datamine_adapter.private_read') as read: with self.assertRaises(Rejected):api_read('https://example.invalid/cme/api/v2/download','missing') read.assert_not_called() def test_redirects_fail_closed(self): with self.assertRaises(Rejected):NoRedirect().redirect_request(None,None,None,None,None,None) def test_contract_expiry_mismatch_rejected(self): r=row();r[7]='202506' with self.assertRaises(Rejected):parse_timesales(raw([r]),'MES') def test_parent_filename_and_duplicate_manifest_rejected(self): m=listing(100);m['data'][0]['files'][0]['file_name']='..' with self.assertRaises(Rejected):manifest_entry(m,'20250312-TS_XCME_MES_FUT_0') m=listing(100);m['data'][0]['files']*=2 with self.assertRaises(Rejected):manifest_entry(m,'20250312-TS_XCME_MES_FUT_0') class SecretAuditTests(unittest.TestCase): def test_placeholder_in_any_environment_value_detected_without_echo(self): secret=forbidden_value().decode();result=audit(environment={'UNEXPECTED_NAME':'prefix'+secret}) self.assertEqual(result['status'],'DISCLOSED_PLACEHOLDER_DETECTED');self.assertNotIn(secret,json.dumps(result)) def test_missing_external_token_unverified(self): self.assertEqual(audit(environment={})['status'],'SIGNING_CONFIGURATION_UNVERIFIED') def test_private_external_token_loaded_no_system_wide_claim(self): with tempfile.TemporaryDirectory() as tmp: f=Path(tmp)/'external-token';token='synthetic-unit-test-only-0123456789abcdef';f.write_text(token);f.chmod(0o600) r=audit(str(f),environment={}) self.assertEqual(r['status'],'PROCESS_PLACEHOLDER_ABSENT_EXTERNAL_TOKEN_LOADED') self.assertNotIn(token,json.dumps(r));self.assertFalse(r['production_signing_consumer_verified']) self.assertEqual(r['other_processes_and_system_vaults'],'NOT_INSPECTED') def test_file_placeholder_and_public_file_rejected(self): with tempfile.TemporaryDirectory() as tmp: f=Path(tmp)/'token';f.write_bytes(forbidden_value());f.chmod(0o600) self.assertEqual(audit(str(f),environment={})['status'],'PRIVATE_SIGNING_TOKEN_UNAVAILABLE_OR_UNSAFE') f.write_text('synthetic-unit-test-only-0123456789abcdef');f.chmod(0o644) self.assertEqual(audit(str(f),environment={})['status'],'PRIVATE_SIGNING_TOKEN_UNAVAILABLE_OR_UNSAFE') def test_keychain_adapter_loads_privately_without_echo(self): token=b'synthetic-unit-test-only-0123456789abcdef' completed=subprocess.CompletedProcess([],0,stdout=token+b'\n',stderr=b'') with patch('infinbot_secret_audit.sys.platform','darwin'),patch('infinbot_secret_audit.subprocess.run',return_value=completed) as run: r=audit(keychain_service='SYNTHETIC_TEST_SERVICE',environment={}) self.assertEqual(r['external_source'],'MACOS_KEYCHAIN');self.assertTrue(r['external_token_loaded']) self.assertNotIn(token.decode(),json.dumps(r));self.assertFalse(r['production_signing_consumer_verified']) self.assertEqual(run.call_args.args[0],['/usr/bin/security','find-generic-password','-s','SYNTHETIC_TEST_SERVICE','-w']) def test_non_mac_keychain_fails_closed(self): with patch('infinbot_secret_audit.sys.platform','linux'),patch('infinbot_secret_audit.subprocess.run') as run: self.assertFalse(audit(keychain_service='test',environment={})['external_token_loaded']);run.assert_not_called() if __name__=='__main__':unittest.main()