Richie/dotfiles
1
0
Fork 0
Code Issues 1 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Update Signal alert handling

Browse Source
This commit is contained in:
Richie
2026-09-30 07:58:30 -04:00
parent 2422bb3664
commit b3d8df0c37
5 changed files with 99 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_signal_alert.py
+10 -36
View File
@@ -4,25 +4,23 @@ from __future__ import annotations
import json
import socket
from os import environ
from typing import TYPE_CHECKING
from python.signal_alert import signal_alert
import pytest
from python.signal_alert import SignalRPCError, signal_alert
if TYPE_CHECKING:
from pytest_mock import MockerFixture
def test_signal_alert(mocker: MockerFixture) -> None:
environ["SIGNAL_ALERT_FROM_PHONE"] = "1234567890"
environ["SIGNAL_ALERT_TO_PHONE"] = "0987654321"
mock_logger = mocker.patch("python.signal_alert.logger")
mock_socket = mocker.patch("python.signal_alert.socket.socket")
client = mock_socket.return_value.__enter__.return_value
client.recv.side_effect = [b'{"jsonrpc":"2.0","result":{},"id":"signal-alert"}\n']
signal_alert("test")
assert signal_alert("test", "1234567890", "0987654321") is None
mock_logger.info.assert_not_called()
mock_socket.assert_called_once_with(socket.AF_UNIX, socket.SOCK_STREAM)
@@ -41,55 +39,31 @@ def test_signal_alert(mocker: MockerFixture) -> None:
}
def test_signal_alert_no_phones(mocker: MockerFixture) -> None:
environ.pop("SIGNAL_ALERT_FROM_PHONE", None)
environ.pop("SIGNAL_ALERT_TO_PHONE", None)
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")
def test_signal_alert_socket_error(mocker: MockerFixture) -> None:
environ["SIGNAL_ALERT_FROM_PHONE"] = "1234567890"
environ["SIGNAL_ALERT_TO_PHONE"] = "0987654321"
mock_logger = mocker.patch("python.signal_alert.logger")
mocker.patch("python.signal_alert.socket.socket", side_effect=ConnectionError("connection failed"))
signal_alert("test")
with pytest.raises(SignalRPCError, match="Signal RPC failed") as exc_info:
signal_alert("test", "1234567890", "0987654321")
mock_logger.exception.assert_called_once_with("Unable to communicate with signal-cli")
assert isinstance(exc_info.value.__cause__, ConnectionError)
def test_signal_alert_rpc_error(mocker: MockerFixture) -> None:
environ["SIGNAL_ALERT_FROM_PHONE"] = "1234567890"
environ["SIGNAL_ALERT_TO_PHONE"] = "0987654321"
mock_logger = mocker.patch("python.signal_alert.logger")
mock_socket = mocker.patch("python.signal_alert.socket.socket")
client = mock_socket.return_value.__enter__.return_value
client.recv.side_effect = [b'{"jsonrpc":"2.0","error":{"code":-1,"message":"failed"},"id":"signal-alert"}\n']
signal_alert("test")
mock_logger.error.assert_called_once_with(
"Signal RPC returned an error: %s",
{"code": -1, "message": "failed"},
)
with pytest.raises(SignalRPCError, match=r"Signal RPC returned an error:.*failed"):
signal_alert("test", "1234567890", "0987654321")
def test_signal_alert_ignores_notifications(mocker: MockerFixture) -> None:
environ["SIGNAL_ALERT_FROM_PHONE"] = "1234567890"
environ["SIGNAL_ALERT_TO_PHONE"] = "0987654321"
mock_socket = mocker.patch("python.signal_alert.socket.socket")
client = mock_socket.return_value.__enter__.return_value
client.recv.side_effect = [
b'{"jsonrpc":"2.0","method":"receive"}\n{"jsonrpc":"2.0","result":{},"id":"signal-alert"}\n'
]
signal_alert("test")
assert signal_alert("test", "1234567890", "0987654321") is None
assert client.recv.call_count == 1
tests/test_snapshot_manager.py
+43 -6
View File
@@ -16,6 +16,17 @@ if TYPE_CHECKING:
from pytest_mock import MockerFixture
SNAPSHOT_MANAGER = "python.tools.snapshot_manager"
FROM_PHONE = "1234567890"
TO_PHONE = "0987654321"
def patch_phone_numbers(mocker: MockerFixture) -> None:
"""Patch the Signal phone number environment variables."""
phone_numbers = {
"SIGNAL_ALERT_FROM_PHONE": FROM_PHONE,
"SIGNAL_ALERT_TO_PHONE": TO_PHONE,
}
mocker.patch(f"{SNAPSHOT_MANAGER}.getenv", side_effect=phone_numbers.get)
def patch_utcnow(mocker: MockerFixture, datetime_value: datetime) -> None:
@@ -34,6 +45,7 @@ def create_mock_snapshot(mocker: MockerFixture, name: str) -> Snapshot:
def test_main(mocker: MockerFixture, fs: FakeFilesystem) -> None:
"""Test main."""
load_config_data.cache_clear()
patch_phone_numbers(mocker)
mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
@@ -58,12 +70,15 @@ def test_main(mocker: MockerFixture, fs: FakeFilesystem) -> None:
"daily": 0,
"monthly": 0,
},
FROM_PHONE,
TO_PHONE,
)
def test_main_create_snapshot_failure(mocker: MockerFixture, fs: FakeFilesystem) -> None:
"""Test main."""
load_config_data.cache_clear()
patch_phone_numbers(mocker)
mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
@@ -78,7 +93,11 @@ def test_main_create_snapshot_failure(mocker: MockerFixture, fs: FakeFilesystem)
fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
main(Path("/mock_snapshot_config.toml"))
mock_signal_alert.assert_called_once_with("test_dataset failed to create snapshot 2023-01-01T00:00:00")
mock_signal_alert.assert_called_once_with(
"test_dataset failed to create snapshot 2023-01-01T00:00:00",
FROM_PHONE,
TO_PHONE,
)
mock_get_datasets.assert_called_once()
mock_get_snapshots_to_delete.assert_not_called()
@@ -86,6 +105,7 @@ def test_main_create_snapshot_failure(mocker: MockerFixture, fs: FakeFilesystem)
def test_main_exception(mocker: MockerFixture, fs: FakeFilesystem) -> None:
"""Test main."""
load_config_data.cache_clear()
patch_phone_numbers(mocker)
mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
@@ -103,7 +123,7 @@ def test_main_exception(mocker: MockerFixture, fs: FakeFilesystem) -> None:
assert isinstance(pytest_wrapped_e.value, SystemExit)
assert pytest_wrapped_e.value.code == 1
mock_signal_alert.assert_called_once_with("snapshot_manager failed")
mock_signal_alert.assert_called_once_with("snapshot_manager failed", FROM_PHONE, TO_PHONE)
mock_get_datasets.assert_called_once()
mock_get_snapshots_to_delete.assert_not_called()
@@ -120,7 +140,12 @@ def test_get_snapshots_to_delete(mocker: MockerFixture) -> None:
mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
get_snapshots_to_delete(mock_dataset, {"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0})
get_snapshots_to_delete(
mock_dataset,
{"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0},
FROM_PHONE,
TO_PHONE,
)
mock_signal_alert.assert_not_called()
mock_dataset.delete_snapshot.assert_called_once_with("auto_202509150415")
@@ -135,7 +160,12 @@ def test_get_snapshots_to_delete_no_snapshot(mocker: MockerFixture) -> None:
mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
get_snapshots_to_delete(mock_dataset, {"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0})
get_snapshots_to_delete(
mock_dataset,
{"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0},
FROM_PHONE,
TO_PHONE,
)
mock_signal_alert.assert_not_called()
mock_dataset.delete_snapshot.assert_not_called()
@@ -153,10 +183,17 @@ def test_get_snapshots_to_delete_errored(mocker: MockerFixture) -> None:
mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
get_snapshots_to_delete(mock_dataset, {"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0})
get_snapshots_to_delete(
mock_dataset,
{"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0},
FROM_PHONE,
TO_PHONE,
)
mock_signal_alert.assert_called_once_with(
"test_dataset@auto_202509150415 failed to delete: snapshot has dependent clones"
"test_dataset@auto_202509150415 failed to delete: snapshot has dependent clones",
FROM_PHONE,
TO_PHONE,
)
mock_dataset.delete_snapshot.assert_called_once_with("auto_202509150415")