tim
98fd08e39b
The charge tests and the power tests were one if/elif chain. If mains power returned while the charge was below the warning threshold, the loop did not report it, did not cancel the shutdown, and did not clear the flags for the next power failure. The power tests now run first, on their own. The charge tests apply only while the battery discharges for two consecutive readings, so a low battery on mains power starts no shutdown. A device that reports no battery current is judged by its charge alone.
755 lines
34 KiB
Python
755 lines
34 KiB
Python
import pytest
|
|
import smtplib
|
|
from unittest.mock import call, patch, MagicMock
|
|
|
|
import bmspy.ups as ups_mod
|
|
from bmspy.classes import BMSScalarField, BMSMultiField, BMSInfoField, UPS
|
|
from bmspy.ups import _get_field_value, _resolve_ups_device, handle_shutdown, handle_email
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures: reset module-level globals between tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_globals():
|
|
ups_mod.scheduled_shutdown = False
|
|
ups_mod.critical_sent = False
|
|
ups_mod.warning_sent = False
|
|
ups_mod.alert_sent = False
|
|
yield
|
|
ups_mod.scheduled_shutdown = False
|
|
ups_mod.critical_sent = False
|
|
ups_mod.warning_sent = False
|
|
ups_mod.alert_sent = False
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _configured_devices():
|
|
"""The main() tests use one configured device unless they patch this themselves."""
|
|
with patch("bmspy.ups.client.list_devices", return_value=["testups"]):
|
|
yield
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_field_value
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetFieldValue:
|
|
def _ups(self):
|
|
return UPS.from_dict({
|
|
"bms_current_amps": {
|
|
"help": "Current",
|
|
"raw_value": -2.5,
|
|
"value": "-2.50",
|
|
"units": "A",
|
|
},
|
|
"bms_capacity_charge_ratio": {
|
|
"help": "Percent Charge",
|
|
"raw_value": 0.75,
|
|
"value": "0.75",
|
|
"units": "%",
|
|
},
|
|
"bms_date": {
|
|
"help": "Date",
|
|
"info": "2023-01-15",
|
|
},
|
|
"bms_cells": {
|
|
"help": "Cells",
|
|
"label": "cell",
|
|
"raw_values": {1: 3.6},
|
|
"values": {1: "3.600"},
|
|
"units": "V",
|
|
},
|
|
})
|
|
|
|
def test_found_scalar_field(self):
|
|
ups = self._ups()
|
|
assert _get_field_value(ups, "bms_current_amps") == pytest.approx(-2.5)
|
|
|
|
def test_found_charge_ratio(self):
|
|
ups = self._ups()
|
|
assert _get_field_value(ups, "bms_capacity_charge_ratio") == pytest.approx(0.75)
|
|
|
|
def test_field_not_found_returns_none(self):
|
|
ups = self._ups()
|
|
assert _get_field_value(ups, "nonexistent_field") is None
|
|
|
|
def test_info_field_returns_none(self):
|
|
# BMSInfoField is not a scalar, should return None
|
|
ups = self._ups()
|
|
assert _get_field_value(ups, "bms_date") is None
|
|
|
|
def test_multi_field_returns_none(self):
|
|
# BMSMultiField is not a scalar, should return None
|
|
ups = self._ups()
|
|
assert _get_field_value(ups, "bms_cells") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# handle_shutdown
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHandleShutdown:
|
|
def test_shutdown_calls_os_system(self):
|
|
with patch("bmspy.ups.os.system") as mock_sys:
|
|
handle_shutdown(action="shutdown", delay=5)
|
|
mock_sys.assert_called_once_with("/sbin/shutdown 5")
|
|
|
|
def test_shutdown_sets_scheduled_flag(self):
|
|
with patch("bmspy.ups.os.system"):
|
|
handle_shutdown(action="shutdown", delay=5)
|
|
assert ups_mod.scheduled_shutdown is not False
|
|
|
|
def test_shutdown_does_not_reschedule_if_already_scheduled(self):
|
|
ups_mod.scheduled_shutdown = 9999999999.0
|
|
with patch("bmspy.ups.os.system") as mock_sys:
|
|
handle_shutdown(action="shutdown", delay=5)
|
|
mock_sys.assert_not_called()
|
|
|
|
def test_cancel_calls_os_system(self):
|
|
with patch("bmspy.ups.os.system") as mock_sys:
|
|
handle_shutdown(action="cancel")
|
|
mock_sys.assert_called_once_with("/sbin/shutdown -c")
|
|
|
|
def test_cancel_does_not_set_flag(self):
|
|
with patch("bmspy.ups.os.system"):
|
|
handle_shutdown(action="cancel")
|
|
assert ups_mod.scheduled_shutdown is False
|
|
|
|
def test_second_incident_after_cancel_schedules_again(self):
|
|
with patch("bmspy.ups.os.system") as mock_sys:
|
|
handle_shutdown(action="shutdown", delay=5)
|
|
handle_shutdown(action="cancel")
|
|
assert ups_mod.scheduled_shutdown is False
|
|
handle_shutdown(action="shutdown", delay=5)
|
|
assert [call.args[0] for call in mock_sys.call_args_list] == [
|
|
"/sbin/shutdown 5", "/sbin/shutdown -c", "/sbin/shutdown 5",
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# handle_email
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHandleEmail:
|
|
def _mock_smtp(self):
|
|
mock_server = MagicMock()
|
|
mock_smtp_class = MagicMock()
|
|
mock_smtp_class.return_value.__enter__ = MagicMock(return_value=mock_server)
|
|
mock_smtp_class.return_value.__exit__ = MagicMock(return_value=False)
|
|
return mock_smtp_class, mock_server
|
|
|
|
def test_basic_send(self):
|
|
mock_smtp_class, mock_server = self._mock_smtp()
|
|
with patch("bmspy.ups.smtplib.SMTP", mock_smtp_class):
|
|
handle_email(
|
|
text="test message",
|
|
level="Alert",
|
|
recipient="user@example.com",
|
|
)
|
|
mock_server.sendmail.assert_called_once()
|
|
|
|
def test_ssl_port_triggers_starttls(self):
|
|
mock_smtp_class, mock_server = self._mock_smtp()
|
|
with patch("bmspy.ups.smtplib.SMTP", mock_smtp_class), \
|
|
patch("bmspy.ups.ssl.create_default_context") as mock_ssl:
|
|
handle_email(
|
|
text="test",
|
|
level="Alert",
|
|
recipient="user@example.com",
|
|
port=465,
|
|
)
|
|
mock_server.starttls.assert_called_once()
|
|
|
|
def test_port_587_also_triggers_starttls(self):
|
|
mock_smtp_class, mock_server = self._mock_smtp()
|
|
with patch("bmspy.ups.smtplib.SMTP", mock_smtp_class), \
|
|
patch("bmspy.ups.ssl.create_default_context"):
|
|
handle_email(
|
|
text="test",
|
|
level="Alert",
|
|
recipient="user@example.com",
|
|
port=587,
|
|
)
|
|
mock_server.starttls.assert_called_once()
|
|
|
|
def test_no_ssl_no_starttls(self):
|
|
mock_smtp_class, mock_server = self._mock_smtp()
|
|
with patch("bmspy.ups.smtplib.SMTP", mock_smtp_class):
|
|
handle_email(
|
|
text="test",
|
|
level="Alert",
|
|
recipient="user@example.com",
|
|
port=25,
|
|
)
|
|
mock_server.starttls.assert_not_called()
|
|
|
|
def test_with_credentials_calls_login(self):
|
|
mock_smtp_class, mock_server = self._mock_smtp()
|
|
with patch("bmspy.ups.smtplib.SMTP", mock_smtp_class):
|
|
handle_email(
|
|
text="test",
|
|
level="Alert",
|
|
recipient="user@example.com",
|
|
mailuser="user",
|
|
mailpass="pass",
|
|
)
|
|
mock_server.login.assert_called_once_with("user", "pass")
|
|
|
|
def test_without_credentials_no_login(self):
|
|
mock_smtp_class, mock_server = self._mock_smtp()
|
|
with patch("bmspy.ups.smtplib.SMTP", mock_smtp_class):
|
|
handle_email(
|
|
text="test",
|
|
level="Alert",
|
|
recipient="user@example.com",
|
|
)
|
|
mock_server.login.assert_not_called()
|
|
|
|
def test_recipient_without_at_gets_hostname_appended(self):
|
|
mock_smtp_class, mock_server = self._mock_smtp()
|
|
with patch("bmspy.ups.smtplib.SMTP", mock_smtp_class), \
|
|
patch("bmspy.ups.socket.gethostname", return_value="myhost"):
|
|
handle_email(
|
|
text="test",
|
|
level="Alert",
|
|
recipient="root",
|
|
)
|
|
# sendmail recipient should be root@myhost
|
|
args = mock_server.sendmail.call_args[0]
|
|
assert args[1] == "root@myhost"
|
|
|
|
def test_full_address_is_not_changed(self):
|
|
mock_smtp_class, mock_server = self._mock_smtp()
|
|
with patch("bmspy.ups.smtplib.SMTP", mock_smtp_class), \
|
|
patch("bmspy.ups.socket.gethostname", return_value="myhost"):
|
|
handle_email(text="test", level="Alert", recipient="user@example.com")
|
|
assert mock_server.sendmail.call_args[0][1] == "user@example.com"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _resolve_ups_device
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _simple_ups_data(*names: str) -> dict[str, UPS]:
|
|
return {name: UPS.from_dict({
|
|
"bms_current_amps": {"help": "A", "raw_value": 0.0, "value": "0.00", "units": "A"},
|
|
}) for name in names}
|
|
|
|
|
|
class TestResolveUpsDevice:
|
|
def test_single_device_no_request_returns_it(self):
|
|
data = _simple_ups_data("myups")
|
|
assert _resolve_ups_device(data, None) == "myups"
|
|
|
|
def test_requested_device_found(self):
|
|
data = _simple_ups_data("ups1", "ups2")
|
|
assert _resolve_ups_device(data, "ups2") == "ups2"
|
|
|
|
def test_requested_device_not_found_returns_none(self, capsys):
|
|
data = _simple_ups_data("ups1")
|
|
result = _resolve_ups_device(data, "missing")
|
|
assert result is None
|
|
assert "missing" in capsys.readouterr().out
|
|
|
|
def test_multiple_devices_no_request_returns_none(self, capsys):
|
|
data = _simple_ups_data("ups1", "ups2")
|
|
result = _resolve_ups_device(data, None)
|
|
assert result is None
|
|
out = capsys.readouterr().out
|
|
assert "ups1" in out or "ups2" in out
|
|
|
|
def test_empty_data_returns_none(self, capsys):
|
|
result = _resolve_ups_device({}, None)
|
|
assert result is None
|
|
assert "no UPS" in capsys.readouterr().out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ups main() - comprehensive loop testing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_ups(current_amps: float, charge_ratio: float) -> UPS:
|
|
"""Build a UPS with bms_current_amps and bms_capacity_charge_ratio."""
|
|
return UPS.from_dict({
|
|
"bms_current_amps": {
|
|
"help": "Current",
|
|
"raw_value": current_amps,
|
|
"value": str(current_amps),
|
|
"units": "A",
|
|
},
|
|
"bms_capacity_charge_ratio": {
|
|
"help": "Charge Ratio",
|
|
"raw_value": charge_ratio,
|
|
"value": str(charge_ratio),
|
|
"units": "%",
|
|
},
|
|
})
|
|
|
|
|
|
class TestUpsMain:
|
|
"""Test ups.main() loop behavior by running a limited number of iterations."""
|
|
|
|
def _run_main_with_data(self, data_sequence, argv=None):
|
|
"""Run main() with a sequence of UPS data, stopping when data runs out.
|
|
|
|
The first call to read_data is the device-discovery call and returns
|
|
data_sequence[0]. Subsequent calls consume data_sequence in order.
|
|
"""
|
|
# discovery call returns data_sequence[0], then loop calls follow
|
|
discovery_done = [False]
|
|
loop_count = [0]
|
|
|
|
def _read_data(*args, **kwargs):
|
|
if not discovery_done[0]:
|
|
discovery_done[0] = True
|
|
return data_sequence[0]
|
|
if loop_count[0] >= len(data_sequence):
|
|
raise StopIteration("done")
|
|
result = data_sequence[loop_count[0]]
|
|
loop_count[0] += 1
|
|
return result
|
|
|
|
with patch("sys.argv", ["bmspy-ups"] + (argv or [])):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=_read_data), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.os.system"):
|
|
ups_mod.main()
|
|
return loop_count[0]
|
|
|
|
def _make_data(self, current_amps, charge_ratio):
|
|
return {"testups": _make_ups(current_amps, charge_ratio)}
|
|
|
|
def test_main_history_pruned_at_10(self):
|
|
"""History is pruned to max 10 items."""
|
|
# Send 15 readings to ensure pruning happens
|
|
data_seq = [self._make_data(0.0, 0.80)] * 15
|
|
|
|
with patch("sys.argv", ["bmspy-ups"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.handle_email"):
|
|
ups_mod.main()
|
|
|
|
def test_main_runs_with_not_enough_history(self):
|
|
"""With < 4 readings, should print 'not enough readings' and continue."""
|
|
data = [self._make_data(0.0, 0.95)] * 5
|
|
self._run_main_with_data(data)
|
|
|
|
def test_main_debug_2_prints_not_enough_readings(self, capsys):
|
|
data = [self._make_data(0.0, 0.95)] * 5
|
|
with patch("sys.argv", ["bmspy-ups", "-v", "-v"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data[0]] + data + [StopIteration]),\
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.os.system"):
|
|
ups_mod.main()
|
|
|
|
def test_main_below_critical_threshold_triggers_shutdown(self):
|
|
"""When charge drops below critical threshold, shutdown should be triggered."""
|
|
# Need 5+ readings (>3 in history) to trigger comparisons
|
|
ups_mod.scheduled_shutdown = False
|
|
ups_mod.critical_sent = False
|
|
data_seq = []
|
|
# Populate history with 5 readings all at 20% (below 30% critical)
|
|
for _ in range(5):
|
|
data_seq.append(self._make_data(-1.0, 0.20))
|
|
|
|
mock_shutdown = MagicMock()
|
|
mock_email = MagicMock()
|
|
with patch("sys.argv", ["bmspy-ups", "--critical", "30"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown", mock_shutdown), \
|
|
patch("bmspy.ups.handle_email", mock_email):
|
|
ups_mod.main()
|
|
mock_shutdown.assert_called()
|
|
|
|
def test_main_below_warning_threshold_sends_email(self):
|
|
"""When charge drops below warning threshold, email alert should be sent."""
|
|
ups_mod.scheduled_shutdown = False
|
|
ups_mod.warning_sent = False
|
|
data_seq = []
|
|
# 5 readings at 60% (below 75% warning, above 30% critical)
|
|
for _ in range(5):
|
|
data_seq.append(self._make_data(-1.0, 0.60))
|
|
|
|
mock_email = MagicMock()
|
|
with patch("sys.argv", ["bmspy-ups", "--warning", "75", "--critical", "30"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.handle_email", mock_email):
|
|
ups_mod.main()
|
|
mock_email.assert_called()
|
|
|
|
def test_main_mail_port_option_is_a_number(self):
|
|
"""A text port never equals 587, so the STARTTLS check would not match."""
|
|
ups_mod.warning_sent = False
|
|
data_seq = [self._make_data(-1.0, 0.60)] * 5
|
|
mock_smtp = MagicMock()
|
|
server = mock_smtp.return_value.__enter__.return_value
|
|
with patch("sys.argv", ["bmspy-ups", "--port", "587"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.smtplib.SMTP", mock_smtp):
|
|
ups_mod.main()
|
|
mock_smtp.assert_called_with("localhost", 587)
|
|
server.starttls.assert_called()
|
|
|
|
def test_main_discharge_alert(self):
|
|
"""When current goes negative, discharge alert should be sent.
|
|
|
|
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),
|
|
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()
|
|
with patch("sys.argv", ["bmspy-ups"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.handle_email", mock_email):
|
|
ups_mod.main()
|
|
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.
|
|
|
|
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),
|
|
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()
|
|
mock_shutdown = MagicMock()
|
|
with patch("sys.argv", ["bmspy-ups"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown", mock_shutdown), \
|
|
patch("bmspy.ups.handle_email", mock_email):
|
|
ups_mod.main()
|
|
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"),
|
|
]
|
|
|
|
OUTAGE = [(1.0, 0.95)] * 8 + [(-5.0, 0.90)] * 4 # mains power fails at reading 9
|
|
|
|
def test_power_return_below_the_warning_level_is_reported(self):
|
|
"""The charge is 60% when mains power returns at reading 17."""
|
|
readings = self.OUTAGE + [(-5.0, 0.60)] * 4 + [(3.0, 0.60)] * 8 + [(3.0, 0.80)] * 4
|
|
assert self._events(readings) == [
|
|
(10, "Power loss alert"), (14, "Warning"), (18, "cancel"), (18, "Recovery alert"),
|
|
]
|
|
# The next power failure can send its emails again.
|
|
assert (ups_mod.alert_sent, ups_mod.warning_sent, ups_mod.critical_sent) == (False, False, False)
|
|
|
|
def test_power_return_below_the_critical_level_cancels_the_shutdown(self):
|
|
"""The charge is 25% when mains power returns at reading 17."""
|
|
readings = self.OUTAGE + [(-5.0, 0.25)] * 4 + [(3.0, 0.25)] * 8
|
|
events = self._events(readings)
|
|
assert events[:3] == [(10, "Power loss alert"), (14, "shutdown"), (14, "Critical alert")]
|
|
# The shutdown is canceled, and the low charge on mains power does not start it again.
|
|
assert events[-2:] == [(18, "cancel"), (18, "Recovery alert")]
|
|
assert [number for number, event in events if event == "shutdown"] == [14, 15, 16]
|
|
|
|
def test_second_power_failure_is_reported_again(self):
|
|
readings = self.OUTAGE + [(-5.0, 0.60)] * 4 + [(3.0, 0.60)] * 6
|
|
readings = readings + [(-5.0, 0.60)] * 4
|
|
assert [event for _, event in self._events(readings)] == [
|
|
"Power loss alert", "Warning", "cancel", "Recovery alert", "Power loss alert", "Warning",
|
|
]
|
|
|
|
def test_one_discharging_reading_on_mains_power_does_not_shut_down(self):
|
|
"""One reading can be wrong. A shutdown needs two discharging readings."""
|
|
events = self._events([(3.0, 0.20)] * 5 + [(-0.1, 0.20)] + [(3.0, 0.20)] * 4)
|
|
assert [event for _, event in events if event in ("shutdown", "Critical alert", "Warning")] == []
|
|
|
|
def test_low_battery_on_mains_power_does_not_shut_down(self):
|
|
"""At a restart after a long power failure, the battery is low and it charges."""
|
|
assert self._events([(3.0, 0.20)] * 8) == []
|
|
|
|
def test_device_without_current_is_judged_by_its_charge(self):
|
|
"""The River models report the charge but no battery current."""
|
|
events = self._events([(None, 0.95)] * 5 + [(None, 0.20)] * 2)
|
|
assert events == [(7, "shutdown"), (7, "Critical alert")]
|
|
|
|
def test_main_not_enough_history_debug_2(self, capsys):
|
|
"""With debug=2 and < 4 readings, prints 'not enough readings'."""
|
|
data_seq = [self._make_data(0.0, 0.95)] * 5
|
|
with patch("sys.argv", ["bmspy-ups", "-v", "-v"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.handle_email"):
|
|
ups_mod.main()
|
|
captured = capsys.readouterr()
|
|
assert "not enough readings" in captured.out.lower()
|
|
|
|
def test_main_below_warning_debug_1(self, capsys):
|
|
"""With debug=1, warning message is printed."""
|
|
ups_mod.warning_sent = False
|
|
data_seq = [self._make_data(-1.0, 0.60)] * 5
|
|
with patch("sys.argv", ["bmspy-ups", "-v", "--warning", "75", "--critical", "30"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.handle_email"):
|
|
ups_mod.main()
|
|
captured = capsys.readouterr()
|
|
assert "warning" in captured.out.lower()
|
|
|
|
def test_main_discharge_alert_debug_1(self, capsys):
|
|
"""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.88),
|
|
]
|
|
with patch("sys.argv", ["bmspy-ups", "-v"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.handle_email"):
|
|
ups_mod.main()
|
|
captured = capsys.readouterr()
|
|
assert "discharging" in captured.out.lower()
|
|
|
|
def test_main_power_regained_debug_1(self, capsys):
|
|
"""With debug=1, power regained message is printed."""
|
|
ups_mod.alert_sent = True
|
|
data_seq = [
|
|
self._make_data(-1.0, 0.80),
|
|
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),
|
|
]
|
|
with patch("sys.argv", ["bmspy-ups", "-v"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.handle_email"):
|
|
ups_mod.main()
|
|
captured = capsys.readouterr()
|
|
assert "power regained" in captured.out.lower()
|
|
|
|
def test_main_debug_2_prints_current_and_capacity(self, capsys):
|
|
"""With debug=2, current and capacity are printed."""
|
|
data_seq = [self._make_data(1.0, 0.80)] * 5
|
|
with patch("sys.argv", ["bmspy-ups", "-v", "-v"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.handle_email"):
|
|
ups_mod.main()
|
|
captured = capsys.readouterr()
|
|
assert "current" in captured.out.lower() or "capacity" in captured.out.lower()
|
|
|
|
def test_main_debug_1_prints_thresholds(self, capsys):
|
|
"""With debug=1 and below threshold, threshold messages are printed."""
|
|
ups_mod.critical_sent = False
|
|
data_seq = [self._make_data(-1.0, 0.20)] * 5
|
|
with patch("sys.argv", ["bmspy-ups", "-v", "--critical", "30"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data", side_effect=[data_seq[0]] + data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.handle_email"):
|
|
ups_mod.main()
|
|
captured = capsys.readouterr()
|
|
assert "critical" in captured.out.lower() or "threshold" in captured.out.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# main() — device selection errors and --device flag
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestUpsMainDeviceSelection:
|
|
def _make_data(self, name, current_amps=0.0, charge_ratio=0.8):
|
|
return {name: _make_ups(current_amps, charge_ratio)}
|
|
|
|
def test_main_waits_at_startup_until_the_device_has_data(self, capsys):
|
|
"""A failed first read must not end the monitor: systemd would not restart it."""
|
|
with patch("sys.argv", ["bmspy-ups"]), \
|
|
patch("bmspy.ups.client.list_devices", return_value=["ups1"]), \
|
|
patch("bmspy.ups.client.read_data",
|
|
side_effect=[{}, {}, self._make_data("ups1"), StopIteration("done")]) as read_data, \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep") as mock_sleep:
|
|
with pytest.raises(StopIteration):
|
|
ups_mod.main()
|
|
# Two empty answers, the answer with the device, then the first loop read.
|
|
assert [c.kwargs for c in read_data.call_args_list] == [{"ups": "ups1"}] * 4
|
|
assert mock_sleep.call_args_list == [call(ups_mod.STARTUP_RETRY_PERIOD)] * 2
|
|
# The reason is reported once, not on each retry.
|
|
assert capsys.readouterr().out.count("UPS 'ups1' has no data yet") == 1
|
|
|
|
def test_main_selects_from_configured_devices_not_from_those_with_data(self):
|
|
"""One device with data must not be picked when the server has several."""
|
|
with patch("sys.argv", ["bmspy-ups", "--device", "ups2"]), \
|
|
patch("bmspy.ups.client.list_devices", return_value=["ups1", "ups2"]), \
|
|
patch("bmspy.ups.client.read_data",
|
|
side_effect=[{}, self._make_data("ups2"), StopIteration("done")]) as read_data, \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep") as mock_sleep:
|
|
with pytest.raises(StopIteration):
|
|
ups_mod.main()
|
|
assert [c.kwargs for c in read_data.call_args_list] == [{"ups": "ups2"}] * 3
|
|
assert mock_sleep.call_count == 1
|
|
|
|
@pytest.mark.parametrize("argv,expected", [
|
|
([], "multiple UPS devices found, specify one with --device: ups1, ups2"),
|
|
(["--device", "ghost"], "UPS 'ghost' not found. Available: ups1, ups2"),
|
|
])
|
|
def test_main_wrong_or_missing_device_option_is_an_error_exit(self, capsys, argv, expected):
|
|
"""Only ups1 has data, but both are configured: do not guess, and do not wait."""
|
|
with patch("sys.argv", ["bmspy-ups"] + argv), \
|
|
patch("bmspy.ups.client.list_devices", return_value=["ups1", "ups2"]), \
|
|
patch("bmspy.ups.client.read_data", return_value=self._make_data("ups1")) as read_data, \
|
|
patch("bmspy.ups.client.handle_registration"):
|
|
with pytest.raises(SystemExit) as exit_info:
|
|
ups_mod.main()
|
|
# The argparse usage status: systemd reports it and does not restart.
|
|
assert exit_info.value.code == ups_mod.USAGE_ERROR_STATUS == 2
|
|
assert expected in capsys.readouterr().out
|
|
read_data.assert_not_called()
|
|
|
|
def test_main_ups_flag_selects_device(self):
|
|
"""--device selects the correct device from multiple available ones."""
|
|
data_seq = [self._make_data("ups2")] * 6
|
|
with patch("sys.argv", ["bmspy-ups", "--device", "ups2"]), \
|
|
patch("bmspy.ups.client.list_devices", return_value=["ups1", "ups2"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data",
|
|
side_effect=[{"ups1": _make_ups(0.0, 0.9), "ups2": _make_ups(0.0, 0.8)}]
|
|
+ data_seq + [StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep"), \
|
|
patch("bmspy.ups.handle_shutdown"), \
|
|
patch("bmspy.ups.handle_email"):
|
|
ups_mod.main()
|
|
|
|
def test_main_loop_calls_read_data_with_its_real_parameters(self):
|
|
"""The loop must use the client's parameter names, not only a mock's."""
|
|
from unittest.mock import create_autospec
|
|
|
|
read_data = create_autospec(
|
|
ups_mod.client.read_data,
|
|
side_effect=[{"testups": _make_ups(0.0, 0.9)}, StopIteration("done")],
|
|
)
|
|
with patch("sys.argv", ["bmspy-ups", "--socket", "/tmp/test.sock"]), \
|
|
patch("bmspy.ups.client.read_data", read_data), \
|
|
patch("bmspy.ups.client.handle_registration"):
|
|
with pytest.raises(StopIteration):
|
|
ups_mod.main()
|
|
assert read_data.call_args_list[1] == (("/tmp/test.sock", "ups"), {"ups": "testups"})
|
|
|
|
def test_main_device_disappears_in_loop_sleeps(self):
|
|
"""When the named device is missing from a loop response, main() sleeps and retries."""
|
|
initial = {"testups": _make_ups(0.0, 0.8)}
|
|
# Second call returns empty dict (device gone), third raises StopIteration
|
|
with patch("sys.argv", ["bmspy-ups"]):
|
|
with pytest.raises(StopIteration):
|
|
with patch("bmspy.ups.client.read_data",
|
|
side_effect=[initial, {}, StopIteration("done")]), \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep") as mock_sleep:
|
|
ups_mod.main()
|
|
mock_sleep.assert_called()
|