426 lines
15 KiB
Python
426 lines
15 KiB
Python
"""
|
|
Unit tests for exchange_validation module.
|
|
|
|
Tests cover:
|
|
- Exchange extraction from strategy data
|
|
- Mode-specific validation (backtest, paper, live)
|
|
- Exchange name canonicalization
|
|
- Error code generation
|
|
- Edge cases (empty data, EDM unavailable, etc.)
|
|
"""
|
|
|
|
import pytest
|
|
import sys
|
|
import os
|
|
|
|
# Add src to path for imports
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
|
|
|
|
from exchange_validation import (
|
|
canonicalize_exchange,
|
|
extract_required_exchanges,
|
|
validate_for_backtest,
|
|
validate_for_paper,
|
|
validate_for_live,
|
|
validate_exchange_requirements,
|
|
ValidationErrorCode,
|
|
ExchangeValidationResult,
|
|
)
|
|
|
|
|
|
class TestCanonicalizeExchange:
|
|
"""Tests for exchange name canonicalization."""
|
|
|
|
def test_lowercase_normalization(self):
|
|
"""Test that exchange names are normalized to lowercase."""
|
|
assert canonicalize_exchange('BINANCE') == 'binance'
|
|
assert canonicalize_exchange('Binance') == 'binance'
|
|
assert canonicalize_exchange('binance') == 'binance'
|
|
|
|
def test_alias_mapping(self):
|
|
"""Test that known aliases are mapped correctly."""
|
|
assert canonicalize_exchange('okex') == 'okx'
|
|
assert canonicalize_exchange('OKEX') == 'okx'
|
|
assert canonicalize_exchange('huobi') == 'htx'
|
|
assert canonicalize_exchange('gate') == 'gateio'
|
|
|
|
def test_passthrough_unknown(self):
|
|
"""Test that unknown exchange names pass through lowercased."""
|
|
assert canonicalize_exchange('someexchange') == 'someexchange'
|
|
assert canonicalize_exchange('NEWEXCHANGE') == 'newexchange'
|
|
|
|
def test_empty_string(self):
|
|
"""Test handling of empty string."""
|
|
assert canonicalize_exchange('') == ''
|
|
assert canonicalize_exchange(' ') == ''
|
|
|
|
def test_whitespace_stripping(self):
|
|
"""Test that whitespace is stripped."""
|
|
assert canonicalize_exchange(' binance ') == 'binance'
|
|
assert canonicalize_exchange('\tkucoin\n') == 'kucoin'
|
|
|
|
|
|
class TestExtractRequiredExchanges:
|
|
"""Tests for extracting required exchanges from strategy data."""
|
|
|
|
def test_empty_strategy(self):
|
|
"""Test handling of empty/None strategy."""
|
|
assert extract_required_exchanges(None) == set()
|
|
assert extract_required_exchanges({}) == set()
|
|
|
|
def test_extract_from_data_sources_tuple(self):
|
|
"""Test extraction from data_sources as tuples."""
|
|
strategy = {
|
|
'strategy_components': {
|
|
'data_sources': [
|
|
('binance', 'BTC/USDT', '1h'),
|
|
('kucoin', 'ETH/USDT', '5m'),
|
|
]
|
|
}
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'binance', 'kucoin'}
|
|
|
|
def test_extract_from_data_sources_list(self):
|
|
"""Test extraction from data_sources as lists."""
|
|
strategy = {
|
|
'strategy_components': {
|
|
'data_sources': [
|
|
['binance', 'BTC/USDT', '1h'],
|
|
]
|
|
}
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'binance'}
|
|
|
|
def test_extract_from_data_sources_dict(self):
|
|
"""Test extraction from data_sources as dicts."""
|
|
strategy = {
|
|
'strategy_components': {
|
|
'data_sources': [
|
|
{'exchange': 'binance', 'symbol': 'BTC/USDT', 'timeframe': '1h'},
|
|
]
|
|
}
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'binance'}
|
|
|
|
def test_extract_from_default_source(self):
|
|
"""Test extraction from default_source."""
|
|
strategy = {
|
|
'default_source': {
|
|
'exchange': 'kucoin',
|
|
'symbol': 'BTC/USDT',
|
|
'timeframe': '15m'
|
|
}
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'kucoin'}
|
|
|
|
def test_extract_combined_sources(self):
|
|
"""Test extraction from both data_sources and default_source."""
|
|
strategy = {
|
|
'strategy_components': {
|
|
'data_sources': [
|
|
('binance', 'BTC/USDT', '1h'),
|
|
]
|
|
},
|
|
'default_source': {
|
|
'exchange': 'kucoin',
|
|
'symbol': 'ETH/USDT',
|
|
}
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'binance', 'kucoin'}
|
|
|
|
def test_extract_with_canonicalization(self):
|
|
"""Test that extracted exchanges are canonicalized."""
|
|
strategy = {
|
|
'strategy_components': {
|
|
'data_sources': [
|
|
('BINANCE', 'BTC/USDT', '1h'),
|
|
('okex', 'ETH/USDT', '5m'), # Should become 'okx'
|
|
]
|
|
}
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'binance', 'okx'}
|
|
|
|
def test_extract_deduplication(self):
|
|
"""Test that duplicate exchanges are deduplicated."""
|
|
strategy = {
|
|
'strategy_components': {
|
|
'data_sources': [
|
|
('binance', 'BTC/USDT', '1h'),
|
|
('Binance', 'ETH/USDT', '5m'),
|
|
('BINANCE', 'LTC/USDT', '15m'),
|
|
]
|
|
}
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'binance'}
|
|
|
|
def test_extract_json_string_components(self):
|
|
"""Test extraction when strategy_components is JSON string."""
|
|
import json
|
|
strategy = {
|
|
'strategy_components': json.dumps({
|
|
'data_sources': [
|
|
['binance', 'BTC/USDT', '1h'],
|
|
]
|
|
})
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'binance'}
|
|
|
|
|
|
class TestValidateForBacktest:
|
|
"""Tests for backtest mode validation."""
|
|
|
|
def test_empty_requirements(self):
|
|
"""Test that empty requirements are valid."""
|
|
result = validate_for_backtest(set(), ['binance', 'kucoin'])
|
|
assert result.valid is True
|
|
|
|
def test_all_available(self):
|
|
"""Test validation when all exchanges are available."""
|
|
result = validate_for_backtest(
|
|
{'binance', 'kucoin'},
|
|
['binance', 'kucoin', 'kraken']
|
|
)
|
|
assert result.valid is True
|
|
|
|
def test_missing_exchanges(self):
|
|
"""Test validation when exchanges are missing."""
|
|
result = validate_for_backtest(
|
|
{'binance', 'kucoin', 'unknown_exchange'},
|
|
['binance', 'kucoin']
|
|
)
|
|
assert result.valid is False
|
|
assert result.error_code == ValidationErrorCode.MISSING_EDM_DATA
|
|
assert result.missing_exchanges == {'unknown_exchange'}
|
|
assert 'unknown_exchange' in result.message
|
|
|
|
def test_empty_edm_list(self):
|
|
"""Test validation when EDM list is empty."""
|
|
result = validate_for_backtest({'binance'}, [])
|
|
assert result.valid is False
|
|
assert result.error_code == ValidationErrorCode.MISSING_EDM_DATA
|
|
|
|
def test_none_edm_list(self):
|
|
"""Test validation when EDM list is None."""
|
|
result = validate_for_backtest({'binance'}, None)
|
|
assert result.valid is False
|
|
assert result.error_code == ValidationErrorCode.MISSING_EDM_DATA
|
|
|
|
def test_canonicalization_in_comparison(self):
|
|
"""Test that EDM exchanges are canonicalized during comparison."""
|
|
result = validate_for_backtest(
|
|
{'binance'},
|
|
['BINANCE'] # Should match after canonicalization
|
|
)
|
|
assert result.valid is True
|
|
|
|
def test_edm_returns_dicts(self):
|
|
"""Test that EDM response with dicts (containing 'name' key) is handled."""
|
|
result = validate_for_backtest(
|
|
{'binance', 'kucoin'},
|
|
[{'name': 'binance', 'timeframes': ['1m', '1h']},
|
|
{'name': 'kucoin', 'timeframes': ['5m', '15m']}]
|
|
)
|
|
assert result.valid is True
|
|
|
|
def test_edm_returns_dicts_missing(self):
|
|
"""Test that missing exchanges are detected with dict format."""
|
|
result = validate_for_backtest(
|
|
{'binance', 'kraken'},
|
|
[{'name': 'binance'}]
|
|
)
|
|
assert result.valid is False
|
|
assert result.missing_exchanges == {'kraken'}
|
|
|
|
|
|
class TestValidateForPaper:
|
|
"""Tests for paper mode validation."""
|
|
|
|
def test_empty_requirements(self):
|
|
"""Test that empty requirements are valid."""
|
|
result = validate_for_paper(set(), [])
|
|
assert result.valid is True
|
|
|
|
def test_valid_ccxt_exchange(self):
|
|
"""Test validation with valid ccxt exchange."""
|
|
result = validate_for_paper({'binance'}, [])
|
|
assert result.valid is True
|
|
|
|
def test_invalid_exchange(self):
|
|
"""Test validation with invalid exchange name."""
|
|
result = validate_for_paper({'totally_fake_exchange_xyz'}, [])
|
|
assert result.valid is False
|
|
assert result.error_code == ValidationErrorCode.INVALID_EXCHANGE
|
|
assert 'totally_fake_exchange_xyz' in result.missing_exchanges
|
|
|
|
def test_mixed_valid_invalid(self):
|
|
"""Test validation with mix of valid and invalid exchanges."""
|
|
result = validate_for_paper({'binance', 'fake_exchange'}, [])
|
|
assert result.valid is False
|
|
assert result.missing_exchanges == {'fake_exchange'}
|
|
|
|
|
|
class TestValidateForLive:
|
|
"""Tests for live mode validation."""
|
|
|
|
def test_empty_requirements(self):
|
|
"""Test that empty requirements are valid."""
|
|
result = validate_for_live(set(), [])
|
|
assert result.valid is True
|
|
|
|
def test_configured_exchange(self):
|
|
"""Test validation when exchange is configured."""
|
|
result = validate_for_live({'binance'}, ['binance', 'kucoin'])
|
|
assert result.valid is True
|
|
|
|
def test_missing_config(self):
|
|
"""Test validation when exchange is not configured."""
|
|
result = validate_for_live({'kraken'}, ['binance', 'kucoin'])
|
|
assert result.valid is False
|
|
assert result.error_code == ValidationErrorCode.MISSING_CONFIG
|
|
assert result.missing_exchanges == {'kraken'}
|
|
assert 'API keys' in result.message
|
|
|
|
def test_multiple_missing(self):
|
|
"""Test validation with multiple missing exchanges."""
|
|
result = validate_for_live(
|
|
{'binance', 'kraken', 'bybit'},
|
|
['binance']
|
|
)
|
|
assert result.valid is False
|
|
assert result.missing_exchanges == {'kraken', 'bybit'}
|
|
|
|
def test_canonicalization_in_comparison(self):
|
|
"""Test that configured exchanges are canonicalized."""
|
|
result = validate_for_live({'binance'}, ['BINANCE'])
|
|
assert result.valid is True
|
|
|
|
|
|
class TestValidateExchangeRequirements:
|
|
"""Tests for main validation entrypoint."""
|
|
|
|
def test_routes_to_backtest(self):
|
|
"""Test that backtest mode routes correctly."""
|
|
result = validate_exchange_requirements(
|
|
required_exchanges={'binance'},
|
|
user_configured_exchanges=[],
|
|
edm_available_exchanges=['binance'],
|
|
mode='backtest'
|
|
)
|
|
assert result.valid is True
|
|
|
|
def test_routes_to_paper(self):
|
|
"""Test that paper mode routes correctly."""
|
|
result = validate_exchange_requirements(
|
|
required_exchanges={'binance'},
|
|
user_configured_exchanges=[],
|
|
edm_available_exchanges=[],
|
|
mode='paper'
|
|
)
|
|
assert result.valid is True # binance is valid ccxt exchange
|
|
|
|
def test_routes_to_live(self):
|
|
"""Test that live mode routes correctly."""
|
|
result = validate_exchange_requirements(
|
|
required_exchanges={'binance'},
|
|
user_configured_exchanges=['binance'],
|
|
edm_available_exchanges=[],
|
|
mode='live'
|
|
)
|
|
assert result.valid is True
|
|
|
|
def test_empty_requirements_all_modes(self):
|
|
"""Test that empty requirements are valid for all modes."""
|
|
for mode in ['backtest', 'paper', 'live']:
|
|
result = validate_exchange_requirements(
|
|
required_exchanges=set(),
|
|
user_configured_exchanges=[],
|
|
edm_available_exchanges=[],
|
|
mode=mode
|
|
)
|
|
assert result.valid is True, f"Failed for mode: {mode}"
|
|
|
|
def test_unknown_mode_rejected(self):
|
|
"""Test that unknown mode is rejected."""
|
|
result = validate_exchange_requirements(
|
|
required_exchanges={'binance'},
|
|
user_configured_exchanges=[],
|
|
edm_available_exchanges=[],
|
|
mode='invalid_mode'
|
|
)
|
|
assert result.valid is False
|
|
assert 'Invalid trading mode' in result.message
|
|
|
|
|
|
class TestExchangeValidationResult:
|
|
"""Tests for ExchangeValidationResult class."""
|
|
|
|
def test_valid_result_to_dict(self):
|
|
"""Test serialization of valid result."""
|
|
result = ExchangeValidationResult(valid=True)
|
|
d = result.to_dict()
|
|
assert d == {"valid": True}
|
|
|
|
def test_invalid_result_to_dict(self):
|
|
"""Test serialization of invalid result."""
|
|
result = ExchangeValidationResult(
|
|
valid=False,
|
|
error_code=ValidationErrorCode.MISSING_CONFIG,
|
|
missing_exchanges={'binance', 'kucoin'},
|
|
message="API keys required"
|
|
)
|
|
d = result.to_dict()
|
|
assert d["valid"] is False
|
|
assert d["error_code"] == "missing_config"
|
|
assert set(d["missing_exchanges"]) == {'binance', 'kucoin'}
|
|
assert d["message"] == "API keys required"
|
|
|
|
|
|
class TestEdgeCases:
|
|
"""Tests for edge cases and error handling."""
|
|
|
|
def test_none_values_in_data_sources(self):
|
|
"""Test handling of None values in data sources."""
|
|
strategy = {
|
|
'strategy_components': {
|
|
'data_sources': [
|
|
(None, 'BTC/USDT', '1h'),
|
|
('binance', 'BTC/USDT', '1h'),
|
|
]
|
|
}
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'binance'}
|
|
|
|
def test_empty_tuple_in_data_sources(self):
|
|
"""Test handling of empty tuples in data sources."""
|
|
strategy = {
|
|
'strategy_components': {
|
|
'data_sources': [
|
|
(),
|
|
('binance', 'BTC/USDT', '1h'),
|
|
]
|
|
}
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'binance'}
|
|
|
|
def test_single_element_tuple(self):
|
|
"""Test handling of single element tuple (just exchange)."""
|
|
strategy = {
|
|
'strategy_components': {
|
|
'data_sources': [
|
|
('binance',),
|
|
]
|
|
}
|
|
}
|
|
result = extract_required_exchanges(strategy)
|
|
assert result == {'binance'}
|