Spaces:
Configuration error
Configuration error
| """ | |
| 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 | |