refactor: extract signal_alert into its own module
treefmt / nix fmt (pull_request) Successful in 5s
pytest / pytest (pull_request) Successful in 24s
build_systems / build-bob (pull_request) Successful in 45s
build_systems / build-brain (pull_request) Successful in 45s
build_systems / build-rhapsody-in-green (pull_request) Successful in 58s
build_systems / build-jeeves (pull_request) Successful in 2m15s
treefmt / nix fmt (push) Successful in 5s
build_systems / build-jeeves (push) Successful in 9s
pytest / pytest (push) Successful in 24s
build_systems / build-brain (push) Successful in 30s
build_systems / build-bob (push) Successful in 32s
build_systems / build-rhapsody-in-green (push) Successful in 44s

Move signal_alert out of python/common.py into a dedicated
python/signal_alert.py module and update its importers
(validate_system.py, snapshot_manager.py) to the new path.

Relocate the signal_alert tests from tests/test_common.py into
tests/test_signal_alert.py, repatching python.signal_alert.logger and
python.signal_alert.Apprise to match the new module.
This commit was merged in pull request #39.
This commit is contained in:
2026-07-12 13:16:28 -04:00
committed by Richie
parent 4a410dbdf8
commit 80a521f297
6 changed files with 76 additions and 62 deletions
-23
View File
@@ -4,11 +4,8 @@ from __future__ import annotations
import logging import logging
from datetime import UTC, datetime from datetime import UTC, datetime
from os import getenv
from subprocess import PIPE, Popen from subprocess import PIPE, Popen
from apprise import Apprise
from python.logging_config import configure_logger as _configure_logger from python.logging_config import configure_logger as _configure_logger
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -39,26 +36,6 @@ def bash_wrapper(command: str) -> tuple[str, int]:
return output.decode(), process.returncode 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: def utcnow() -> datetime:
"""Get the current UTC time.""" """Get the current UTC time."""
return datetime.now(tz=UTC) return datetime.now(tz=UTC)
+30
View File
@@ -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)
+2 -1
View File
@@ -9,7 +9,8 @@ from socket import gethostname
import typer 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 from python.system_tests.components import systemd_tests, zpool_tests
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
+2 -1
View File
@@ -12,7 +12,8 @@ from re import search
import typer 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 from python.zfs import Dataset, get_datasets
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
+1 -37
View File
@@ -2,15 +2,7 @@
from __future__ import annotations from __future__ import annotations
from os import environ from python.common import bash_wrapper, utcnow
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
def test_utcnow() -> None: def test_utcnow() -> None:
@@ -18,34 +10,6 @@ def test_utcnow() -> None:
utcnow() 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: def test_test_bash_wrapper() -> None:
"""test_test_bash_wrapper.""" """test_test_bash_wrapper."""
stdout, returncode = bash_wrapper("echo test") stdout, returncode = bash_wrapper("echo test")
+41
View File
@@ -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")