aps2mqtt/test_control_alerts.py at main · MathildeDec/aps2mqtt · GitHub

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
"""Test ECU control and alert functionality"""
import pytest
from datetime import datetime, timedelta
from aps2mqtt.control import ECUControl
from aps2mqtt.alerts import AlertManager, Alert
from aps2mqtt.models import ECUData, InverterData


def test_ecu_control_initialization():
    """Test ECU control initialization"""
    control = ECUControl("192.168.1.100", "TestSSID", "TestPassword")
    assert control.ecu_ip == "192.168.1.100"
    assert control.wifi_ssid == "TestSSID"
    assert control.wifi_password == "TestPassword"


def test_ecu_compatibility_check():
    """Test ECU compatibility detection"""
    control = ECUControl("192.168.1.100")

    # ECU-R-Pro (2162)
    assert control.is_compatible_ecu("216200063835") == True

    # ECU-C (215)
    assert control.is_compatible_ecu("215012345678") == True

    # Old ECU-R (2160)
    assert control.is_compatible_ecu("216012345678") == False

    # ECU-B
    assert control.is_compatible_ecu("210012345678") == False


def test_alert_manager_initialization():
    """Test alert manager initialization"""
    manager = AlertManager(
        offline_threshold_minutes=5,
        low_production_threshold_percent=30.0,
        high_temperature_threshold=65.0
    )
    assert manager.offline_threshold == timedelta(minutes=5)
    assert manager.low_production_threshold == 0.30
    assert manager.high_temperature_threshold == 65.0


def test_inverter_offline_alert():
    """Test inverter offline alert generation"""
    manager = AlertManager(offline_threshold_minutes=5)

    # Create ECU data with offline inverter
    inverter1 = InverterData(
        uid="123456789012",
        online=False,
        current_software_version="1.0.0",
        signal=0
    )

    inverter2 = InverterData(
        uid="123456789013",
        online=True,
        current_software_version="1.0.0",
        signal=95,
        power=[100, 105]
    )

    ecu_data = ECUData(
        ecu_id="216200063835",
        timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        today_energy=10.5,
        lifetime_energy=5000.0,
        current_power=205,
        qty_of_inverters=2,
        qty_of_online_inverters=1,
        firmware="1.2.0",
        inverters=[inverter1, inverter2]
    )

    # Check alerts
    alerts = manager.check_alerts(ecu_data)

    # Should have at least one alert for offline inverter
    assert len(alerts) > 0
    offline_alerts = [a for a in alerts if a.alert_type == 'inverter_offline']
    assert len(offline_alerts) > 0


def test_high_temperature_alert():
    """Test high temperature alert"""
    manager = AlertManager(high_temperature_threshold=65.0)

    # Create inverter with high temperature
    inverter = InverterData(
        uid="123456789012",
        online=True,
        current_software_version="1.0.0",
        signal=95,
        temperature=70,  # Above threshold
        power=[100, 105]
    )

    ecu_data = ECUData(
        ecu_id="216200063835",
        timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        today_energy=10.5,
        lifetime_energy=5000.0,
        current_power=205,
        qty_of_inverters=1,
        qty_of_online_inverters=1,
        firmware="1.2.0",
        inverters=[inverter]
    )

    alerts = manager.check_alerts(ecu_data)
    temp_alerts = [a for a in alerts if a.alert_type == 'high_temperature']
    assert len(temp_alerts) > 0
    assert temp_alerts[0].value == 70


def test_alert_summary():
    """Test alert summary generation"""
    manager = AlertManager()

    # Add some test alerts manually
    manager.active_alerts = [
        Alert('inverter_offline', 'critical', 'Test 1', datetime.now()),
        Alert('high_temperature', 'warning', 'Test 2', datetime.now()),
        Alert('inverter_offline', 'warning', 'Test 3', datetime.now())
    ]

    summary = manager.get_alert_summary()
    assert summary['total'] == 3
    assert summary['critical'] == 1
    assert summary['warning'] == 2
    assert summary['inverter_offline'] == 2
    assert summary['high_temperature'] == 1


def test_alert_filtering():
    """Test getting alerts by severity"""
    manager = AlertManager()

    manager.active_alerts = [
        Alert('inverter_offline', 'critical', 'Test 1', datetime.now()),
        Alert('high_temperature', 'warning', 'Test 2', datetime.now()),
        Alert('low_production', 'warning', 'Test 3', datetime.now())
    ]

    critical_alerts = manager.get_active_alerts(severity='critical')
    assert len(critical_alerts) == 1

    warning_alerts = manager.get_active_alerts(severity='warning')
    assert len(warning_alerts) == 2


if __name__ == "__main__":
    pytest.main([__file__, "-v"])