tim
6b636cde81
If the server had no data for the device, bmspy-ups slept and read again with no report. A battery that stopped answering during a power failure gave no email and no shutdown. bmspy-ups now sends one email after 5 minutes with no data, and one more when the data returns. A new email for a loss needs another 5 minutes with no data after that return. The same rule applies to the wait at startup.
813 lines
37 KiB
Python
813 lines
37 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.
|
|
|
|
A reading of None is a read in which the server has no data for the
|
|
device. Return (reading number, event) for each email and each
|
|
shutdown call.
|
|
"""
|
|
sequence = [{} if reading is None else self._make_data(*reading) for reading in readings]
|
|
for data, reading in zip(sequence, readings):
|
|
if reading is not None and reading[0] is None:
|
|
del data["testups"]._fields["bms_current_amps"]
|
|
events, number, clock = [], [0], [1000.0]
|
|
|
|
def sleep(seconds):
|
|
clock[0] += seconds
|
|
|
|
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", side_effect=sleep), \
|
|
patch("bmspy.ups.time.monotonic", side_effect=lambda: clock[0]), \
|
|
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_no_data_for_five_minutes_is_reported_once_and_the_return_once(self):
|
|
"""Readings are 30 seconds apart, so the tenth read with no data is at 5 minutes."""
|
|
readings = (
|
|
[(1.0, 0.95)] * 3 # readings 1 to 3: data
|
|
+ [None] * 25 # 4 to 28: no data for 12.5 minutes
|
|
+ [(1.0, 0.95)] * 3 # 29 to 31: the data returns
|
|
+ [None] * 9 # 32 to 40: 4.5 minutes, below the limit
|
|
+ [(1.0, 0.95)] * 2 # 41 and 42
|
|
+ [None] * 12 # 43 to 54: no data for 6 minutes
|
|
)
|
|
assert self._events(readings) == [
|
|
(13, "Data loss alert"), (29, "Data recovery alert"), (52, "Data loss alert"),
|
|
]
|
|
|
|
def test_data_loss_email_names_the_device(self):
|
|
send_email = MagicMock()
|
|
with patch("bmspy.ups.time.monotonic", side_effect=[0, 299, 300, 900, 901]):
|
|
monitor = ups_mod._DataMonitor(send_email)
|
|
monitor.update("network", False) # 299 seconds: below the limit
|
|
send_email.assert_not_called()
|
|
monitor.update("network", False) # 300 seconds
|
|
monitor.update("network", False) # no second email
|
|
monitor.update("network", True)
|
|
assert [c.kwargs for c in send_email.call_args_list] == [
|
|
{"text": "no data from UPS 'network' for 5 minutes", "level": "Data loss alert"},
|
|
{"text": "data from UPS 'network' regained", "level": "Data recovery 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_no_data_at_startup_for_five_minutes_is_reported(self):
|
|
"""The startup wait reads every 5 seconds, so read 61 is at 5 minutes."""
|
|
clock = [1000.0]
|
|
|
|
def sleep(seconds):
|
|
clock[0] += seconds
|
|
|
|
reads = [{}] * 70 + [self._make_data("ups1"), self._make_data("ups1"), StopIteration("done")]
|
|
mock_email = MagicMock()
|
|
with patch("sys.argv", ["bmspy-ups", "--to", "admin@example.com"]), \
|
|
patch("bmspy.ups.client.list_devices", return_value=["ups1"]), \
|
|
patch("bmspy.ups.client.read_data", side_effect=reads) as read_data, \
|
|
patch("bmspy.ups.client.handle_registration"), \
|
|
patch("bmspy.ups.time.sleep", side_effect=sleep), \
|
|
patch("bmspy.ups.time.monotonic", side_effect=lambda: clock[0]), \
|
|
patch("bmspy.ups.handle_email", mock_email):
|
|
with pytest.raises(StopIteration):
|
|
ups_mod.main()
|
|
assert [(c.kwargs["level"], c.kwargs["recipient"]) for c in mock_email.call_args_list] == [
|
|
("Data loss alert", "admin@example.com"), ("Data recovery alert", "admin@example.com"),
|
|
]
|
|
# The loss email goes out during the wait, and the other one after it.
|
|
assert read_data.call_count == 73
|
|
|
|
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()
|