320 lines
10 KiB
Python
320 lines
10 KiB
Python
"""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)
|