567 lines
22 KiB
Python
567 lines
22 KiB
Python
"""
|
|
Tests for strategy execution (run/stop/status) flow.
|
|
|
|
These tests cover the new run_strategy, stop_strategy, and get_strategy_status
|
|
message handlers and their underlying implementation.
|
|
"""
|
|
import pytest
|
|
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
class TestStartStrategyValidation:
|
|
"""Tests for start_strategy input validation and authorization."""
|
|
|
|
@pytest.fixture
|
|
def mock_brighter_trades(self):
|
|
"""Create a mock BrighterTrades instance with required dependencies."""
|
|
with patch.dict('sys.modules', {'eventlet': MagicMock()}):
|
|
from BrighterTrades import BrighterTrades
|
|
|
|
# Create mock dependencies
|
|
mock_socketio = MagicMock()
|
|
|
|
with patch.object(BrighterTrades, '__init__', lambda x, y: None):
|
|
bt = BrighterTrades(mock_socketio)
|
|
|
|
# Set up required attributes
|
|
bt.strategies = MagicMock()
|
|
bt.strategies.active_instances = {}
|
|
bt.strategies.data_cache = MagicMock()
|
|
|
|
# Mock users dependency
|
|
bt.users = MagicMock()
|
|
bt.users.get_username = MagicMock(return_value='test_user')
|
|
|
|
# Mock get_user_info
|
|
bt.get_user_info = MagicMock(return_value='test_user')
|
|
|
|
# Mock exchanges
|
|
bt.exchanges = MagicMock()
|
|
bt.exchanges.get_price = MagicMock(return_value=50000.0)
|
|
bt.users.get_exchanges = MagicMock(return_value=['binance'])
|
|
bt.users.get_api_keys = MagicMock(return_value={'key': 'k', 'secret': 's'})
|
|
mock_exchange = MagicMock()
|
|
mock_exchange.testnet = True
|
|
mock_exchange.configured = True
|
|
bt.exchanges.get_exchange = MagicMock(return_value=mock_exchange)
|
|
bt.exchanges.connect_exchange = MagicMock(return_value=True)
|
|
|
|
# Mock EDM client for exchange validation
|
|
bt.edm_client = MagicMock()
|
|
bt.edm_client.get_exchanges_sync = MagicMock(return_value=['binance', 'kucoin'])
|
|
|
|
# Mock strategies.get_strategy_by_tbl_key for exchange validation
|
|
bt.strategies.get_strategy_by_tbl_key = MagicMock(return_value={
|
|
'strategy_components': {},
|
|
'default_source': {}
|
|
})
|
|
|
|
return bt
|
|
|
|
def test_start_strategy_invalid_mode(self, mock_brighter_trades):
|
|
"""Test that invalid mode returns error."""
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='invalid_mode'
|
|
)
|
|
|
|
assert result['success'] is False
|
|
assert 'Invalid mode' in result['message']
|
|
|
|
def test_start_strategy_strategy_not_found(self, mock_brighter_trades):
|
|
"""Test that missing strategy returns error."""
|
|
# Mock empty result from database
|
|
mock_brighter_trades.strategies.data_cache.get_rows_from_datacache.return_value = MagicMock(empty=True)
|
|
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='nonexistent',
|
|
mode='paper'
|
|
)
|
|
|
|
assert result['success'] is False
|
|
assert 'not found' in result['message']
|
|
|
|
def test_start_strategy_authorization_check(self, mock_brighter_trades):
|
|
"""Test that non-owner, non-subscriber cannot run private strategy."""
|
|
import pandas as pd
|
|
|
|
# Mock strategy owned by different user (creator is user_id 2, not 1)
|
|
mock_strategy = pd.DataFrame([{
|
|
'tbl_key': 'test-strategy',
|
|
'name': 'Test Strategy',
|
|
'creator': 2, # Different user_id
|
|
'public': False,
|
|
'strategy_components': json.dumps({'generated_code': 'pass'})
|
|
}])
|
|
mock_brighter_trades.strategies.data_cache.get_rows_from_datacache.return_value = mock_strategy
|
|
|
|
# Mock that user is NOT subscribed
|
|
mock_brighter_trades.strategies.is_subscribed = MagicMock(return_value=False)
|
|
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='paper'
|
|
)
|
|
|
|
assert result['success'] is False
|
|
assert 'subscribe' in result['message'].lower()
|
|
|
|
def test_start_strategy_authorization_non_subscriber_denied(self, mock_brighter_trades):
|
|
"""
|
|
Test that non-owner, non-subscriber cannot run strategy (even private).
|
|
"""
|
|
import pandas as pd
|
|
|
|
mock_strategy = pd.DataFrame([{
|
|
'tbl_key': 'test-strategy',
|
|
'name': 'Test Strategy',
|
|
'creator': 2, # Different user_id
|
|
'public': False,
|
|
'strategy_components': json.dumps({'generated_code': 'pass'})
|
|
}])
|
|
mock_brighter_trades.strategies.data_cache.get_rows_from_datacache.return_value = mock_strategy
|
|
|
|
# Mock that user is NOT subscribed
|
|
mock_brighter_trades.strategies.is_subscribed = MagicMock(return_value=False)
|
|
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='paper'
|
|
)
|
|
|
|
assert result['success'] is False
|
|
assert 'subscribe' in result['message'].lower()
|
|
|
|
def test_start_strategy_live_mode_uses_live_active_instance_key(self, mock_brighter_trades):
|
|
"""Live mode now runs in actual live mode with proper instance keying."""
|
|
import pandas as pd
|
|
|
|
# Strategy owned by the running user (no subscription needed)
|
|
mock_strategy = pd.DataFrame([{
|
|
'tbl_key': 'test-strategy',
|
|
'name': 'Test Strategy',
|
|
'creator': 'test_user', # Same as mock user (user_id=1)
|
|
'public': True,
|
|
'strategy_components': json.dumps({'generated_code': 'pass'})
|
|
}])
|
|
mock_brighter_trades.strategies.data_cache.get_rows_from_datacache.return_value = mock_strategy
|
|
mock_brighter_trades.strategies.create_strategy_instance = MagicMock()
|
|
mock_brighter_trades.strategies.create_strategy_instance.return_value = MagicMock(
|
|
strategy_name='Test Strategy'
|
|
)
|
|
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='live'
|
|
)
|
|
|
|
assert result['success'] is True
|
|
assert result['actual_mode'] == 'live'
|
|
assert (1, 'test-strategy', 'live') in mock_brighter_trades.strategies.active_instances
|
|
|
|
def test_start_strategy_subscribed_strategy_allowed(self, mock_brighter_trades):
|
|
"""Test that a subscribed user can run a public strategy."""
|
|
import pandas as pd
|
|
|
|
# Mock public strategy owned by different user
|
|
mock_strategy = pd.DataFrame([{
|
|
'tbl_key': 'test-strategy',
|
|
'name': 'Public Strategy',
|
|
'creator': 'other_user',
|
|
'public': True,
|
|
'strategy_components': json.dumps({'generated_code': 'pass'})
|
|
}])
|
|
mock_brighter_trades.strategies.data_cache.get_rows_from_datacache.return_value = mock_strategy
|
|
mock_brighter_trades.strategies.create_strategy_instance = MagicMock()
|
|
mock_brighter_trades.strategies.create_strategy_instance.return_value = MagicMock(
|
|
strategy_name='Public Strategy'
|
|
)
|
|
# Mock the subscription check to return True
|
|
mock_brighter_trades.strategies.is_subscribed = MagicMock(return_value=True)
|
|
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='paper'
|
|
)
|
|
|
|
assert result['success'] is True
|
|
|
|
def test_start_strategy_unsubscribed_public_strategy_denied(self, mock_brighter_trades):
|
|
"""Test that unsubscribed user cannot run a public strategy they don't own."""
|
|
import pandas as pd
|
|
|
|
# Mock public strategy owned by different user
|
|
mock_strategy = pd.DataFrame([{
|
|
'tbl_key': 'test-strategy',
|
|
'name': 'Public Strategy',
|
|
'creator': 'other_user',
|
|
'public': True,
|
|
'strategy_components': json.dumps({'generated_code': 'pass'})
|
|
}])
|
|
mock_brighter_trades.strategies.data_cache.get_rows_from_datacache.return_value = mock_strategy
|
|
# Mock the subscription check to return False
|
|
mock_brighter_trades.strategies.is_subscribed = MagicMock(return_value=False)
|
|
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='paper'
|
|
)
|
|
|
|
assert result['success'] is False
|
|
assert 'subscribe' in result['message'].lower()
|
|
|
|
def test_start_strategy_already_running(self, mock_brighter_trades):
|
|
"""Test that strategy cannot be started twice in same mode."""
|
|
import pandas as pd
|
|
|
|
mock_strategy = pd.DataFrame([{
|
|
'tbl_key': 'test-strategy',
|
|
'name': 'Test Strategy',
|
|
'creator': 'test_user',
|
|
'public': False,
|
|
'strategy_components': json.dumps({'generated_code': 'pass'})
|
|
}])
|
|
mock_brighter_trades.strategies.data_cache.get_rows_from_datacache.return_value = mock_strategy
|
|
|
|
# Mark as already running
|
|
mock_brighter_trades.strategies.active_instances[(1, 'test-strategy', 'paper')] = MagicMock()
|
|
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='paper'
|
|
)
|
|
|
|
assert result['success'] is False
|
|
assert 'already running' in result['message']
|
|
|
|
def test_start_strategy_no_generated_code(self, mock_brighter_trades):
|
|
"""Test that strategy without code returns error."""
|
|
import pandas as pd
|
|
|
|
# Strategy with empty code
|
|
mock_strategy = pd.DataFrame([{
|
|
'tbl_key': 'test-strategy',
|
|
'name': 'Empty Strategy',
|
|
'creator': 'test_user',
|
|
'public': False,
|
|
'strategy_components': json.dumps({'generated_code': ''})
|
|
}])
|
|
mock_brighter_trades.strategies.data_cache.get_rows_from_datacache.return_value = mock_strategy
|
|
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='paper'
|
|
)
|
|
|
|
assert result['success'] is False
|
|
assert 'no generated code' in result['message']
|
|
|
|
def test_start_strategy_wrong_code_key(self, mock_brighter_trades):
|
|
"""Test that old 'code' key doesn't work (must be 'generated_code')."""
|
|
import pandas as pd
|
|
|
|
# Strategy with old key name
|
|
mock_strategy = pd.DataFrame([{
|
|
'tbl_key': 'test-strategy',
|
|
'name': 'Old Format Strategy',
|
|
'creator': 'test_user',
|
|
'public': False,
|
|
'strategy_components': json.dumps({'code': 'pass'}) # Wrong key!
|
|
}])
|
|
mock_brighter_trades.strategies.data_cache.get_rows_from_datacache.return_value = mock_strategy
|
|
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='paper'
|
|
)
|
|
|
|
assert result['success'] is False
|
|
assert 'no generated code' in result['message']
|
|
|
|
def test_start_strategy_testnet_override_bypasses_prod_gate(self, mock_brighter_trades, monkeypatch):
|
|
"""If config forces testnet, a non-testnet request should not be blocked as production."""
|
|
import pandas as pd
|
|
import config
|
|
|
|
mock_strategy = pd.DataFrame([{
|
|
'tbl_key': 'test-strategy',
|
|
'name': 'Test Strategy',
|
|
'creator': 'test_user',
|
|
'public': False,
|
|
'strategy_components': json.dumps({'generated_code': 'pass'})
|
|
}])
|
|
mock_brighter_trades.strategies.data_cache.get_rows_from_datacache.return_value = mock_strategy
|
|
mock_brighter_trades.strategies.create_strategy_instance = MagicMock()
|
|
mock_brighter_trades.strategies.create_strategy_instance.return_value = MagicMock(
|
|
strategy_name='Test Strategy'
|
|
)
|
|
|
|
monkeypatch.setattr(config, 'TESTNET_MODE', True, raising=False)
|
|
monkeypatch.setattr(config, 'ALLOW_LIVE_PRODUCTION', False, raising=False)
|
|
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='live',
|
|
testnet=False
|
|
)
|
|
|
|
assert result['success'] is True
|
|
assert result['testnet'] is True
|
|
|
|
|
|
class TestStopStrategy:
|
|
"""Tests for stop_strategy functionality."""
|
|
|
|
@pytest.fixture
|
|
def mock_brighter_trades(self):
|
|
"""Create a mock BrighterTrades instance."""
|
|
with patch.dict('sys.modules', {'eventlet': MagicMock()}):
|
|
from BrighterTrades import BrighterTrades
|
|
|
|
with patch.object(BrighterTrades, '__init__', lambda x, y: None):
|
|
bt = BrighterTrades(MagicMock())
|
|
bt.strategies = MagicMock()
|
|
bt.strategies.active_instances = {}
|
|
return bt
|
|
|
|
def test_stop_strategy_not_running(self, mock_brighter_trades):
|
|
"""Test stopping a strategy that isn't running."""
|
|
result = mock_brighter_trades.stop_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='paper'
|
|
)
|
|
|
|
assert result['success'] is False
|
|
assert 'not running' in result['message'].lower() or 'No running strategy' in result['message']
|
|
|
|
def test_stop_strategy_success(self, mock_brighter_trades):
|
|
"""Test successfully stopping a running strategy."""
|
|
# Set up running strategy
|
|
mock_instance = MagicMock()
|
|
mock_instance.strategy_name = 'Test Strategy'
|
|
mock_instance.broker = MagicMock()
|
|
mock_instance.broker.get_balance.return_value = 10500.0
|
|
mock_instance.broker.get_available_balance.return_value = 10500.0
|
|
mock_instance.trade_history = [{'pnl': 100}, {'pnl': 400}]
|
|
|
|
mock_brighter_trades.strategies.active_instances[(1, 'test-strategy', 'paper')] = mock_instance
|
|
|
|
result = mock_brighter_trades.stop_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='paper'
|
|
)
|
|
|
|
assert result['success'] is True
|
|
assert 'stopped' in result['message']
|
|
assert result['final_stats']['final_balance'] == 10500.0
|
|
assert result['final_stats']['total_trades'] == 2
|
|
|
|
# Verify removed from active instances
|
|
assert (1, 'test-strategy', 'paper') not in mock_brighter_trades.strategies.active_instances
|
|
|
|
def test_stop_strategy_live_mode_falls_back_to_paper_instance(self, mock_brighter_trades):
|
|
"""Stopping live should find fallback paper instance."""
|
|
mock_instance = MagicMock()
|
|
mock_instance.strategy_name = 'Test Strategy'
|
|
mock_instance.broker = MagicMock()
|
|
mock_instance.broker.get_balance.return_value = 10000.0
|
|
mock_instance.broker.get_available_balance.return_value = 10000.0
|
|
mock_instance.trade_history = []
|
|
|
|
mock_brighter_trades.strategies.active_instances[(1, 'test-strategy', 'paper')] = mock_instance
|
|
|
|
result = mock_brighter_trades.stop_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='live'
|
|
)
|
|
|
|
assert result['success'] is True
|
|
assert (1, 'test-strategy', 'paper') not in mock_brighter_trades.strategies.active_instances
|
|
|
|
|
|
class TestGetStrategyStatus:
|
|
"""Tests for get_strategy_status functionality."""
|
|
|
|
@pytest.fixture
|
|
def mock_brighter_trades(self):
|
|
"""Create a mock BrighterTrades instance."""
|
|
with patch.dict('sys.modules', {'eventlet': MagicMock()}):
|
|
from BrighterTrades import BrighterTrades
|
|
|
|
with patch.object(BrighterTrades, '__init__', lambda x, y: None):
|
|
bt = BrighterTrades(MagicMock())
|
|
bt.strategies = MagicMock()
|
|
bt.strategies.active_instances = {}
|
|
return bt
|
|
|
|
def test_get_status_no_running(self, mock_brighter_trades):
|
|
"""Test status when no strategies running."""
|
|
result = mock_brighter_trades.get_strategy_status(user_id=1)
|
|
|
|
assert result['success'] is True
|
|
assert result['running_strategies'] == []
|
|
assert result['count'] == 0
|
|
|
|
def test_get_status_with_running(self, mock_brighter_trades):
|
|
"""Test status with running strategies."""
|
|
# Set up running strategies
|
|
mock_instance1 = MagicMock()
|
|
mock_instance1.strategy_name = 'Strategy 1'
|
|
mock_instance1.strategy_instance_id = 'inst-1'
|
|
mock_instance1.broker = MagicMock()
|
|
mock_instance1.broker.get_balance.return_value = 10500.0
|
|
mock_instance1.broker.get_available_balance.return_value = 10000.0
|
|
mock_instance1.broker.get_all_positions.return_value = []
|
|
mock_instance1.trade_history = []
|
|
|
|
mock_instance2 = MagicMock()
|
|
mock_instance2.strategy_name = 'Strategy 2'
|
|
mock_instance2.strategy_instance_id = 'inst-2'
|
|
mock_instance2.broker = MagicMock()
|
|
mock_instance2.broker.get_balance.return_value = 9500.0
|
|
mock_instance2.broker.get_available_balance.return_value = 9500.0
|
|
mock_instance2.broker.get_all_positions.return_value = []
|
|
mock_instance2.trade_history = [{'pnl': -500}]
|
|
|
|
mock_brighter_trades.strategies.active_instances = {
|
|
(1, 'strat-1', 'paper'): mock_instance1,
|
|
(1, 'strat-2', 'paper'): mock_instance2,
|
|
(2, 'strat-3', 'paper'): MagicMock(), # Different user
|
|
}
|
|
|
|
result = mock_brighter_trades.get_strategy_status(user_id=1)
|
|
|
|
assert result['success'] is True
|
|
assert result['count'] == 2 # Only user 1's strategies
|
|
|
|
strat_ids = [s['strategy_id'] for s in result['running_strategies']]
|
|
assert 'strat-1' in strat_ids
|
|
assert 'strat-2' in strat_ids
|
|
assert 'strat-3' not in strat_ids # Different user
|
|
|
|
def test_get_status_filter_by_strategy(self, mock_brighter_trades):
|
|
"""Test status filtered by specific strategy."""
|
|
mock_instance = MagicMock()
|
|
mock_instance.strategy_name = 'Strategy 1'
|
|
mock_instance.strategy_instance_id = 'inst-1'
|
|
mock_instance.broker = MagicMock()
|
|
mock_instance.broker.get_balance.return_value = 10500.0
|
|
mock_instance.broker.get_available_balance.return_value = 10000.0
|
|
mock_instance.broker.get_all_positions.return_value = []
|
|
mock_instance.trade_history = []
|
|
|
|
mock_brighter_trades.strategies.active_instances = {
|
|
(1, 'strat-1', 'paper'): mock_instance,
|
|
(1, 'strat-2', 'paper'): MagicMock(),
|
|
}
|
|
|
|
result = mock_brighter_trades.get_strategy_status(
|
|
user_id=1,
|
|
strategy_id='strat-1'
|
|
)
|
|
|
|
assert result['count'] == 1
|
|
assert result['running_strategies'][0]['strategy_id'] == 'strat-1'
|
|
|
|
|
|
class TestMessageHandlerNumericParsing:
|
|
"""Tests for safe numeric parsing in message handlers."""
|
|
|
|
def test_run_strategy_invalid_initial_balance(self):
|
|
"""Test that invalid initial_balance is handled gracefully."""
|
|
# This would need integration test with actual message handler
|
|
# For now, verify the structure is correct
|
|
pass
|
|
|
|
def test_run_strategy_negative_balance_rejected(self):
|
|
"""Test that negative balance is rejected."""
|
|
pass
|
|
|
|
def test_run_strategy_invalid_commission_rejected(self):
|
|
"""Test that invalid commission is rejected."""
|
|
pass
|
|
|
|
|
|
class TestLiveModeWarning:
|
|
"""Tests for live mode fallback messaging."""
|
|
|
|
@pytest.fixture
|
|
def mock_brighter_trades(self):
|
|
"""Create a mock BrighterTrades instance."""
|
|
with patch.dict('sys.modules', {'eventlet': MagicMock()}):
|
|
from BrighterTrades import BrighterTrades
|
|
import pandas as pd
|
|
|
|
with patch.object(BrighterTrades, '__init__', lambda x, y: None):
|
|
bt = BrighterTrades(MagicMock())
|
|
bt.strategies = MagicMock()
|
|
bt.strategies.active_instances = {}
|
|
bt.strategies.data_cache = MagicMock()
|
|
bt.users = MagicMock()
|
|
bt.users.get_username = MagicMock(return_value='test_user')
|
|
bt.get_user_info = MagicMock(
|
|
side_effect=lambda user_name, info: 1 if info == 'User_id' else 'test_user'
|
|
)
|
|
bt.exchanges = MagicMock()
|
|
bt.exchanges.get_price = MagicMock(return_value=50000.0)
|
|
bt.users.get_exchanges = MagicMock(return_value=['binance'])
|
|
bt.users.get_api_keys = MagicMock(return_value={'key': 'k', 'secret': 's'})
|
|
mock_exchange = MagicMock()
|
|
mock_exchange.testnet = True
|
|
mock_exchange.configured = True
|
|
bt.exchanges.get_exchange = MagicMock(return_value=mock_exchange)
|
|
bt.exchanges.connect_exchange = MagicMock(return_value=True)
|
|
|
|
# Mock EDM client for exchange validation
|
|
bt.edm_client = MagicMock()
|
|
bt.edm_client.get_exchanges_sync = MagicMock(return_value=['binance', 'kucoin'])
|
|
|
|
# Set up valid strategy
|
|
mock_strategy = pd.DataFrame([{
|
|
'tbl_key': 'test-strategy',
|
|
'name': 'Test Strategy',
|
|
'creator': 'test_user',
|
|
'public': False,
|
|
'strategy_components': '{"generated_code": "pass"}'
|
|
}])
|
|
bt.strategies.data_cache.get_rows_from_datacache.return_value = mock_strategy
|
|
bt.strategies.create_strategy_instance = MagicMock()
|
|
bt.strategies.create_strategy_instance.return_value = MagicMock(
|
|
strategy_name='Test Strategy'
|
|
)
|
|
|
|
# Mock strategies.get_strategy_by_tbl_key for exchange validation
|
|
bt.strategies.get_strategy_by_tbl_key = MagicMock(return_value={
|
|
'strategy_components': {},
|
|
'default_source': {}
|
|
})
|
|
|
|
return bt
|
|
|
|
def test_live_mode_returns_success(self, mock_brighter_trades):
|
|
"""Test that live mode request succeeds in live mode."""
|
|
result = mock_brighter_trades.start_strategy(
|
|
user_id=1,
|
|
strategy_id='test-strategy',
|
|
mode='live'
|
|
)
|
|
|
|
# Should succeed in live mode
|
|
assert result['success'] is True
|
|
assert result['actual_mode'] == 'live'
|