common/optional/zfs/snapshot.nix

@@ -9,7 +9,7 @@ let
   cfg = config.services.snapshot_manager;
   snapshotManagerPackages =
     ps: with ps; [
-      apprise
+      httpx
       typer
     ];
 in

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)

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)