289 lines
11 KiB
Python
289 lines
11 KiB
Python
"""
|
||
BitgetTradingAPI.get_balance() 单元测试
|
||
|
||
覆盖场景：
|
||
1. UTA 模式 → privateUtaGetV3AccountAssets（真实路径）
|
||
2. 经典账户模式 → privateMixGetV2MixAccountAccounts
|
||
3. API 异常 → 返回 {}
|
||
4. 空 assets / 无 USDT
|
||
5. get_account_state 端到端解析
|
||
"""
|
||
import importlib.util
|
||
import os
|
||
import sys
|
||
import types
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock
|
||
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||
|
||
|
||
def load_bitget_sdk_class():
|
||
sdk_path = (
|
||
Path(__file__).resolve().parents[1]
|
||
/ 'app' / 'services' / 'bitget_trading_api_sdk.py'
|
||
)
|
||
|
||
if 'ccxt' not in sys.modules:
|
||
ccxt_module = types.ModuleType('ccxt')
|
||
ccxt_module.BaseError = Exception
|
||
ccxt_module.bitget = MagicMock()
|
||
sys.modules['ccxt'] = ccxt_module
|
||
|
||
if 'app.config' not in sys.modules:
|
||
mock_cfg = types.ModuleType('app.config')
|
||
mock_cfg.get_settings = MagicMock()
|
||
sys.modules['app.config'] = mock_cfg
|
||
|
||
if 'app.utils.logger' not in sys.modules:
|
||
mock_log = types.ModuleType('app.utils.logger')
|
||
mock_log.logger = MagicMock()
|
||
sys.modules['app.utils.logger'] = mock_log
|
||
|
||
module_name = 'app.services.bitget_trading_api_sdk_bal_test'
|
||
spec = importlib.util.spec_from_file_location(module_name, sdk_path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
sys.modules[module_name] = module
|
||
spec.loader.exec_module(module)
|
||
return module.BitgetTradingAPI
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 测试 1: UTA 模式，正常返回（匹配真实 API）
|
||
# ──────────────────────────────────────────────
|
||
def test_uta_balance_success():
|
||
"""UTA 模式 → privateUtaGetV3AccountAssets 返回 1975.37"""
|
||
BitgetTradingAPI = load_bitget_sdk_class()
|
||
api = BitgetTradingAPI.__new__(BitgetTradingAPI)
|
||
api.exchange = MagicMock()
|
||
api.use_unified_account = True
|
||
|
||
api.exchange.privateUtaGetV3AccountAssets.return_value = {
|
||
'code': '00000',
|
||
'data': {
|
||
'assets': [
|
||
{
|
||
'coin': 'USDT',
|
||
'available': '1975.37457492',
|
||
'locked': '0',
|
||
'equity': '1975.37457492',
|
||
}
|
||
],
|
||
},
|
||
}
|
||
|
||
balance = api.get_balance()
|
||
|
||
assert balance['USDT']['available'] == '1975.37457492'
|
||
assert balance['USDT']['equity'] == '1975.37457492'
|
||
assert balance['USDT']['frozen'] == '0'
|
||
api.exchange.privateUtaGetV3AccountAssets.assert_called_once_with({'coin': 'USDT'})
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 测试 2: UTA 多币种返回
|
||
# ──────────────────────────────────────────────
|
||
def test_uta_multi_coins():
|
||
"""UTA 返回多个币种，正确解析"""
|
||
BitgetTradingAPI = load_bitget_sdk_class()
|
||
api = BitgetTradingAPI.__new__(BitgetTradingAPI)
|
||
api.exchange = MagicMock()
|
||
api.use_unified_account = True
|
||
|
||
api.exchange.privateUtaGetV3AccountAssets.return_value = {
|
||
'code': '00000',
|
||
'data': {
|
||
'assets': [
|
||
{'coin': 'USDT', 'available': '1000.00', 'locked': '200.00', 'equity': '1200.00'},
|
||
{'coin': 'BGB', 'available': '50.00', 'locked': '0', 'equity': '50.00'},
|
||
],
|
||
},
|
||
}
|
||
|
||
balance = api.get_balance()
|
||
|
||
assert balance['USDT']['available'] == '1000.00'
|
||
assert balance['USDT']['frozen'] == '200.00'
|
||
assert balance['BGB']['available'] == '50.00'
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 测试 3: 经典合约账户模式
|
||
# ──────────────────────────────────────────────
|
||
def test_contract_account_balance():
|
||
"""经典账户 → privateMixGetV2MixAccountAccounts"""
|
||
BitgetTradingAPI = load_bitget_sdk_class()
|
||
api = BitgetTradingAPI.__new__(BitgetTradingAPI)
|
||
api.exchange = MagicMock()
|
||
api.use_unified_account = False
|
||
|
||
api.exchange.privateMixGetV2MixAccountAccounts.return_value = {
|
||
'code': '00000',
|
||
'data': [
|
||
{
|
||
'marginCoin': 'USDT',
|
||
'available': '800.50',
|
||
'locked': '100.00',
|
||
'accountEquity': '900.50',
|
||
}
|
||
],
|
||
}
|
||
|
||
balance = api.get_balance()
|
||
|
||
assert balance['USDT']['available'] == '800.50'
|
||
assert balance['USDT']['frozen'] == '100.00'
|
||
assert balance['USDT']['equity'] == '900.50'
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 测试 4: API 异常 → 返回 {}
|
||
# ──────────────────────────────────────────────
|
||
def test_api_exception_returns_empty():
|
||
"""API 抛异常 → 返回空字典"""
|
||
BitgetTradingAPI = load_bitget_sdk_class()
|
||
api = BitgetTradingAPI.__new__(BitgetTradingAPI)
|
||
api.exchange = MagicMock()
|
||
api.use_unified_account = True
|
||
|
||
api.exchange.privateUtaGetV3AccountAssets.side_effect = Exception("network error")
|
||
|
||
balance = api.get_balance()
|
||
|
||
assert balance == {}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 测试 5: UTA 空 assets → 返回 {}
|
||
# ──────────────────────────────────────────────
|
||
def test_uta_empty_assets():
|
||
"""UTA API 返回空 assets 列表"""
|
||
BitgetTradingAPI = load_bitget_sdk_class()
|
||
api = BitgetTradingAPI.__new__(BitgetTradingAPI)
|
||
api.exchange = MagicMock()
|
||
api.use_unified_account = True
|
||
|
||
api.exchange.privateUtaGetV3AccountAssets.return_value = {
|
||
'code': '00000',
|
||
'data': {'assets': []},
|
||
}
|
||
|
||
balance = api.get_balance()
|
||
|
||
assert balance == {}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 测试 6: UTA assets 无 coin 字段 → 跳过
|
||
# ──────────────────────────────────────────────
|
||
def test_uta_skip_no_coin():
|
||
"""assets 中 entry 无 coin → 跳过该条目"""
|
||
BitgetTradingAPI = load_bitget_sdk_class()
|
||
api = BitgetTradingAPI.__new__(BitgetTradingAPI)
|
||
api.exchange = MagicMock()
|
||
api.use_unified_account = True
|
||
|
||
api.exchange.privateUtaGetV3AccountAssets.return_value = {
|
||
'code': '00000',
|
||
'data': {
|
||
'assets': [
|
||
{'coin': '', 'available': '100'},
|
||
{'coin': 'USDT', 'available': '500.00', 'locked': '0', 'equity': '500.00'},
|
||
],
|
||
},
|
||
}
|
||
|
||
balance = api.get_balance()
|
||
|
||
assert 'USDT' in balance
|
||
assert balance['USDT']['available'] == '500.00'
|
||
assert len(balance) == 1
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 测试 7: UTA assets 中字段为 None → 解析为 '0'
|
||
# ──────────────────────────────────────────────
|
||
def test_uta_none_fields():
|
||
"""UTA assets 中 available/locked 为 None → 解析为 '0'"""
|
||
BitgetTradingAPI = load_bitget_sdk_class()
|
||
api = BitgetTradingAPI.__new__(BitgetTradingAPI)
|
||
api.exchange = MagicMock()
|
||
api.use_unified_account = True
|
||
|
||
api.exchange.privateUtaGetV3AccountAssets.return_value = {
|
||
'code': '00000',
|
||
'data': {
|
||
'assets': [
|
||
{'coin': 'USDT', 'available': None, 'locked': None, 'equity': None},
|
||
],
|
||
},
|
||
}
|
||
|
||
balance = api.get_balance()
|
||
|
||
assert balance['USDT']['available'] == '0' # _pick_first_value 对 None 返回默认值 '0'
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 测试 8: get_account_state 端到端解析（正常余额）
|
||
# ──────────────────────────────────────────────
|
||
def test_get_account_state_e2e():
|
||
"""get_account_state 正确解析 balance 返回值"""
|
||
balance_return = {
|
||
'USDT': {
|
||
'available': '1975.50',
|
||
'frozen': '24.50',
|
||
'locked': '24.50',
|
||
'equity': '2000.00',
|
||
}
|
||
}
|
||
|
||
usdt = balance_return.get('USDT', {})
|
||
available = float(usdt.get('available', 0) or 0)
|
||
frozen = float(usdt.get('frozen', 0) or 0)
|
||
equity = float(usdt.get('equity', 0) or 0)
|
||
account_value = equity if equity > 0 else (available + frozen)
|
||
|
||
assert available == 1975.50
|
||
assert frozen == 24.50
|
||
assert equity == 2000.00
|
||
assert account_value == 2000.00
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 测试 9: balance 返回 {} 时 get_account_state 各字段为 0
|
||
# ──────────────────────────────────────────────
|
||
def test_get_account_state_empty_balance():
|
||
"""balance 返回空 → 全部为 0"""
|
||
balance_return = {}
|
||
|
||
usdt = balance_return.get('USDT', {})
|
||
available = float(usdt.get('available', 0) or 0)
|
||
frozen = float(usdt.get('frozen', 0) or 0)
|
||
equity = float(usdt.get('equity', 0) or 0)
|
||
account_value = equity if equity > 0 else (available + frozen)
|
||
|
||
assert available == 0.0
|
||
assert frozen == 0.0
|
||
assert account_value == 0.0
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 测试 10: UTA 返回 data=None → 返回 {}
|
||
# ──────────────────────────────────────────────
|
||
def test_uta_data_none():
|
||
"""UTA API 返回 data=None"""
|
||
BitgetTradingAPI = load_bitget_sdk_class()
|
||
api = BitgetTradingAPI.__new__(BitgetTradingAPI)
|
||
api.exchange = MagicMock()
|
||
api.use_unified_account = True
|
||
|
||
api.exchange.privateUtaGetV3AccountAssets.return_value = {
|
||
'code': '00000',
|
||
'data': None,
|
||
}
|
||
|
||
balance = api.get_balance()
|
||
|
||
assert balance == {}
|