dekun
53863559f4
Retarget git remote, install path, and deploy docs from crypto_monitor to crypto_monitor_user. Co-authored-by: Cursor <cursoragent@cursor.com>
45 lines
1.8 KiB
Python
45 lines
1.8 KiB
Python
"""gate_transfer_lib 单元测试."""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import unittest
|
|
|
|
from lib.exchange.gate_transfer_lib import count_auto_transfer_blockers
|
|
|
|
|
|
class GateTransferLibTest(unittest.TestCase):
|
|
def test_counts_order_monitors_first(self):
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE order_monitors (status TEXT)")
|
|
conn.execute("CREATE TABLE trend_pullback_plans (status TEXT, first_order_done INTEGER)")
|
|
conn.execute("INSERT INTO order_monitors VALUES ('active')")
|
|
conn.execute("INSERT INTO trend_pullback_plans VALUES ('active', 1)")
|
|
conn.commit()
|
|
n = count_auto_transfer_blockers(conn, count_order_monitors=lambda c: 1)
|
|
self.assertEqual(n, 1)
|
|
conn.close()
|
|
|
|
def test_counts_trend_plan_when_no_order_monitors(self):
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE order_monitors (status TEXT)")
|
|
conn.execute("CREATE TABLE trend_pullback_plans (status TEXT, first_order_done INTEGER)")
|
|
conn.execute("INSERT INTO trend_pullback_plans VALUES ('active', 1)")
|
|
conn.commit()
|
|
n = count_auto_transfer_blockers(conn, count_order_monitors=lambda c: 0)
|
|
self.assertEqual(n, 1)
|
|
conn.close()
|
|
|
|
def test_ignores_trend_plan_without_first_order(self):
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE order_monitors (status TEXT)")
|
|
conn.execute("CREATE TABLE trend_pullback_plans (status TEXT, first_order_done INTEGER)")
|
|
conn.execute("INSERT INTO trend_pullback_plans VALUES ('active', 0)")
|
|
conn.commit()
|
|
n = count_auto_transfer_blockers(conn, count_order_monitors=lambda c: 0)
|
|
self.assertEqual(n, 0)
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|