Spaces:
Configuration error
Configuration error
File size: 11,831 Bytes
83bcbe3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 | """
tests/test_alert_engine.py — Unit tests for the alert engine state machine.
Run with:
python -m pytest tests/test_alert_engine.py -v
"""
import sys
import os
from datetime import datetime, timezone, timedelta
from unittest.mock import patch
# Add project root to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from alert_engine import AlertEngine, AlertEvent, RangeState
import config
class TestAlertEngine:
"""Tests for AlertEngine state transitions and dedup logic."""
def setup_method(self):
"""Fresh engine + captured alerts for each test.
Cooldown is patched to 0 for most tests so re-entry is always allowed.
Use TestAlertEngineCooldown for cooldown-specific tests.
"""
self.alerts: list[AlertEvent] = []
self.engine = AlertEngine(on_alert=lambda e: self.alerts.append(e))
# ── Basic state transitions ──────────────────────────────────────────
def test_outside_to_inside_triggers_alert(self):
"""Price entering the range should fire exactly one alert."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.on_price_update("BTC", 85000) # enters range
assert len(self.alerts) == 1
assert self.alerts[0].asset_name == "BTC"
assert self.alerts[0].price == 85000
def test_inside_to_inside_no_duplicate(self):
"""Price staying inside range should NOT fire additional alerts."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.on_price_update("BTC", 85000) # enter → alert
self.engine.on_price_update("BTC", 86000) # still inside → no alert
self.engine.on_price_update("BTC", 89999) # still inside → no alert
assert len(self.alerts) == 1
def test_inside_to_outside_resets(self):
"""Price exiting range should reset state so re-entry triggers again."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.on_price_update("BTC", 85000) # enter → alert
self.engine.on_price_update("BTC", 95000) # exit → reset
self.engine.on_price_update("BTC", 82000) # re-enter → alert again
assert len(self.alerts) == 2
def test_outside_to_outside_no_alert(self):
"""Price staying outside should produce no alerts."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.on_price_update("BTC", 50000)
self.engine.on_price_update("BTC", 55000)
self.engine.on_price_update("BTC", 95000)
assert len(self.alerts) == 0
# ── Boundary conditions ──────────────────────────────────────────────
def test_price_at_min_boundary_is_inside(self):
"""Price exactly at min_price should be considered inside."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("ETH", 1700, 2200)
self.engine.on_price_update("ETH", 1700)
assert len(self.alerts) == 1
def test_price_at_max_boundary_is_inside(self):
"""Price exactly at max_price should be considered inside."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("ETH", 1700, 2200)
self.engine.on_price_update("ETH", 2200)
assert len(self.alerts) == 1
def test_price_just_below_min_is_outside(self):
"""Price just below min should not trigger."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("ETH", 1700, 2200)
self.engine.on_price_update("ETH", 1699.99)
assert len(self.alerts) == 0
def test_price_just_above_max_is_outside(self):
"""Price just above max should not trigger."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("ETH", 1700, 2200)
self.engine.on_price_update("ETH", 2200.01)
assert len(self.alerts) == 0
# ── Asset management ─────────────────────────────────────────────────
def test_add_asset(self):
self.engine.add_asset("GOLD", 2900, 3200)
tracked = self.engine.get_tracked_assets()
assert "GOLD" in tracked
assert tracked["GOLD"] == (2900, 3200)
def test_remove_asset(self):
self.engine.add_asset("GOLD", 2900, 3200)
assert self.engine.remove_asset("GOLD") is True
assert self.engine.remove_asset("GOLD") is False # already removed
assert "GOLD" not in self.engine.get_tracked_assets()
def test_update_range_resets_state(self):
"""Updating an asset's range should reset state to OUTSIDE."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.on_price_update("BTC", 85000) # inside → INSIDE state
assert len(self.alerts) == 1
# Update range — state resets to OUTSIDE
self.engine.add_asset("BTC", 80000, 88000)
self.engine.on_price_update("BTC", 85000) # re-enters → alert again
assert len(self.alerts) == 2
def test_untracked_asset_ignored(self):
"""Price update for an untracked asset should be silently ignored."""
self.engine.on_price_update("UNKNOWN", 100)
assert len(self.alerts) == 0
def test_case_insensitive_names(self):
"""Asset names should be case-insensitive."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("btc", 80000, 90000)
self.engine.on_price_update("Btc", 85000)
assert len(self.alerts) == 1
assert self.alerts[0].asset_name == "BTC"
# ── Multiple assets ──────────────────────────────────────────────────
def test_multiple_assets_independent(self):
"""Alerts for different assets should be independent."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.add_asset("ETH", 1700, 2200)
self.engine.on_price_update("BTC", 85000) # BTC enters
self.engine.on_price_update("ETH", 1500) # ETH outside
self.engine.on_price_update("ETH", 2000) # ETH enters
assert len(self.alerts) == 2
assert self.alerts[0].asset_name == "BTC"
assert self.alerts[1].asset_name == "ETH"
def test_get_asset_details(self):
"""get_asset_details should return comprehensive info."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.on_price_update("BTC", 85000)
details = self.engine.get_asset_details()
assert len(details) == 1
d = details[0]
assert d["name"] == "BTC"
assert d["state"] == "INSIDE"
assert d["last_price"] == 85000
assert d["last_alert"] is not None
# ── Alert history ────────────────────────────────────────────────────
def test_alert_stored_in_history(self):
"""Triggered alerts should appear in get_alert_history()."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.on_price_update("BTC", 85000)
history = self.engine.get_alert_history(limit=10)
assert len(history) == 1
assert history[0].asset_name == "BTC"
def test_history_newest_first(self):
"""get_alert_history should return newest events first."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.add_asset("ETH", 1700, 2200)
self.engine.on_price_update("BTC", 85000)
self.engine.on_price_update("ETH", 2000)
history = self.engine.get_alert_history(limit=10)
assert history[0].asset_name == "ETH"
assert history[1].asset_name == "BTC"
# ── Pause / Resume ───────────────────────────────────────────────────
def test_pause_suppresses_alerts(self):
"""Paused engine should not fire alert callbacks."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.pause_all()
self.engine.on_price_update("BTC", 85000)
assert len(self.alerts) == 0
def test_resume_re_enables_alerts(self):
"""Resuming after pause should allow alerts to fire again."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.pause_all()
self.engine.on_price_update("BTC", 85000) # paused → no alert, state=INSIDE
self.engine.on_price_update("BTC", 95000) # exits → OUTSIDE
self.engine.resume_all()
self.engine.on_price_update("BTC", 85000) # re-enters → alert fires
assert len(self.alerts) == 1
class TestAlertEngineCooldown:
"""Tests specifically for the cooldown behaviour."""
def setup_method(self):
self.alerts: list[AlertEvent] = []
self.engine = AlertEngine(on_alert=lambda e: self.alerts.append(e))
def test_cooldown_blocks_rapid_reentry(self):
"""Re-entry within cooldown period should NOT fire another alert."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 300):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.on_price_update("BTC", 85000) # alert fires
self.engine.on_price_update("BTC", 95000) # exits
self.engine.on_price_update("BTC", 85000) # re-enters — blocked by cooldown
assert len(self.alerts) == 1
def test_cooldown_zero_allows_immediate_reentry(self):
"""With cooldown=0, re-entry should always fire."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 0):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.on_price_update("BTC", 85000) # alert
self.engine.on_price_update("BTC", 95000) # exit
self.engine.on_price_update("BTC", 85000) # re-enter → alert
assert len(self.alerts) == 2
def test_cooldown_expired_allows_reentry(self):
"""After cooldown expires, re-entry should fire again."""
with patch.object(config, "ALERT_COOLDOWN_SECONDS", 300):
self.engine.add_asset("BTC", 80000, 90000)
self.engine.on_price_update("BTC", 85000) # alert fires
# Manually backdate the last_alert_time past cooldown
asset = self.engine._assets["BTC"]
asset.last_alert_time = datetime.now(timezone.utc) - timedelta(seconds=301)
self.engine.on_price_update("BTC", 95000) # exits
self.engine.on_price_update("BTC", 85000) # re-enters after cooldown
assert len(self.alerts) == 2
|