diff --git a/bmspy/ups.py b/bmspy/ups.py index 05a2ea5..4643259 100644 --- a/bmspy/ups.py +++ b/bmspy/ups.py @@ -269,9 +269,10 @@ def main() -> None: history.popleft() if len(history) > 3: - comparison_1 = history[1] - comparison_2 = history[2] - comparison_3 = history[3] + # The newest reading is last. Compare it with the readings before it. + comparison_1 = history[-2] + comparison_2 = history[-3] + comparison_3 = history[-4] else: if debug > 1: print("Not enough readings, sleeping...") diff --git a/tests/test_ups.py b/tests/test_ups.py index d7cc0ee..a6d638e 100644 --- a/tests/test_ups.py +++ b/tests/test_ups.py @@ -414,17 +414,15 @@ class TestUpsMain: def test_main_discharge_alert(self): """When current goes negative, discharge alert should be sent. - Discharge alert requires: current < 0 AND h1 < 0 AND h2 >= 0. - After 5 reads with seq [+1, -1, +1, +1, -1]: - h=[+1,-1,+1,+1,-1], current=-1, h1=-1, h2=+1 → triggers alert. + The alert needs two negative readings after a reading that is not negative. """ ups_mod.alert_sent = False data_seq = [ - self._make_data(1.0, 0.90), # r1: +1 - self._make_data(-1.0, 0.89), # r2: -1 (will be h1) - self._make_data(1.0, 0.90), # r3: +1 (will be h2) - self._make_data(1.0, 0.90), # r4: +1 - self._make_data(-1.0, 0.88), # r5: -1 (current) → alert! + self._make_data(1.0, 0.90), + self._make_data(1.0, 0.90), + self._make_data(1.0, 0.90), # the last reading with mains power + self._make_data(-1.0, 0.89), # the first discharging reading + self._make_data(-1.0, 0.88), # the second one: alert ] mock_email = MagicMock() @@ -436,23 +434,22 @@ class TestUpsMain: patch("bmspy.ups.handle_shutdown"), \ patch("bmspy.ups.handle_email", mock_email): ups_mod.main() - mock_email.assert_called() + mock_email.assert_called_once() + assert mock_email.call_args.kwargs["level"] == "Power loss alert" def test_main_power_regained_alert(self): """When current goes from negative to positive, recovery alert sent. - Recovery requires: current >= 0 AND h1 >= 0 AND h2 < 0. - After 5 reads with seq [-1, +1, -1, +1, +1]: - h=[-1,+1,-1,+1,+1], current=+1, h1=+1, h2=-1 → recovery! + The recovery needs two readings that are not negative after a negative one. """ ups_mod.alert_sent = True ups_mod.scheduled_shutdown = False data_seq = [ - self._make_data(-1.0, 0.80), # r1: -1 - self._make_data(1.0, 0.81), # r2: +1 (will be h1) - self._make_data(-1.0, 0.79), # r3: -1 (will be h2) - self._make_data(1.0, 0.81), # r4: +1 - self._make_data(1.0, 0.82), # r5: +1 (current) → recovery! + self._make_data(-1.0, 0.80), + self._make_data(-1.0, 0.80), + self._make_data(-1.0, 0.79), # the last discharging reading + self._make_data(1.0, 0.81), # the first reading with mains power + self._make_data(1.0, 0.82), # the second one: recovery ] mock_email = MagicMock() @@ -465,7 +462,58 @@ class TestUpsMain: patch("bmspy.ups.handle_shutdown", mock_shutdown), \ patch("bmspy.ups.handle_email", mock_email): ups_mod.main() - mock_email.assert_called() + mock_email.assert_called_once() + assert mock_email.call_args.kwargs["level"] == "Recovery alert" + mock_shutdown.assert_called_once_with(action="cancel", debug=0) + + def _events(self, readings, argv=None): + """Run main() over (current, charge ratio) readings, 30 seconds apart. + + Return (reading number, event) for each email and each shutdown call. + """ + sequence = [self._make_data(*reading) for reading in readings] + for data, (current, _) in zip(sequence, readings): + if current is None: + del data["testups"]._fields["bms_current_amps"] + events, number = [], [0] + + def read(*args, **kwargs): + if number[0] > len(sequence): + raise StopIteration("done") + number[0] += 1 + # The first call is the read at startup. It is not in the history. + return sequence[max(number[0] - 2, 0)] + + with patch("sys.argv", ["bmspy-ups"] + (argv or [])), \ + patch("bmspy.ups.client.read_data", side_effect=read), \ + patch("bmspy.ups.client.handle_registration"), \ + patch("bmspy.ups.time.sleep"), \ + patch("bmspy.ups.handle_shutdown", + side_effect=lambda action, **kw: events.append((number[0] - 1, action))), \ + patch("bmspy.ups.handle_email", + side_effect=lambda text, level, **kw: events.append((number[0] - 1, level))): + with pytest.raises(StopIteration): + ups_mod.main() + return events + + def test_power_failure_and_recovery_are_reported_at_the_second_reading(self): + """Mains power fails at reading 9 and returns at reading 15. + + Ten readings fill the history. The comparison must use the newest + readings, not the oldest ones in the history. + """ + readings = [(1.0, 0.95)] * 8 + [(-5.0, 0.94)] * 6 + [(1.0, 0.93)] * 12 + assert self._events(readings) == [ + (10, "Power loss alert"), (16, "cancel"), (16, "Recovery alert"), + ] + + def test_shutdown_starts_at_the_second_reading_below_critical(self): + """The charge falls below the critical threshold at reading 13.""" + readings = [(-5.0, 0.95)] * 12 + [(-5.0, 0.20)] * 3 + assert self._events(readings) == [ + (14, "shutdown"), (14, "Critical alert"), (15, "shutdown"), + ] + def test_main_not_enough_history_debug_2(self, capsys): """With debug=2 and < 4 readings, prints 'not enough readings'.""" @@ -500,10 +548,10 @@ class TestUpsMain: """With debug=1, discharge alert message is printed.""" ups_mod.alert_sent = False data_seq = [ + self._make_data(1.0, 0.90), + self._make_data(1.0, 0.90), self._make_data(1.0, 0.90), self._make_data(-1.0, 0.89), - self._make_data(1.0, 0.90), - self._make_data(1.0, 0.90), self._make_data(-1.0, 0.88), ] with patch("sys.argv", ["bmspy-ups", "-v"]): @@ -522,7 +570,7 @@ class TestUpsMain: ups_mod.alert_sent = True data_seq = [ self._make_data(-1.0, 0.80), - self._make_data(1.0, 0.81), + self._make_data(-1.0, 0.80), self._make_data(-1.0, 0.79), self._make_data(1.0, 0.81), self._make_data(1.0, 0.82),