From fb58bac89de1fb3598bbd64c2720181eac4f4337 Mon Sep 17 00:00:00 2001 From: Richie Cahill Date: Sat, 19 Sep 2026 22:28:55 -0400 Subject: [PATCH] refactor(signal): replace Apprise with httpx --- common/optional/zfs/snapshot.nix | 2 +- python/signal_alert.py | 30 +++++++++++++++---- tests/test_signal_alert.py | 50 ++++++++++++++++++++++++++++---- 3 files changed, 71 insertions(+), 11 deletions(-) diff --git a/common/optional/zfs/snapshot.nix b/common/optional/zfs/snapshot.nix index d4b908f..43d018a 100644 --- a/common/optional/zfs/snapshot.nix +++ b/common/optional/zfs/snapshot.nix @@ -9,7 +9,7 @@ let cfg = config.services.snapshot_manager; snapshotManagerPackages = ps: with ps; [ - apprise + httpx typer ]; in diff --git a/python/signal_alert.py b/python/signal_alert.py index 07632ad..08e85f3 100644 --- a/python/signal_alert.py +++ b/python/signal_alert.py @@ -5,10 +5,13 @@ from __future__ import annotations import logging from os import getenv -from apprise import Apprise +import httpx logger = logging.getLogger(__name__) +SIGNAL_API_URL = "http://localhost:8989/v2/send" +SIGNAL_API_TIMEOUT = 4.0 + def signal_alert(body: str, title: str = "") -> None: """Send a signal alert. @@ -17,14 +20,31 @@ def signal_alert(body: str, title: str = "") -> None: 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's Signal integration did not support titles, so preserve that behavior. + if title: + logger.debug("Signal does not support notification titles; ignoring title") - apprise_client.notify(title=title, body=body) + try: + response = httpx.post( + SIGNAL_API_URL, + json={ + "message": body, + "number": from_phone, + "recipients": [to_phone], + "text_mode": "normal", + }, + timeout=SIGNAL_API_TIMEOUT, + follow_redirects=True, + ) + except httpx.HTTPError: + logger.exception("Unable to contact the Signal API") + return + + if response.status_code not in {httpx.codes.OK, httpx.codes.CREATED}: + logger.error("Signal API returned HTTP status %d", response.status_code) diff --git a/tests/test_signal_alert.py b/tests/test_signal_alert.py index 26281ec..de056ae 100644 --- a/tests/test_signal_alert.py +++ b/tests/test_signal_alert.py @@ -5,7 +5,7 @@ from __future__ import annotations from os import environ from typing import TYPE_CHECKING -from apprise import Apprise +import httpx from python.signal_alert import signal_alert @@ -19,14 +19,23 @@ def test_signal_alert(mocker: MockerFixture) -> None: 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) + mock_response = mocker.MagicMock(spec=httpx.Response, status_code=httpx.codes.CREATED) + mock_post = mocker.patch("python.signal_alert.httpx.post", return_value=mock_response) 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") + mock_post.assert_called_once_with( + "http://localhost:8989/v2/send", + json={ + "message": "test", + "number": "1234567890", + "recipients": ["0987654321"], + "text_mode": "normal", + }, + timeout=4.0, + follow_redirects=True, + ) def test_signal_alert_no_phones(mocker: MockerFixture) -> None: @@ -39,3 +48,34 @@ def test_signal_alert_no_phones(mocker: MockerFixture) -> None: signal_alert("test") mock_logger.info.assert_called_once_with("SIGNAL_ALERT_FROM_PHONE or SIGNAL_ALERT_TO_PHONE not set") + + +def test_signal_alert_http_error(mocker: MockerFixture) -> None: + """HTTP errors are logged rather than propagated.""" + environ["SIGNAL_ALERT_FROM_PHONE"] = "1234567890" + environ["SIGNAL_ALERT_TO_PHONE"] = "0987654321" + + mock_logger = mocker.patch("python.signal_alert.logger") + request = httpx.Request("POST", "http://localhost:8989/v2/send") + mocker.patch( + "python.signal_alert.httpx.post", + side_effect=httpx.ConnectError("connection failed", request=request), + ) + + signal_alert("test") + + mock_logger.exception.assert_called_once_with("Unable to contact the Signal API") + + +def test_signal_alert_unsuccessful_response(mocker: MockerFixture) -> None: + """Unexpected response statuses are logged.""" + environ["SIGNAL_ALERT_FROM_PHONE"] = "1234567890" + environ["SIGNAL_ALERT_TO_PHONE"] = "0987654321" + + mock_logger = mocker.patch("python.signal_alert.logger") + mock_response = mocker.MagicMock(spec=httpx.Response, status_code=httpx.codes.BAD_GATEWAY) + mocker.patch("python.signal_alert.httpx.post", return_value=mock_response) + + signal_alert("test") + + mock_logger.error.assert_called_once_with("Signal API returned HTTP status %d", httpx.codes.BAD_GATEWAY) -- 2.55.0