import unittest from unittest.mock import patch, MagicMock class TestBrighterTrades(unittest.TestCase): def setUp(self): """Set up test fixtures with mocked dependencies.""" # Create a mock socketio instance self.mock_socketio = MagicMock() # Patch the dependencies before importing BrighterTrades self.patches = [] # Patch DataCache self.data_cache_patcher = patch('src.BrighterTrades.DataCache') self.mock_data_cache_class = self.data_cache_patcher.start() self.mock_data_cache = MagicMock() self.mock_data_cache_class.return_value = self.mock_data_cache self.patches.append(self.data_cache_patcher) # Patch ExchangeInterface self.exchange_interface_patcher = patch('src.BrighterTrades.ExchangeInterface') self.mock_exchange_interface_class = self.exchange_interface_patcher.start() self.mock_exchanges = MagicMock() self.mock_exchange_interface_class.return_value = self.mock_exchanges self.patches.append(self.exchange_interface_patcher) # Patch Users self.users_patcher = patch('src.BrighterTrades.Users') self.mock_users_class = self.users_patcher.start() self.mock_users = MagicMock() self.mock_users_class.return_value = self.mock_users self.patches.append(self.users_patcher) # Patch Strategies self.strategies_patcher = patch('src.BrighterTrades.Strategies') self.mock_strategies_class = self.strategies_patcher.start() self.mock_strategies = MagicMock() self.mock_strategies_class.return_value = self.mock_strategies self.patches.append(self.strategies_patcher) # Patch Candles self.candles_patcher = patch('src.BrighterTrades.Candles') self.mock_candles_class = self.candles_patcher.start() self.mock_candles = MagicMock() self.mock_candles_class.return_value = self.mock_candles self.patches.append(self.candles_patcher) # Patch Indicators self.indicators_patcher = patch('src.BrighterTrades.Indicators') self.mock_indicators_class = self.indicators_patcher.start() self.mock_indicators = MagicMock() self.mock_indicators_class.return_value = self.mock_indicators self.patches.append(self.indicators_patcher) # Patch Trades self.trades_patcher = patch('src.BrighterTrades.Trades') self.mock_trades_class = self.trades_patcher.start() self.mock_trades = MagicMock() self.mock_trades_class.return_value = self.mock_trades self.patches.append(self.trades_patcher) # Now import and create BrighterTrades with mocked dependencies from src.BrighterTrades import BrighterTrades self.brighter_trades = BrighterTrades(self.mock_socketio) # Assign mocks to the instance for easier access in tests self.brighter_trades.users = self.mock_users self.brighter_trades.candles = self.mock_candles self.brighter_trades.exchanges = self.mock_exchanges self.brighter_trades.strategies = self.mock_strategies def tearDown(self): """Stop all patches.""" for patcher in self.patches: patcher.stop() def test_create_new_user(self): """Test creating a new user.""" self.mock_users.create_new_user.return_value = True self.mock_users.log_in_user.return_value = True result = self.brighter_trades.create_new_user('test@example.com', 'testuser', 'password') self.assertTrue(result) self.mock_users.create_new_user.assert_called_with( email='test@example.com', username='testuser', password='password' ) self.mock_users.log_in_user.assert_called_with( username='testuser', password='password' ) def test_log_user_in_out_login(self): """Test user login.""" self.mock_users.log_in_user.return_value = True result = self.brighter_trades.log_user_in_out('testuser', 'login', 'password') self.assertTrue(result) self.mock_users.log_in_user.assert_called_with( username='testuser', password='password' ) def test_log_user_in_out_logout(self): """Test user logout.""" self.mock_users.log_out_user.return_value = True result = self.brighter_trades.log_user_in_out('testuser', 'logout') self.assertTrue(result) self.mock_users.log_out_user.assert_called_with(username='testuser') def test_get_user_info_chart_view(self): """Test getting user chart view info.""" expected_chart_view = {'timeframe': '1d', 'exchange': 'binance', 'market': 'BTC/USDT'} self.mock_users.get_chart_view.return_value = expected_chart_view result = self.brighter_trades.get_user_info('testuser', 'Chart View') self.assertEqual(result, expected_chart_view) self.mock_users.get_chart_view.assert_called_with(user_name='testuser') def test_get_market_info_candle_history(self): """Test getting candle history.""" expected_candles = [ {'open': 1, 'high': 2, 'low': 0.5, 'close': 1.5, 'volume': 1000} ] self.mock_candles.get_candle_history.return_value = expected_candles chart_view = {'market': 'BTC/USDT', 'timeframe': '1d', 'exchange': 'binance'} result = self.brighter_trades.get_market_info( 'Candle History', chart_view=chart_view, num_records=10, user_name='testuser' ) self.assertEqual(result, expected_candles) self.mock_candles.get_candle_history.assert_called_with( num_records=10, symbol='BTC/USDT', interval='1d', exchange_name='binance', user_name='testuser' ) def test_received_new_trade_forwards_broker_key(self): """Manual trade requests should pass broker_key through to Trades.new_trade.""" self.mock_trades.new_trade.return_value = ('Success', 'trade-123') mock_trade = MagicMock() mock_trade.to_json.return_value = {'unique_id': 'trade-123'} self.mock_trades.get_trade_by_id.return_value = mock_trade result = self.brighter_trades.received_new_trade({ 'target': 'test_exchange', 'exchange': 'paper', 'symbol': 'BTC/USDT', 'price': 50000.0, 'side': 'LONG', 'orderType': 'MARKET', 'quantity': 0.1, 'broker_key': 'paper_margin_isolated', 'collateral': 100.0, 'leverage': 2.0 }, user_id=1) self.assertTrue(result['success']) self.assertEqual( self.mock_trades.new_trade.call_args.kwargs['broker_key'], 'paper_margin_isolated' ) def test_received_new_margin_trade_forwards_collateral_and_leverage(self): """Margin trade requests should forward collateral/leverage without requiring spot quantity semantics.""" self.mock_trades.new_trade.return_value = ('Success', 'trade-456') mock_trade = MagicMock() mock_trade.to_json.return_value = {'unique_id': 'trade-456'} self.mock_trades.get_trade_by_id.return_value = mock_trade result = self.brighter_trades.received_new_trade({ 'target': 'test_exchange', 'exchange': 'kucoin', 'symbol': 'BTC/USDT', 'price': 50000.0, 'side': 'LONG', 'orderType': 'MARKET', 'quantity': 0.0, 'broker_key': 'paper_margin_isolated', 'collateral': 100.0, 'leverage': 3.0 }, user_id=1) self.assertTrue(result['success']) self.assertEqual(self.mock_trades.new_trade.call_args.kwargs['collateral'], 100.0) self.assertEqual(self.mock_trades.new_trade.call_args.kwargs['leverage'], 3.0) def test_connect_or_config_exchange_reconnects_when_credentials_change(self): """Existing exchange should reconnect immediately when API credentials change.""" connected_cache = MagicMock() connected_cache.empty = False self.mock_data_cache.get_serialized_datacache.return_value = connected_cache old_keys = {'key': 'old_key', 'secret': 'old_secret', 'passphrase': 'old_phrase'} new_keys = {'key': 'new_key', 'secret': 'new_secret', 'passphrase': 'new_phrase'} self.mock_users.get_api_keys.return_value = old_keys self.mock_exchanges.connect_exchange.return_value = True result = self.brighter_trades.connect_or_config_exchange( user_name='testuser', exchange_name='kucoin', api_keys=new_keys ) self.assertEqual(result['status'], 'success') self.mock_exchanges.connect_exchange.assert_called_with( exchange_name='kucoin', user_name='testuser', api_keys=new_keys, testnet=False ) self.mock_users.update_api_keys.assert_called_with( api_keys=new_keys, exchange='kucoin', user_name='testuser' ) def test_connect_or_config_exchange_reconnects_when_already_connected(self): """Existing exchange should reconnect to ensure fresh ccxt client (fixes pickle corruption).""" connected_cache = MagicMock() connected_cache.empty = False self.mock_data_cache.get_serialized_datacache.return_value = connected_cache same_keys = {'key': 'same_key', 'secret': 'same_secret', 'passphrase': 'same_phrase'} self.mock_users.get_api_keys.return_value = same_keys self.mock_exchanges.connect_exchange.return_value = True result = self.brighter_trades.connect_or_config_exchange( user_name='testuser', exchange_name='kucoin', api_keys=same_keys ) # Should reconnect to fix pickle corruption issues with ccxt client self.assertEqual(result['status'], 'success') self.mock_exchanges.connect_exchange.assert_called_once() self.mock_users.update_api_keys.assert_not_called() if __name__ == '__main__': unittest.main()