tim/bmspy
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Compare the newest readings in bmspy-ups

Browse Source 
The loop compared the current reading with history[1], history[2], and history[3]. The history holds the oldest reading first, so these were three of the oldest readings. A plain power failure sent no email, and the shutdown started 4 minutes late. The loop now uses the three readings before the current one.

Four tests used reading sequences that matched the old comparison. They now use plain sequences.
This commit is contained in:
tim
2026-10-10 10:59:35 +02:00
parent 9ab1d909ed
commit 3bc86b85a2
2 changed files with 73 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_ups.py
+69 -21
View File
@@ -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),