import copy import json import os import sys import tempfile import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import Mock sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from verify_live_feeds import ForesixaLiveFeedVerifier, load_worker_config NOW = datetime(2026, 10, 5, 23, 0, tzinfo=timezone.utc) CONFIG = {'foresixa_version': 'test', 'ingestion_lifecycle': { 'account_mode': 'PAPER_SANDBOX', 'verified_provider_endpoint': 'https://paper-api.alpaca.markets'}, 'active_worker_streams': {'crypto_worker': {'stream_feed': 'ONLINE_STREAMING_ACTIVE', 'symbols_monitored': ['BTC/USD', 'ETH/USD']}, 'futures_worker': { 'stream_feed': 'ONLINE_STREAMING_ACTIVE', 'symbols_monitored': ['/MES'], 'official_cme_point_multiplier': 5}}} SETTINGS = {'ALPACA_PAPER_KEY': 'PKtestvalidliteral12345', 'ALPACA_PAPER_SECRET': 'testvalidliteral98765'} ACCOUNT = {'status': 'ACTIVE', 'cash': '1', 'equity': '1', 'buying_power': '1', 'trading_blocked': False} def quotes(**changes): return {'quotes': {symbol: {'bp': '100', 'ap': '101', 't': NOW.isoformat(), **changes} for symbol in ['BTC/USD', 'ETH/USD']}} class FeedTests(unittest.TestCase): def report(self, responses, config=None, settings=None): self.transport = Mock(side_effect=responses) verifier = ForesixaLiveFeedVerifier(config=copy.deepcopy(config or CONFIG), get=self.transport, clock=lambda: NOW) return json.loads(verifier.execute_exchange_stream_validation( settings=SETTINGS if settings is None else settings)) def test_fresh_quotes_are_separate_read_only_requests(self): result = self.report([ACCOUNT, quotes()]) self.assertEqual(self.transport.call_count, 2) self.assertEqual(self.transport.call_args_list[0].args[0], 'https://paper-api.alpaca.markets/v2/account') self.assertTrue(self.transport.call_args_list[1].args[0].startswith('https://data.alpaca.markets/')) self.assertIn('BTC%2FUSD', self.transport.call_args_list[1].args[0]) self.assertEqual(set(result['crypto_observations']['quotes']), {'BTC/USD', 'ETH/USD'}) self.assertEqual(result['status'], 'PARTIAL_CHECKS_PASSED_FUTURES_UNCONFIGURED') self.assertFalse(result['stream_verified']) self.assertFalse(result['live_clearance_granted']) self.assertFalse(result['trial_started']) self.assertEqual(result['broker_orders_submitted'], 0) def test_config_labels_never_verify_futures(self): result = self.report([ACCOUNT, quotes()]) self.assertEqual(result['multi_worker_feed_registry']['futures_feed_status'], 'FUTURES_FEED_UNCONFIGURED') def test_account_success_does_not_verify_missing_feed(self): result = self.report([ACCOUNT, {}]) self.assertTrue(result['network_peripheral_gates']['broker_handshake_clear']) self.assertEqual(result['network_peripheral_gates']['market_data_connectivity'], 'NOT_VERIFIED') def test_quote_can_succeed_while_account_fails(self): result = self.report([{}, quotes()]) self.assertFalse(result['network_peripheral_gates']['broker_handshake_clear']) self.assertEqual(result['network_peripheral_gates']['market_data_connectivity'], 'FRESH_CRYPTO_REST_QUOTES_VERIFIED') self.assertEqual(result['status'], 'CHECKS_NOT_VERIFIED') def test_stale_future_and_naive_timestamps_block(self): for at in (NOW-timedelta(seconds=61), NOW+timedelta(seconds=6), NOW.replace(tzinfo=None)): with self.subTest(at=at): result = self.report([ACCOUNT, quotes(t=at.isoformat())]) self.assertNotIn('crypto_observations', result) def test_invalid_prices_and_missing_symbols_block(self): bad = [quotes(bp='NaN'), quotes(ap='0'), quotes(bp='102'), quotes(bp=None), {'quotes': {'BTC/USD': quotes()['quotes']['BTC/USD']}}, {'quotes': {'BTC/USD': [], 'ETH/USD': []}}] for payload in bad: with self.subTest(payload=payload): self.assertNotIn('crypto_observations', self.report([ACCOUNT, payload])) def test_wrong_endpoint_blocks_before_network(self): for base in ('https://alpaca.markets', 'https://api.alpaca.markets', 'http://paper-api.alpaca.markets', 'https://attacker.example'): config = copy.deepcopy(CONFIG) config['ingestion_lifecycle']['verified_provider_endpoint'] = base self.assertEqual(self.report([], config)['status'], 'CONFIGURATION_BLOCKED') self.transport.assert_not_called() def test_missing_and_live_credentials_cannot_be_used_as_paper(self): for settings in ({}, {'ALPACA_LIVE_KEY': SETTINGS['ALPACA_PAPER_KEY'], 'ALPACA_LIVE_SECRET': SETTINGS['ALPACA_PAPER_SECRET']}, {**SETTINGS, 'ALPACA_BASE_URL': 'https://api.alpaca.markets'}, {**SETTINGS, 'ALPACA_PAPER_KEY': 'mockliteral123456789'}): self.assertEqual(self.report([], settings=settings)['status'], 'CONFIGURATION_BLOCKED') self.transport.assert_not_called() def test_no_credentials_or_raw_exception_text_in_output(self): for responses in ([RuntimeError(SETTINGS['ALPACA_PAPER_SECRET']), RuntimeError(SETTINGS['ALPACA_PAPER_KEY'])], [{**ACCOUNT, 'id': SETTINGS['ALPACA_PAPER_SECRET']}, quotes()]): output = json.dumps(self.report(responses)) for secret in SETTINGS.values(): self.assertNotIn(secret, output) def test_bad_config_and_multiplier_block_before_network(self): for field, value in (('symbols_monitored', ['BTC/USD', 'BTC/USD']), ('symbols_monitored', ['/MES'])): config = copy.deepcopy(CONFIG) config['active_worker_streams']['crypto_worker'][field] = value self.assertEqual(self.report([], config)['status'], 'CONFIGURATION_BLOCKED') self.transport.assert_not_called() for multiplier in (50, True, 'NaN'): config = copy.deepcopy(CONFIG) config['active_worker_streams']['futures_worker']['official_cme_point_multiplier'] = multiplier self.assertEqual(self.report([], config)['status'], 'CONFIGURATION_BLOCKED') self.transport.assert_not_called() def test_private_file_does_not_mutate_environment_or_contents(self): with tempfile.TemporaryDirectory() as directory: file = Path(directory)/'credentials.env' file.write_text('\n'.join(k+'='+v for k, v in SETTINGS.items())) file.chmod(0o600) before, environment = file.read_bytes(), dict(os.environ) verifier = ForesixaLiveFeedVerifier(config=CONFIG, get=Mock(side_effect=[ACCOUNT, quotes()]), clock=lambda: NOW) result = json.loads(verifier.execute_exchange_stream_validation(credentials_path=file)) self.assertEqual(result['status'], 'PARTIAL_CHECKS_PASSED_FUTURES_UNCONFIGURED') self.assertEqual(file.read_bytes(), before) self.assertEqual(dict(os.environ), environment) def test_unsafe_or_missing_private_file_blocks(self): with tempfile.TemporaryDirectory() as directory: file = Path(directory)/'credentials.env' transport = Mock() verifier = ForesixaLiveFeedVerifier(config=CONFIG, get=transport, clock=lambda: NOW) self.assertEqual(json.loads(verifier.execute_exchange_stream_validation(credentials_path=file))['status'], 'CONFIGURATION_BLOCKED') self.assertFalse(file.exists()) file.write_text('\n'.join(k+'='+v for k, v in SETTINGS.items())) file.chmod(0o644) self.assertEqual(json.loads(verifier.execute_exchange_stream_validation(credentials_path=file))['status'], 'CONFIGURATION_BLOCKED') transport.assert_not_called() def test_config_loader_rejects_duplicate_nonfinite_and_oversized_json(self): with tempfile.TemporaryDirectory() as directory: file = Path(directory)/'config.json' for content in ('{"x":1,"x":2}', '{"x":NaN}', ' '*16385, '[]'): file.write_text(content) with self.assertRaises(Exception): load_worker_config(file) if __name__ == '__main__': unittest.main()