diff --git a/python/common.py b/python/common.py index 0fb07c2..3703ca2 100644 --- a/python/common.py +++ b/python/common.py @@ -4,11 +4,8 @@ from __future__ import annotations import logging from datetime import UTC, datetime -from os import getenv from subprocess import PIPE, Popen -from apprise import Apprise - from python.logging_config import configure_logger as _configure_logger logger = logging.getLogger(__name__) @@ -39,26 +36,6 @@ def bash_wrapper(command: str) -> tuple[str, int]: return output.decode(), process.returncode -def signal_alert(body: str, title: str = "") -> None: - """Send a signal alert. - - Args: - body (str): The body of the alert. - title (str, optional): The title of the alert. Defaults to "". - """ - apprise_client = Apprise() - - from_phone = getenv("SIGNAL_ALERT_FROM_PHONE") - to_phone = getenv("SIGNAL_ALERT_TO_PHONE") - if not from_phone or not to_phone: - logger.info("SIGNAL_ALERT_FROM_PHONE or SIGNAL_ALERT_TO_PHONE not set") - return - - apprise_client.add(f"signal://localhost:8989/{from_phone}/{to_phone}") - - apprise_client.notify(title=title, body=body) - - def utcnow() -> datetime: """Get the current UTC time.""" return datetime.now(tz=UTC) diff --git a/python/signal_alert.py b/python/signal_alert.py new file mode 100644 index 0000000..07632ad --- /dev/null +++ b/python/signal_alert.py @@ -0,0 +1,30 @@ +"""signal_alert.""" + +from __future__ import annotations + +import logging +from os import getenv + +from apprise import Apprise + +logger = logging.getLogger(__name__) + + +def signal_alert(body: str, title: str = "") -> None: + """Send a signal alert. + + Args: + body (str): The body of the alert. + title (str, optional): The title of the alert. Defaults to "". + """ + apprise_client = Apprise() + + from_phone = getenv("SIGNAL_ALERT_FROM_PHONE") + to_phone = getenv("SIGNAL_ALERT_TO_PHONE") + if not from_phone or not to_phone: + logger.info("SIGNAL_ALERT_FROM_PHONE or SIGNAL_ALERT_TO_PHONE not set") + return + + apprise_client.add(f"signal://localhost:8989/{from_phone}/{to_phone}") + + apprise_client.notify(title=title, body=body) diff --git a/python/system_tests/validate_system.py b/python/system_tests/validate_system.py index 66cabcf..ef75525 100644 --- a/python/system_tests/validate_system.py +++ b/python/system_tests/validate_system.py @@ -9,7 +9,8 @@ from socket import gethostname import typer -from python.common import configure_logger, signal_alert +from python.common import configure_logger +from python.signal_alert import signal_alert from python.system_tests.components import systemd_tests, zpool_tests logger = logging.getLogger(__name__) diff --git a/python/tools/snapshot_manager.py b/python/tools/snapshot_manager.py index e86c046..866c6a5 100644 --- a/python/tools/snapshot_manager.py +++ b/python/tools/snapshot_manager.py @@ -12,7 +12,8 @@ from re import search import typer -from python.common import configure_logger, signal_alert, utcnow +from python.common import configure_logger, utcnow +from python.signal_alert import signal_alert from python.zfs import Dataset, get_datasets logger = logging.getLogger(__name__) diff --git a/tests/test_common.py b/tests/test_common.py index f8689f2..ca86873 100644 --- a/tests/test_common.py +++ b/tests/test_common.py @@ -2,15 +2,7 @@ from __future__ import annotations -from os import environ -from typing import TYPE_CHECKING - -from apprise import Apprise - -from python.common import bash_wrapper, signal_alert, utcnow - -if TYPE_CHECKING: - from pytest_mock import MockerFixture +from python.common import bash_wrapper, utcnow def test_utcnow() -> None: @@ -18,34 +10,6 @@ def test_utcnow() -> None: utcnow() -def test_signal_alert(mocker: MockerFixture) -> None: - """test_signal_alert.""" - environ["SIGNAL_ALERT_FROM_PHONE"] = "1234567890" - environ["SIGNAL_ALERT_TO_PHONE"] = "0987654321" - - mock_logger = mocker.patch("python.common.logger") - mock_apprise_client = mocker.MagicMock(spec=Apprise) - mocker.patch("python.common.Apprise", return_value=mock_apprise_client) - - signal_alert("test") - - mock_logger.info.assert_not_called() - mock_apprise_client.add.assert_called_once_with("signal://localhost:8989/1234567890/0987654321") - mock_apprise_client.notify.assert_called_once_with(title="", body="test") - - -def test_signal_alert_no_phones(mocker: MockerFixture) -> None: - """test_signal_alert_no_phones.""" - if "SIGNAL_ALERT_FROM_PHONE" in environ: - del environ["SIGNAL_ALERT_FROM_PHONE"] - if "SIGNAL_ALERT_TO_PHONE" in environ: - del environ["SIGNAL_ALERT_TO_PHONE"] - mock_logger = mocker.patch("python.common.logger") - signal_alert("test") - - mock_logger.info.assert_called_once_with("SIGNAL_ALERT_FROM_PHONE or SIGNAL_ALERT_TO_PHONE not set") - - def test_test_bash_wrapper() -> None: """test_test_bash_wrapper.""" stdout, returncode = bash_wrapper("echo test") diff --git a/tests/test_signal_alert.py b/tests/test_signal_alert.py new file mode 100644 index 0000000..26281ec --- /dev/null +++ b/tests/test_signal_alert.py @@ -0,0 +1,41 @@ +"""test_signal_alert.""" + +from __future__ import annotations + +from os import environ +from typing import TYPE_CHECKING + +from apprise import Apprise + +from python.signal_alert import signal_alert + +if TYPE_CHECKING: + from pytest_mock import MockerFixture + + +def test_signal_alert(mocker: MockerFixture) -> None: + """test_signal_alert.""" + environ["SIGNAL_ALERT_FROM_PHONE"] = "1234567890" + environ["SIGNAL_ALERT_TO_PHONE"] = "0987654321" + + mock_logger = mocker.patch("python.signal_alert.logger") + mock_apprise_client = mocker.MagicMock(spec=Apprise) + mocker.patch("python.signal_alert.Apprise", return_value=mock_apprise_client) + + signal_alert("test") + + mock_logger.info.assert_not_called() + mock_apprise_client.add.assert_called_once_with("signal://localhost:8989/1234567890/0987654321") + mock_apprise_client.notify.assert_called_once_with(title="", body="test") + + +def test_signal_alert_no_phones(mocker: MockerFixture) -> None: + """test_signal_alert_no_phones.""" + if "SIGNAL_ALERT_FROM_PHONE" in environ: + del environ["SIGNAL_ALERT_FROM_PHONE"] + if "SIGNAL_ALERT_TO_PHONE" in environ: + del environ["SIGNAL_ALERT_TO_PHONE"] + mock_logger = mocker.patch("python.signal_alert.logger") + signal_alert("test") + + mock_logger.info.assert_called_once_with("SIGNAL_ALERT_FROM_PHONE or SIGNAL_ALERT_TO_PHONE not set")