"""Tests for the rule engine.""" from datetime import datetime import pytest from app.models.alerts import ( AlertRules, AlertType, PrecipitationRule, SevereWeatherRule, TemperatureRule, WindRule, ) from app.models.weather import HourlyForecast, WeatherAlert, WeatherForecast from app.services.rule_engine import RuleEngine def make_hourly_forecast( temp: float = 70, precip_prob: float = 0, wind_speed: float = 5, wind_gust: float = 10, hour_offset: int = 0, ) -> HourlyForecast: """Create a test HourlyForecast.""" base_epoch = int(datetime.now().timestamp()) + (hour_offset * 3600) return HourlyForecast( datetime_str=datetime.fromtimestamp(base_epoch).strftime("%H:%M:%S"), datetime_epoch=base_epoch, temp=temp, feelslike=temp, humidity=50, precip=0, precip_prob=precip_prob, snow=0, snow_depth=0, wind_speed=wind_speed, wind_gust=wind_gust, wind_dir=180, pressure=30, visibility=10, cloud_cover=20, uv_index=5, conditions="Clear", icon="clear-day", ) class TestTemperatureRules: """Test temperature alert rules.""" def test_low_temperature_triggers_alert(self) -> None: """Test that temperature below threshold triggers alert.""" rules = AlertRules( temperature=TemperatureRule(enabled=True, below=32, above=100), precipitation=PrecipitationRule(enabled=False), wind=WindRule(enabled=False), severe_weather=SevereWeatherRule(enabled=False), ) engine = RuleEngine(rules) forecast = WeatherForecast( location="test", resolved_address="Test Location", timezone="America/Chicago", hourly_forecasts=[make_hourly_forecast(temp=28, hour_offset=1)], alerts=[], ) triggered = engine.evaluate(forecast) assert len(triggered) == 1 assert triggered[0].alert_type == AlertType.TEMPERATURE_LOW assert triggered[0].value == 28 assert triggered[0].threshold == 32 def test_high_temperature_triggers_alert(self) -> None: """Test that temperature above threshold triggers alert.""" rules = AlertRules( temperature=TemperatureRule(enabled=True, below=32, above=100), precipitation=PrecipitationRule(enabled=False), wind=WindRule(enabled=False), severe_weather=SevereWeatherRule(enabled=False), ) engine = RuleEngine(rules) forecast = WeatherForecast( location="test", resolved_address="Test Location", timezone="America/Chicago", hourly_forecasts=[make_hourly_forecast(temp=105, hour_offset=1)], alerts=[], ) triggered = engine.evaluate(forecast) assert len(triggered) == 1 assert triggered[0].alert_type == AlertType.TEMPERATURE_HIGH assert triggered[0].value == 105 def test_normal_temperature_no_alert(self) -> None: """Test that normal temperature doesn't trigger alert.""" rules = AlertRules( temperature=TemperatureRule(enabled=True, below=32, above=100), precipitation=PrecipitationRule(enabled=False), wind=WindRule(enabled=False), severe_weather=SevereWeatherRule(enabled=False), ) engine = RuleEngine(rules) forecast = WeatherForecast( location="test", resolved_address="Test Location", timezone="America/Chicago", hourly_forecasts=[make_hourly_forecast(temp=70, hour_offset=1)], alerts=[], ) triggered = engine.evaluate(forecast) assert len(triggered) == 0 def test_disabled_rule_no_alert(self) -> None: """Test that disabled rules don't trigger alerts.""" rules = AlertRules( temperature=TemperatureRule(enabled=False, below=32, above=100), precipitation=PrecipitationRule(enabled=False), wind=WindRule(enabled=False), severe_weather=SevereWeatherRule(enabled=False), ) engine = RuleEngine(rules) forecast = WeatherForecast( location="test", resolved_address="Test Location", timezone="America/Chicago", hourly_forecasts=[make_hourly_forecast(temp=20, hour_offset=1)], alerts=[], ) triggered = engine.evaluate(forecast) assert len(triggered) == 0 class TestPrecipitationRules: """Test precipitation alert rules.""" def test_high_precipitation_triggers_alert(self) -> None: """Test that high precipitation probability triggers alert.""" rules = AlertRules( temperature=TemperatureRule(enabled=False), precipitation=PrecipitationRule(enabled=True, probability_above=70), wind=WindRule(enabled=False), severe_weather=SevereWeatherRule(enabled=False), ) engine = RuleEngine(rules) forecast = WeatherForecast( location="test", resolved_address="Test Location", timezone="America/Chicago", hourly_forecasts=[make_hourly_forecast(precip_prob=85, hour_offset=1)], alerts=[], ) triggered = engine.evaluate(forecast) assert len(triggered) == 1 assert triggered[0].alert_type == AlertType.PRECIPITATION assert triggered[0].value == 85 class TestWindRules: """Test wind alert rules.""" def test_high_wind_speed_triggers_alert(self) -> None: """Test that high wind speed triggers alert.""" rules = AlertRules( temperature=TemperatureRule(enabled=False), precipitation=PrecipitationRule(enabled=False), wind=WindRule(enabled=True, speed_above=25, gust_above=40), severe_weather=SevereWeatherRule(enabled=False), ) engine = RuleEngine(rules) forecast = WeatherForecast( location="test", resolved_address="Test Location", timezone="America/Chicago", hourly_forecasts=[make_hourly_forecast(wind_speed=30, hour_offset=1)], alerts=[], ) triggered = engine.evaluate(forecast) assert len(triggered) == 1 assert triggered[0].alert_type == AlertType.WIND_SPEED def test_high_wind_gust_triggers_alert(self) -> None: """Test that high wind gust triggers alert.""" rules = AlertRules( temperature=TemperatureRule(enabled=False), precipitation=PrecipitationRule(enabled=False), wind=WindRule(enabled=True, speed_above=25, gust_above=40), severe_weather=SevereWeatherRule(enabled=False), ) engine = RuleEngine(rules) forecast = WeatherForecast( location="test", resolved_address="Test Location", timezone="America/Chicago", hourly_forecasts=[make_hourly_forecast(wind_gust=50, hour_offset=1)], alerts=[], ) triggered = engine.evaluate(forecast) assert len(triggered) == 1 assert triggered[0].alert_type == AlertType.WIND_GUST def test_both_wind_conditions_trigger_two_alerts(self) -> None: """Test that high wind speed and gust trigger separate alerts.""" rules = AlertRules( temperature=TemperatureRule(enabled=False), precipitation=PrecipitationRule(enabled=False), wind=WindRule(enabled=True, speed_above=25, gust_above=40), severe_weather=SevereWeatherRule(enabled=False), ) engine = RuleEngine(rules) forecast = WeatherForecast( location="test", resolved_address="Test Location", timezone="America/Chicago", hourly_forecasts=[ make_hourly_forecast(wind_speed=30, wind_gust=50, hour_offset=1) ], alerts=[], ) triggered = engine.evaluate(forecast) assert len(triggered) == 2 alert_types = {t.alert_type for t in triggered} assert AlertType.WIND_SPEED in alert_types assert AlertType.WIND_GUST in alert_types class TestSevereWeatherRules: """Test severe weather alert forwarding.""" def test_severe_weather_alert_forwarded(self) -> None: """Test that severe weather alerts from API are forwarded.""" rules = AlertRules( temperature=TemperatureRule(enabled=False), precipitation=PrecipitationRule(enabled=False), wind=WindRule(enabled=False), severe_weather=SevereWeatherRule(enabled=True), ) engine = RuleEngine(rules) api_alert = WeatherAlert( event="Tornado Warning", headline="Tornado Warning in effect", description="A tornado has been spotted in the area.", onset="2024-01-15T12:00:00", ends="2024-01-15T14:00:00", id="NWS-123", language="en", link="https://weather.gov/alerts", ) forecast = WeatherForecast( location="test", resolved_address="Test Location", timezone="America/Chicago", hourly_forecasts=[], alerts=[api_alert], ) triggered = engine.evaluate(forecast) assert len(triggered) == 1 assert triggered[0].alert_type == AlertType.SEVERE_WEATHER assert "Tornado Warning" in triggered[0].title class TestMultipleHours: """Test evaluation across multiple forecast hours.""" def test_multiple_hours_can_trigger_alerts(self) -> None: """Test that alerts can trigger for different hours.""" rules = AlertRules( temperature=TemperatureRule(enabled=True, below=32, above=100), precipitation=PrecipitationRule(enabled=False), wind=WindRule(enabled=False), severe_weather=SevereWeatherRule(enabled=False), ) engine = RuleEngine(rules) forecast = WeatherForecast( location="test", resolved_address="Test Location", timezone="America/Chicago", hourly_forecasts=[ make_hourly_forecast(temp=28, hour_offset=1), make_hourly_forecast(temp=70, hour_offset=2), make_hourly_forecast(temp=25, hour_offset=3), ], alerts=[], ) triggered = engine.evaluate(forecast) assert len(triggered) == 2 assert all(t.alert_type == AlertType.TEMPERATURE_LOW for t in triggered)