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()