修复支撑阻力企业微信重复

tests/test_key_monitor_rs_alert.py

@@ -0,0 +1,86 @@
+"""阻力/支撑提醒：占位与间隔防重复推送。"""
+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()