|
|
|
@ -23,9 +23,9 @@
|
|
|
|
|
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN |
|
|
|
|
# THE SOFTWARE. |
|
|
|
|
import os |
|
|
|
|
import sys |
|
|
|
|
from json import loads |
|
|
|
|
from unittest import mock |
|
|
|
|
from inspect import cleandoc |
|
|
|
|
|
|
|
|
|
import pytest |
|
|
|
|
import requests |
|
|
|
@ -34,6 +34,7 @@ from apprise.plugins.NotifySignalAPI import NotifySignalAPI
|
|
|
|
|
from helpers import AppriseURLTester |
|
|
|
|
from apprise import AppriseAttachment |
|
|
|
|
from apprise import NotifyType |
|
|
|
|
from apprise.config.ConfigBase import ConfigBase |
|
|
|
|
|
|
|
|
|
# Disable logging for a cleaner testing output |
|
|
|
|
import logging |
|
|
|
@ -42,6 +43,19 @@ logging.disable(logging.CRITICAL)
|
|
|
|
|
# Attachment Directory |
|
|
|
|
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var') |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture |
|
|
|
|
def request_mock(mocker): |
|
|
|
|
""" |
|
|
|
|
Prepare requests mock. |
|
|
|
|
""" |
|
|
|
|
mock_post = mocker.patch("requests.post") |
|
|
|
|
mock_post.return_value = requests.Request() |
|
|
|
|
mock_post.return_value.status_code = requests.codes.ok |
|
|
|
|
mock_post.return_value.content = "" |
|
|
|
|
return mock_post |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Our Testing URLs |
|
|
|
|
apprise_url_tests = ( |
|
|
|
|
('signal://', { |
|
|
|
@ -153,20 +167,11 @@ def test_plugin_signal_urls():
|
|
|
|
|
AppriseURLTester(tests=apprise_url_tests).run_all() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@mock.patch('requests.post') |
|
|
|
|
def test_plugin_signal_edge_cases(mock_post): |
|
|
|
|
def test_plugin_signal_edge_cases(request_mock): |
|
|
|
|
""" |
|
|
|
|
NotifySignalAPI() Edge Cases |
|
|
|
|
|
|
|
|
|
""" |
|
|
|
|
|
|
|
|
|
# Prepare our response |
|
|
|
|
response = requests.Request() |
|
|
|
|
response.status_code = requests.codes.ok |
|
|
|
|
|
|
|
|
|
# Prepare Mock |
|
|
|
|
mock_post.return_value = response |
|
|
|
|
|
|
|
|
|
# Initialize some generic (but valid) tokens |
|
|
|
|
source = '+1 (555) 123-3456' |
|
|
|
|
target = '+1 (555) 987-5432' |
|
|
|
@ -181,15 +186,15 @@ def test_plugin_signal_edge_cases(mock_post):
|
|
|
|
|
assert aobj.add("signals://localhost:231/{}/{}".format(source, target)) |
|
|
|
|
assert aobj.notify(title=title, body=body) |
|
|
|
|
|
|
|
|
|
assert mock_post.call_count == 1 |
|
|
|
|
assert request_mock.call_count == 1 |
|
|
|
|
|
|
|
|
|
details = mock_post.call_args_list[0] |
|
|
|
|
details = request_mock.call_args_list[0] |
|
|
|
|
assert details[0][0] == 'https://localhost:231/v2/send' |
|
|
|
|
payload = loads(details[1]['data']) |
|
|
|
|
assert payload['message'] == 'My Title\r\ntest body' |
|
|
|
|
|
|
|
|
|
# Reset our mock object |
|
|
|
|
mock_post.reset_mock() |
|
|
|
|
request_mock.reset_mock() |
|
|
|
|
|
|
|
|
|
aobj = Apprise() |
|
|
|
|
assert aobj.add( |
|
|
|
@ -197,29 +202,69 @@ def test_plugin_signal_edge_cases(mock_post):
|
|
|
|
|
source, target)) |
|
|
|
|
assert aobj.notify(title=title, body=body) |
|
|
|
|
|
|
|
|
|
assert mock_post.call_count == 1 |
|
|
|
|
assert request_mock.call_count == 1 |
|
|
|
|
|
|
|
|
|
details = mock_post.call_args_list[0] |
|
|
|
|
details = request_mock.call_args_list[0] |
|
|
|
|
assert details[0][0] == 'https://localhost:231/v2/send' |
|
|
|
|
payload = loads(details[1]['data']) |
|
|
|
|
# Status flag is set |
|
|
|
|
assert payload['message'] == '[i] My Title\r\ntest body' |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@mock.patch('requests.post') |
|
|
|
|
def test_plugin_signal_test_based_on_feedback(mock_post): |
|
|
|
|
def test_plugin_signal_yaml_config(request_mock): |
|
|
|
|
""" |
|
|
|
|
NotifySignalAPI() User Feedback Test |
|
|
|
|
|
|
|
|
|
NotifySignalAPI() YAML Configuration |
|
|
|
|
""" |
|
|
|
|
|
|
|
|
|
# Prepare our response |
|
|
|
|
response = requests.Request() |
|
|
|
|
response.status_code = requests.codes.ok |
|
|
|
|
|
|
|
|
|
# Prepare Mock |
|
|
|
|
mock_post.return_value = response |
|
|
|
|
# Load our configuration |
|
|
|
|
result, config = ConfigBase.config_parse_yaml(cleandoc(""" |
|
|
|
|
urls: |
|
|
|
|
- signal://signal:8080/+1234567890: |
|
|
|
|
- to: +0987654321 |
|
|
|
|
tag: signal |
|
|
|
|
""")) |
|
|
|
|
|
|
|
|
|
# Verify we loaded correctly |
|
|
|
|
assert isinstance(result, list) |
|
|
|
|
assert len(result) == 1 |
|
|
|
|
assert len(result[0].tags) == 1 |
|
|
|
|
assert 'signal' in result[0].tags |
|
|
|
|
|
|
|
|
|
# Let's get our plugin |
|
|
|
|
plugin = result[0] |
|
|
|
|
assert len(plugin.targets) == 1 |
|
|
|
|
assert '+1234567890' == plugin.source |
|
|
|
|
assert '+0987654321' in plugin.targets |
|
|
|
|
|
|
|
|
|
# |
|
|
|
|
# Test another way to get the same results |
|
|
|
|
# |
|
|
|
|
|
|
|
|
|
# Load our configuration |
|
|
|
|
result, config = ConfigBase.config_parse_yaml(cleandoc(""" |
|
|
|
|
urls: |
|
|
|
|
- signal://signal:8080/+1234567890/+0987654321: |
|
|
|
|
- tag: signal |
|
|
|
|
""")) |
|
|
|
|
|
|
|
|
|
# Verify we loaded correctly |
|
|
|
|
assert isinstance(result, list) |
|
|
|
|
assert len(result) == 1 |
|
|
|
|
assert len(result[0].tags) == 1 |
|
|
|
|
assert 'signal' in result[0].tags |
|
|
|
|
|
|
|
|
|
# Let's get our plugin |
|
|
|
|
plugin = result[0] |
|
|
|
|
assert len(plugin.targets) == 1 |
|
|
|
|
assert '+1234567890' == plugin.source |
|
|
|
|
assert '+0987654321' in plugin.targets |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_plugin_signal_based_on_feedback(request_mock): |
|
|
|
|
""" |
|
|
|
|
NotifySignalAPI() User Feedback Test |
|
|
|
|
|
|
|
|
|
""" |
|
|
|
|
body = "test body" |
|
|
|
|
title = "My Title" |
|
|
|
|
|
|
|
|
@ -231,9 +276,9 @@ def test_plugin_signal_test_based_on_feedback(mock_post):
|
|
|
|
|
assert aobj.notify(title=title, body=body) |
|
|
|
|
|
|
|
|
|
# If a batch, there is only 1 post |
|
|
|
|
assert mock_post.call_count == 1 |
|
|
|
|
assert request_mock.call_count == 1 |
|
|
|
|
|
|
|
|
|
details = mock_post.call_args_list[0] |
|
|
|
|
details = request_mock.call_args_list[0] |
|
|
|
|
assert details[0][0] == 'http://10.0.0.112:8080/v2/send' |
|
|
|
|
payload = loads(details[1]['data']) |
|
|
|
|
assert payload['message'] == 'My Title\r\ntest body' |
|
|
|
@ -244,7 +289,7 @@ def test_plugin_signal_test_based_on_feedback(mock_post):
|
|
|
|
|
assert "+12514444444" in payload['recipients'] |
|
|
|
|
|
|
|
|
|
# Reset our test and turn batch mode off |
|
|
|
|
mock_post.reset_mock() |
|
|
|
|
request_mock.reset_mock() |
|
|
|
|
|
|
|
|
|
aobj = Apprise() |
|
|
|
|
aobj.add( |
|
|
|
@ -254,9 +299,9 @@ def test_plugin_signal_test_based_on_feedback(mock_post):
|
|
|
|
|
assert aobj.notify(title=title, body=body) |
|
|
|
|
|
|
|
|
|
# If a batch, there is only 1 post |
|
|
|
|
assert mock_post.call_count == 2 |
|
|
|
|
assert request_mock.call_count == 2 |
|
|
|
|
|
|
|
|
|
details = mock_post.call_args_list[0] |
|
|
|
|
details = request_mock.call_args_list[0] |
|
|
|
|
assert details[0][0] == 'http://10.0.0.112:8080/v2/send' |
|
|
|
|
payload = loads(details[1]['data']) |
|
|
|
|
assert payload['message'] == 'My Title\r\ntest body' |
|
|
|
@ -264,7 +309,7 @@ def test_plugin_signal_test_based_on_feedback(mock_post):
|
|
|
|
|
assert len(payload['recipients']) == 1 |
|
|
|
|
assert "+12513333333" in payload['recipients'] |
|
|
|
|
|
|
|
|
|
details = mock_post.call_args_list[1] |
|
|
|
|
details = request_mock.call_args_list[1] |
|
|
|
|
assert details[0][0] == 'http://10.0.0.112:8080/v2/send' |
|
|
|
|
payload = loads(details[1]['data']) |
|
|
|
|
assert payload['message'] == 'My Title\r\ntest body' |
|
|
|
@ -274,7 +319,7 @@ def test_plugin_signal_test_based_on_feedback(mock_post):
|
|
|
|
|
# The + is appended |
|
|
|
|
assert "+12514444444" in payload['recipients'] |
|
|
|
|
|
|
|
|
|
mock_post.reset_mock() |
|
|
|
|
request_mock.reset_mock() |
|
|
|
|
|
|
|
|
|
# Test group names |
|
|
|
|
aobj = Apprise() |
|
|
|
@ -285,9 +330,9 @@ def test_plugin_signal_test_based_on_feedback(mock_post):
|
|
|
|
|
assert aobj.notify(title=title, body=body) |
|
|
|
|
|
|
|
|
|
# If a batch, there is only 1 post |
|
|
|
|
assert mock_post.call_count == 1 |
|
|
|
|
assert request_mock.call_count == 1 |
|
|
|
|
|
|
|
|
|
details = mock_post.call_args_list[0] |
|
|
|
|
details = request_mock.call_args_list[0] |
|
|
|
|
assert details[0][0] == 'http://10.0.0.112:8080/v2/send' |
|
|
|
|
payload = loads(details[1]['data']) |
|
|
|
|
assert payload['message'] == 'My Title\r\ntest body' |
|
|
|
@ -304,20 +349,12 @@ def test_plugin_signal_test_based_on_feedback(mock_post):
|
|
|
|
|
assert '/+12514444444' in aobj[0].url() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@mock.patch('requests.post') |
|
|
|
|
def test_notify_signal_plugin_attachments(mock_post): |
|
|
|
|
def test_notify_signal_plugin_attachments(request_mock): |
|
|
|
|
""" |
|
|
|
|
NotifySignalAPI() Attachments |
|
|
|
|
|
|
|
|
|
""" |
|
|
|
|
|
|
|
|
|
okay_response = requests.Request() |
|
|
|
|
okay_response.status_code = requests.codes.ok |
|
|
|
|
okay_response.content = "" |
|
|
|
|
|
|
|
|
|
# Assign our mock object our return value |
|
|
|
|
mock_post.return_value = okay_response |
|
|
|
|
|
|
|
|
|
obj = Apprise.instantiate( |
|
|
|
|
'signal://10.0.0.112:8080/+12512222222/+12513333333/' |
|
|
|
|
'12514444444?batch=no') |
|
|
|
@ -336,13 +373,6 @@ def test_notify_signal_plugin_attachments(mock_post):
|
|
|
|
|
body='body', title='title', notify_type=NotifyType.INFO, |
|
|
|
|
attach=path) is False |
|
|
|
|
|
|
|
|
|
# Get a appropriate "builtin" module name for pythons 2/3. |
|
|
|
|
if sys.version_info.major >= 3: |
|
|
|
|
builtin_open_function = 'builtins.open' |
|
|
|
|
|
|
|
|
|
else: |
|
|
|
|
builtin_open_function = '__builtin__.open' |
|
|
|
|
|
|
|
|
|
# Test Valid Attachment (load 3) |
|
|
|
|
path = ( |
|
|
|
|
os.path.join(TEST_VAR_DIR, 'apprise-test.gif'), |
|
|
|
@ -352,9 +382,7 @@ def test_notify_signal_plugin_attachments(mock_post):
|
|
|
|
|
attach = AppriseAttachment(path) |
|
|
|
|
|
|
|
|
|
# Return our good configuration |
|
|
|
|
mock_post.side_effect = None |
|
|
|
|
mock_post.return_value = okay_response |
|
|
|
|
with mock.patch(builtin_open_function, side_effect=OSError()): |
|
|
|
|
with mock.patch('builtins.open', side_effect=OSError()): |
|
|
|
|
# We can't send the message we can't open the attachment for reading |
|
|
|
|
assert obj.notify( |
|
|
|
|
body='body', title='title', notify_type=NotifyType.INFO, |
|
|
|
@ -367,8 +395,8 @@ def test_notify_signal_plugin_attachments(mock_post):
|
|
|
|
|
assert isinstance(obj, NotifySignalAPI) |
|
|
|
|
|
|
|
|
|
# Now send an attachment normally without issues |
|
|
|
|
mock_post.reset_mock() |
|
|
|
|
request_mock.reset_mock() |
|
|
|
|
assert obj.notify( |
|
|
|
|
body='body', title='title', notify_type=NotifyType.INFO, |
|
|
|
|
attach=attach) is True |
|
|
|
|
assert mock_post.call_count == 1 |
|
|
|
|
assert request_mock.call_count == 1 |
|
|
|
|