Files
crypto_monitor/tests/test_instance_dashboard_lib.py
T
dekun c7aae67881 Add instance data dashboard (default off).
Read-only overview of live orders, key monitors, and strategy; show options/hedge only when present. Toggle via system settings nav prefs.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-17 12:09:29 +08:00

112 lines
3.7 KiB
Python

"""instance_dashboard_lib 单元测试."""
from __future__ import annotations
import sqlite3
import unittest
from lib.instance.instance_dashboard_lib import build_instance_dashboard_payload
def _mem_conn() -> sqlite3.Connection:
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.executescript(
"""
CREATE TABLE order_monitors (
id INTEGER PRIMARY KEY,
symbol TEXT,
exchange_symbol TEXT,
direction TEXT,
status TEXT,
monitor_type TEXT,
key_signal_type TEXT,
trigger_price REAL,
stop_loss REAL,
take_profit REAL
);
CREATE TABLE key_monitors (
id INTEGER PRIMARY KEY,
symbol TEXT,
exchange_symbol TEXT,
direction TEXT,
signal_type TEXT,
upper REAL,
lower REAL,
status TEXT
);
CREATE TABLE trend_pullback_plans (
id INTEGER PRIMARY KEY,
symbol TEXT,
exchange_symbol TEXT,
direction TEXT,
status TEXT,
entry_price REAL
);
CREATE TABLE roll_groups (
id INTEGER PRIMARY KEY,
order_monitor_id INTEGER,
symbol TEXT,
exchange_symbol TEXT,
direction TEXT,
status TEXT
);
"""
)
return conn
class TestInstanceDashboardLib(unittest.TestCase):
def test_empty_sections_and_conditional_hidden(self):
conn = _mem_conn()
payload = build_instance_dashboard_payload(conn, hedge_enabled=True)
self.assertTrue(payload["ok"])
self.assertEqual(payload["orders"]["count"], 0)
self.assertEqual(payload["keys"]["count"], 0)
self.assertEqual(payload["strategy"]["count"], 0)
self.assertFalse(payload["options"]["visible"])
self.assertFalse(payload["hedge_plan"]["visible"])
conn.close()
def test_orders_keys_strategy_and_options_visible(self):
conn = _mem_conn()
conn.execute(
"INSERT INTO order_monitors (symbol, exchange_symbol, direction, status, monitor_type) "
"VALUES ('BTC/USDT', 'BTC/USDT:USDT', 'long', 'active', 'manual')"
)
conn.execute(
"INSERT INTO key_monitors (symbol, direction, signal_type, upper, lower, status) "
"VALUES ('ETH/USDT', 'short', '箱体突破', 3000, 2800, 'active')"
)
conn.execute(
"INSERT INTO trend_pullback_plans (symbol, direction, status, entry_price) "
"VALUES ('SOL/USDT', 'long', 'active', 100)"
)
conn.execute(
"INSERT INTO order_monitors (id, symbol, direction, status) VALUES (9, 'XRP/USDT', 'short', 'active')"
)
conn.execute(
"INSERT INTO roll_groups (order_monitor_id, symbol, direction, status) "
"VALUES (9, 'XRP/USDT', 'short', 'active')"
)
conn.commit()
def fetch_opts():
return [{"inst_id": "ETH-USD-260731-3000-C", "opt_type": "C", "pos": 1, "upl": 1.5}]
payload = build_instance_dashboard_payload(
conn,
fetch_options_positions=fetch_opts,
hedge_enabled=False,
)
self.assertEqual(payload["orders"]["count"], 2)
self.assertEqual(payload["keys"]["count"], 1)
self.assertEqual(payload["strategy"]["count"], 2)
self.assertTrue(payload["options"]["visible"])
self.assertEqual(payload["options"]["count"], 1)
self.assertFalse(payload["hedge_plan"]["visible"])
conn.close()
if __name__ == "__main__":
unittest.main()