dekun
87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
"""阻力/支撑提醒：占位与间隔防重复推送。"""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import unittest
|
|
from datetime import datetime, timedelta
|
|
|
|
from key_monitor_lib import (
|
|
claim_rs_level_notify,
|
|
notify_interval_elapsed,
|
|
run_rs_level_alert_tick,
|
|
)
|
|
|
|
|
|
def _row(**kwargs):
|
|
base = {
|
|
"upper": 2.174,
|
|
"lower": 1.694,
|
|
"notification_count": 0,
|
|
"max_notify": 3,
|
|
"notify_interval_min": 5,
|
|
"direction": "watch",
|
|
"last_notified_at": None,
|
|
"last_rs_bar_ts": None,
|
|
}
|
|
base.update(kwargs)
|
|
return base
|
|
|
|
|
|
class TestRsLevelAlertClaim(unittest.TestCase):
|
|
def setUp(self):
|
|
self.conn = sqlite3.connect(":memory:")
|
|
self.conn.execute(
|
|
"CREATE TABLE key_monitors ("
|
|
"id INTEGER PRIMARY KEY, notification_count INTEGER DEFAULT 0, "
|
|
"direction TEXT, last_notified_at TEXT, last_rs_bar_ts INTEGER)"
|
|
)
|
|
self.conn.execute(
|
|
"INSERT INTO key_monitors (id, notification_count, direction) VALUES (1, 0, 'watch')"
|
|
)
|
|
self.conn.commit()
|
|
|
|
def test_claim_advances_once_per_index(self):
|
|
ok1 = claim_rs_level_notify(
|
|
self.conn, 1, 1, "long", "2026-06-02 00:25:00", 1000, prior_count=0
|
|
)
|
|
self.conn.commit()
|
|
self.assertTrue(ok1)
|
|
ok_dup = claim_rs_level_notify(
|
|
self.conn, 1, 1, "long", "2026-06-02 00:25:03", 1000, prior_count=0
|
|
)
|
|
self.assertFalse(ok_dup)
|
|
ok2 = claim_rs_level_notify(
|
|
self.conn, 1, 2, "long", "2026-06-02 00:30:00", 1000, prior_count=1
|
|
)
|
|
self.conn.commit()
|
|
self.assertTrue(ok2)
|
|
row = self.conn.execute(
|
|
"SELECT notification_count FROM key_monitors WHERE id=1"
|
|
).fetchone()
|
|
self.assertEqual(row[0], 2)
|
|
|
|
def test_second_push_requires_interval(self):
|
|
now = datetime(2026, 6, 2, 0, 26, 0)
|
|
row = _row(
|
|
notification_count=1,
|
|
direction="long",
|
|
last_notified_at="2026-06-02 00:25:00",
|
|
)
|
|
tick = run_rs_level_alert_tick(row, 2.18, 1000, now, default_max_notify=3, default_interval_min=5)
|
|
self.assertIsNone(tick)
|
|
later = datetime(2026, 6, 2, 0, 30, 1)
|
|
tick2 = run_rs_level_alert_tick(
|
|
row, 2.18, 1000, later, default_max_notify=3, default_interval_min=5
|
|
)
|
|
self.assertIsNotNone(tick2)
|
|
self.assertEqual(tick2["notify_index"], 2)
|
|
self.assertEqual(tick2["prior_count"], 1)
|
|
|
|
def test_notify_interval_invalid_timestamp_does_not_spam(self):
|
|
now = datetime(2026, 6, 2, 1, 0, 0)
|
|
self.assertFalse(notify_interval_elapsed("not-a-date", 5, now))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|