96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
"""test_signal_alert."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
from os import environ
|
|
from typing import TYPE_CHECKING
|
|
|
|
from python.signal_alert import 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")
|
|
|
|
mock_logger.info.assert_not_called()
|
|
mock_socket.assert_called_once_with(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
client.settimeout.assert_called_once_with(4.0)
|
|
client.connect.assert_called_once_with("/run/signal-cli/socket")
|
|
request = json.loads(client.sendall.call_args.args[0])
|
|
assert request == {
|
|
"jsonrpc": "2.0",
|
|
"method": "send",
|
|
"params": {
|
|
"account": "1234567890",
|
|
"recipient": ["0987654321"],
|
|
"message": "test",
|
|
},
|
|
"id": "signal-alert",
|
|
}
|
|
|
|
|
|
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")
|
|
|
|
mock_logger.exception.assert_called_once_with("Unable to communicate with signal-cli")
|
|
|
|
|
|
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"},
|
|
)
|
|
|
|
|
|
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 client.recv.call_count == 1
|