diff --git a/bmspy/ups.py b/bmspy/ups.py index cea9ba0..9033bed 100644 --- a/bmspy/ups.py +++ b/bmspy/ups.py @@ -1,7 +1,6 @@ import argparse import atexit import os -import re import time import smtplib import ssl @@ -65,11 +64,13 @@ def handle_shutdown(action: str = "cancel", delay: int = 0, debug: int = 0) -> N if action == "shutdown": if scheduled_shutdown is False: - scheduled_shutdown = time.time() + delay * 60 * 1000 + scheduled_shutdown = time.time() + delay * 60 os.system("/sbin/shutdown {}".format(delay)) elif action == "cancel": os.system("/sbin/shutdown -c") + # Allow the next incident to schedule a shutdown again. + scheduled_shutdown = False return @@ -88,7 +89,7 @@ def handle_email( isSSL = False hostname = socket.gethostname() - if re.match("/@/", recipient) is None: + if "@" not in recipient: recipient = "{}@{}".format(recipient, hostname) sender = "bmspy on {} ".format(hostname, hostname) @@ -169,6 +170,7 @@ def main() -> None: "-p", dest="port", action="store", + type=int, default=25, help="Mail server port (default: 25)", ) @@ -229,7 +231,7 @@ def main() -> None: history = deque() while True: - data = client.read_data(args.socket, "ups", device=device_name) + data = client.read_data(args.socket, "ups", ups=device_name) device_obj = data.get(device_name) if device_obj is None: time.sleep(DAEMON_UPDATE_PERIOD) diff --git a/tests/test_ups.py b/tests/test_ups.py index 7cbfb25..be39147 100644 --- a/tests/test_ups.py +++ b/tests/test_ups.py @@ -110,6 +110,16 @@ class TestHandleShutdown: 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 @@ -203,6 +213,13 @@ class TestHandleEmail: 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 @@ -370,6 +387,23 @@ class TestUpsMain: 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(0.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. @@ -570,6 +604,21 @@ class TestUpsMainDeviceSelection: 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_ups_flag_unknown_device_returns_early(self, capsys): """--device with an unknown device name exits with an error.""" with patch("sys.argv", ["bmspy-ups", "--device", "ghost"]), \