dekun
75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
"""账户内划转执行:门禁与余额回写."""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
class TestOptionsTransferExec(unittest.TestCase):
|
|
def test_block_when_position(self):
|
|
from lib.options.options_transfer_exec_lib import execute_options_internal_transfer
|
|
|
|
conn = MagicMock()
|
|
cfg = {
|
|
"exchange_options": object(),
|
|
"options_api_ready": lambda ex: (True, ""),
|
|
"get_db": lambda: conn,
|
|
"transfer_ccy": MagicMock(),
|
|
}
|
|
with patch(
|
|
"lib.options.options_transfer_exec_lib.transfer_blocked_by_position_msg",
|
|
create=True,
|
|
):
|
|
with patch(
|
|
"lib.options.options_account_pnl_lib.transfer_blocked_by_position_msg",
|
|
return_value="期权持仓中禁止划转",
|
|
):
|
|
out = execute_options_internal_transfer(
|
|
cfg, amount=10, from_acct="funding", to_acct="trading"
|
|
)
|
|
self.assertFalse(out["ok"])
|
|
self.assertTrue(out.get("transfer_blocked"))
|
|
cfg["transfer_ccy"].assert_not_called()
|
|
|
|
def test_success_returns_balances(self):
|
|
from lib.options.options_transfer_exec_lib import execute_options_internal_transfer
|
|
|
|
conn = MagicMock()
|
|
xfer = MagicMock(return_value={"ok": True, "msg": "ok"})
|
|
fetch_bal = MagicMock(
|
|
return_value={
|
|
"trading_usdt": 30.0,
|
|
"funding_usdt": 70.0,
|
|
"trading_usdt_avail": 30.0,
|
|
"funding_usdt_avail": 70.0,
|
|
}
|
|
)
|
|
cfg = {
|
|
"exchange_options": object(),
|
|
"options_api_ready": lambda ex: (True, ""),
|
|
"get_db": lambda: conn,
|
|
"transfer_ccy": xfer,
|
|
"fetch_options_balances": fetch_bal,
|
|
}
|
|
with patch(
|
|
"lib.options.options_account_pnl_lib.transfer_blocked_by_position_msg",
|
|
return_value=None,
|
|
), patch(
|
|
"lib.options.options_db.init_options_tables",
|
|
), patch(
|
|
"lib.exchange.okx_options_lib.invalidate_options_balance_cache",
|
|
), patch(
|
|
"lib.instance.instance_live_push_lib.notify_instance_balance_changed",
|
|
):
|
|
out = execute_options_internal_transfer(
|
|
cfg, amount=10, from_acct="funding", to_acct="trading", ccy="USDT"
|
|
)
|
|
self.assertTrue(out["ok"])
|
|
self.assertAlmostEqual(out["trading_usdt"], 30.0)
|
|
self.assertAlmostEqual(out["funding_usdt"], 70.0)
|
|
xfer.assert_called_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|