mirror of
https://github.com/caronc/apprise-api.git
synced 2024-12-13 10:21:14 +01:00
Honor new Group Functionality introduced in Apprise v1.6.0 (#147)
This commit is contained in:
parent
cbcdab3798
commit
53b341bab1
@ -31,6 +31,10 @@ import os
|
|||||||
import re
|
import re
|
||||||
import apprise
|
import apprise
|
||||||
import requests
|
import requests
|
||||||
|
import inspect
|
||||||
|
|
||||||
|
# Grant access to our Notification Manager Singleton
|
||||||
|
N_MGR = apprise.NotificationManager.NotificationManager()
|
||||||
|
|
||||||
|
|
||||||
class StatefulNotifyTests(SimpleTestCase):
|
class StatefulNotifyTests(SimpleTestCase):
|
||||||
@ -68,19 +72,19 @@ class StatefulNotifyTests(SimpleTestCase):
|
|||||||
mock_post.return_value = request
|
mock_post.return_value = request
|
||||||
|
|
||||||
# Monkey Patch
|
# Monkey Patch
|
||||||
apprise.plugins.NotifyEmail.NotifyEmail.enabled = True
|
N_MGR['mailto'].enabled = True
|
||||||
|
|
||||||
# Preare our list of URLs we want to save
|
# Preare our list of URLs we want to save
|
||||||
urls = [
|
urls = [
|
||||||
'devops=slack://TokenA/TokenB/TokenC',
|
'devops=slack://TokenA/TokenB/TokenC',
|
||||||
'pusbullet=pbul://tokendetails',
|
'pushbullet=pbul://tokendetails',
|
||||||
'general,json=json://hostname',
|
'general,json=json://hostname',
|
||||||
]
|
]
|
||||||
|
|
||||||
# Monkey Patch
|
# Monkey Patch
|
||||||
apprise.plugins.NotifySlack.NotifySlack.enabled = True
|
N_MGR['slack'].enabled = True
|
||||||
apprise.plugins.NotifyPushBullet.NotifyPushBullet.enabled = True
|
N_MGR['pbul'].enabled = True
|
||||||
apprise.plugins.NotifyJSON.NotifyJSON.enabled = True
|
N_MGR['json'].enabled = True
|
||||||
|
|
||||||
# For 10 iterations, repeat these tests to verify that don't change
|
# For 10 iterations, repeat these tests to verify that don't change
|
||||||
# and our saved content is not different on subsequent calls.
|
# and our saved content is not different on subsequent calls.
|
||||||
@ -301,3 +305,197 @@ class StatefulNotifyTests(SimpleTestCase):
|
|||||||
|
|
||||||
# Reset our count
|
# Reset our count
|
||||||
mock_post.reset_mock()
|
mock_post.reset_mock()
|
||||||
|
|
||||||
|
@patch('requests.post')
|
||||||
|
def test_stateful_group_dict_notify(self, mock_post):
|
||||||
|
"""
|
||||||
|
Test the handling of a group defined as a dictionary
|
||||||
|
"""
|
||||||
|
|
||||||
|
# our key to use
|
||||||
|
key = 'test_stateful_group_notify_dict'
|
||||||
|
|
||||||
|
request = Mock()
|
||||||
|
request.content = b'ok'
|
||||||
|
request.status_code = requests.codes.ok
|
||||||
|
mock_post.return_value = request
|
||||||
|
|
||||||
|
# Monkey Patch
|
||||||
|
N_MGR['mailto'].enabled = True
|
||||||
|
|
||||||
|
config = inspect.cleandoc("""
|
||||||
|
version: 1
|
||||||
|
groups:
|
||||||
|
mygroup: user1, user2
|
||||||
|
|
||||||
|
urls:
|
||||||
|
- json:///user:pass@localhost:
|
||||||
|
- to: user1@example.com
|
||||||
|
tag: user1
|
||||||
|
- to: user2@example.com
|
||||||
|
tag: user2
|
||||||
|
""")
|
||||||
|
|
||||||
|
# Monkey Patch
|
||||||
|
N_MGR['json'].enabled = True
|
||||||
|
|
||||||
|
# Add our content
|
||||||
|
response = self.client.post(
|
||||||
|
'/add/{}'.format(key),
|
||||||
|
{'config': config})
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# Now we should be able to see our content
|
||||||
|
response = self.client.post('/get/{}'.format(key))
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
for tag in ('user1', 'user2'):
|
||||||
|
form_data = {
|
||||||
|
'body': '## test notifiction',
|
||||||
|
'format': apprise.NotifyFormat.MARKDOWN,
|
||||||
|
'tag': tag,
|
||||||
|
}
|
||||||
|
form = NotifyForm(data=form_data)
|
||||||
|
assert form.is_valid()
|
||||||
|
|
||||||
|
# Required to prevent None from being passed into
|
||||||
|
# self.client.post()
|
||||||
|
del form.cleaned_data['attachment']
|
||||||
|
|
||||||
|
# We sent the notification successfully
|
||||||
|
response = self.client.post(
|
||||||
|
'/notify/{}'.format(key), form.cleaned_data)
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# Our single endpoint is notified
|
||||||
|
assert mock_post.call_count == 1
|
||||||
|
|
||||||
|
mock_post.reset_mock()
|
||||||
|
|
||||||
|
# Now let's notify by our group
|
||||||
|
form_data = {
|
||||||
|
'body': '## test notifiction',
|
||||||
|
'format': apprise.NotifyFormat.MARKDOWN,
|
||||||
|
'tag': 'mygroup',
|
||||||
|
}
|
||||||
|
|
||||||
|
form = NotifyForm(data=form_data)
|
||||||
|
assert form.is_valid()
|
||||||
|
|
||||||
|
# Required to prevent None from being passed into
|
||||||
|
# self.client.post()
|
||||||
|
del form.cleaned_data['attachment']
|
||||||
|
|
||||||
|
# We sent the notification successfully
|
||||||
|
response = self.client.post(
|
||||||
|
'/notify/{}'.format(key), form.cleaned_data)
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# Our 2 endpoints are notified
|
||||||
|
assert mock_post.call_count == 2
|
||||||
|
|
||||||
|
mock_post.reset_mock()
|
||||||
|
|
||||||
|
# Now empty our data
|
||||||
|
response = self.client.post('/del/{}'.format(key))
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# Reset our count
|
||||||
|
mock_post.reset_mock()
|
||||||
|
|
||||||
|
@patch('requests.post')
|
||||||
|
def test_stateful_group_dictlist_notify(self, mock_post):
|
||||||
|
"""
|
||||||
|
Test the handling of a group defined as a list of dictionaries
|
||||||
|
"""
|
||||||
|
|
||||||
|
# our key to use
|
||||||
|
key = 'test_stateful_group_notify_list_dict'
|
||||||
|
|
||||||
|
request = Mock()
|
||||||
|
request.content = b'ok'
|
||||||
|
request.status_code = requests.codes.ok
|
||||||
|
mock_post.return_value = request
|
||||||
|
|
||||||
|
# Monkey Patch
|
||||||
|
N_MGR['mailto'].enabled = True
|
||||||
|
|
||||||
|
config = inspect.cleandoc("""
|
||||||
|
version: 1
|
||||||
|
groups:
|
||||||
|
- mygroup: user1, user2
|
||||||
|
|
||||||
|
urls:
|
||||||
|
- json:///user:pass@localhost:
|
||||||
|
- to: user1@example.com
|
||||||
|
tag: user1
|
||||||
|
- to: user2@example.com
|
||||||
|
tag: user2
|
||||||
|
""")
|
||||||
|
|
||||||
|
# Monkey Patch
|
||||||
|
N_MGR['json'].enabled = True
|
||||||
|
|
||||||
|
# Add our content
|
||||||
|
response = self.client.post(
|
||||||
|
'/add/{}'.format(key),
|
||||||
|
{'config': config})
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# Now we should be able to see our content
|
||||||
|
response = self.client.post('/get/{}'.format(key))
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
for tag in ('user1', 'user2'):
|
||||||
|
form_data = {
|
||||||
|
'body': '## test notifiction',
|
||||||
|
'format': apprise.NotifyFormat.MARKDOWN,
|
||||||
|
'tag': tag,
|
||||||
|
}
|
||||||
|
form = NotifyForm(data=form_data)
|
||||||
|
assert form.is_valid()
|
||||||
|
|
||||||
|
# Required to prevent None from being passed into
|
||||||
|
# self.client.post()
|
||||||
|
del form.cleaned_data['attachment']
|
||||||
|
|
||||||
|
# We sent the notification successfully
|
||||||
|
response = self.client.post(
|
||||||
|
'/notify/{}'.format(key), form.cleaned_data)
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# Our single endpoint is notified
|
||||||
|
assert mock_post.call_count == 1
|
||||||
|
|
||||||
|
mock_post.reset_mock()
|
||||||
|
|
||||||
|
# Now let's notify by our group
|
||||||
|
form_data = {
|
||||||
|
'body': '## test notifiction',
|
||||||
|
'format': apprise.NotifyFormat.MARKDOWN,
|
||||||
|
'tag': 'mygroup',
|
||||||
|
}
|
||||||
|
|
||||||
|
form = NotifyForm(data=form_data)
|
||||||
|
assert form.is_valid()
|
||||||
|
|
||||||
|
# Required to prevent None from being passed into
|
||||||
|
# self.client.post()
|
||||||
|
del form.cleaned_data['attachment']
|
||||||
|
|
||||||
|
# We sent the notification successfully
|
||||||
|
response = self.client.post(
|
||||||
|
'/notify/{}'.format(key), form.cleaned_data)
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# Our 2 endpoints are notified
|
||||||
|
assert mock_post.call_count == 2
|
||||||
|
|
||||||
|
mock_post.reset_mock()
|
||||||
|
|
||||||
|
# Now empty our data
|
||||||
|
response = self.client.post('/del/{}'.format(key))
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# Reset our count
|
||||||
|
mock_post.reset_mock()
|
||||||
|
Loading…
Reference in New Issue
Block a user