70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
"""test_signal_alert."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
from typing import TYPE_CHECKING
|
|
|
|
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:
|
|
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']
|
|
|
|
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)
|
|
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_socket_error(mocker: MockerFixture) -> None:
|
|
mocker.patch("python.signal_alert.socket.socket", side_effect=ConnectionError("connection failed"))
|
|
|
|
with pytest.raises(SignalRPCError, match="Signal RPC failed") as exc_info:
|
|
signal_alert("test", "1234567890", "0987654321")
|
|
|
|
assert isinstance(exc_info.value.__cause__, ConnectionError)
|
|
|
|
|
|
def test_signal_alert_rpc_error(mocker: MockerFixture) -> None:
|
|
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']
|
|
|
|
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:
|
|
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'
|
|
]
|
|
|
|
assert signal_alert("test", "1234567890", "0987654321") is None
|
|
|
|
assert client.recv.call_count == 1
|