204 lines
8.0 KiB
Python
204 lines
8.0 KiB
Python
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_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
|
|
)
|
|
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()
|