extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2025-01-20 21:08:38 +01:00
Code Issues Packages Projects Releases Wiki Activity

Nextcloud and Nextcloud Talk url_prefix support (#975)

Browse Source
This commit is contained in:
Chris Caron 2023-10-15 13:15:52 -04:00 committed by GitHub
parent 34e52e5d92
commit f6b53ac556
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 182 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
apprise/plugins/NotifyNextcloud.py
View File

@ -114,6 +114,10 @@ class NotifyNextcloud(NotifyBase):
'min': 1,
'default': 21,
},
'url_prefix': {
'name': _('URL Prefix'),
'type': 'string',
},
'to': {
'alias_of': 'targets',
},
@ -127,17 +131,15 @@ class NotifyNextcloud(NotifyBase):
},
}
def __init__(self, targets=None, version=None, headers=None, **kwargs):
def __init__(self, targets=None, version=None, headers=None,
url_prefix=None, **kwargs):
"""
Initialize Nextcloud Object
"""
super().__init__(**kwargs)
# Store our targets
self.targets = parse_list(targets)
if len(self.targets) == 0:
msg = 'At least one Nextcloud target user must be specified.'
self.logger.warning(msg)
raise TypeError(msg)
self.version = self.template_args['version']['default']
if version is not None:
@ -153,6 +155,10 @@ class NotifyNextcloud(NotifyBase):
self.logger.warning(msg)
raise TypeError(msg)
# Support URL Prefix
self.url_prefix = '' if not url_prefix \
else url_prefix.strip('/')
self.headers = {}
if headers:
# Store our extra headers
@ -165,6 +171,11 @@ class NotifyNextcloud(NotifyBase):
Perform Nextcloud Notification
"""
if len(self.targets) == 0:
# There were no services to notify
self.logger.warning('There were no Nextcloud targets to notify.')
return False
# Prepare our Header
headers = {
'User-Agent': self.app_id,
@ -196,11 +207,11 @@ class NotifyNextcloud(NotifyBase):
auth = (self.user, self.password)
# Nextcloud URL based on version used
notify_url = '{schema}://{host}/ocs/v2.php/'\
notify_url = '{schema}://{host}/{url_prefix}/ocs/v2.php/'\
'apps/admin_notifications/' \
'api/v1/notifications/{target}' \
if self.version < 21 else \
'{schema}://{host}/ocs/v2.php/'\
'{schema}://{host}/{url_prefix}/ocs/v2.php/'\
'apps/notifications/'\
'api/v2/admin_notifications/{target}'
@ -208,6 +219,7 @@ class NotifyNextcloud(NotifyBase):
schema='https' if self.secure else 'http',
host=self.host if not isinstance(self.port, int)
else '{}:{}'.format(self.host, self.port),
url_prefix=self.url_prefix,
target=target,
)
@ -277,6 +289,9 @@ class NotifyNextcloud(NotifyBase):
# Set our version
params['version'] = str(self.version)
if self.url_prefix:
params['url_prefix'] = self.url_prefix
# Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
@ -314,7 +329,8 @@ class NotifyNextcloud(NotifyBase):
"""
Returns the number of targets associated with this notification
"""
return len(self.targets)
targets = len(self.targets)
return targets if targets else 1
@staticmethod
def parse_url(url):
@ -343,6 +359,12 @@ class NotifyNextcloud(NotifyBase):
results['version'] = \
NotifyNextcloud.unquote(results['qsd']['version'])
# Support URL Prefixes
if 'url_prefix' in results['qsd'] \
and len(results['qsd']['url_prefix']):
results['url_prefix'] = \
NotifyNextcloud.unquote(results['qsd']['url_prefix'])
# Add our headers that the user can potentially over-ride if they wish
# to to our returned result set and tidy entries by unquoting them
results['headers'] = {

51
apprise/plugins/NotifyNextcloudTalk.py
View File

@ -104,6 +104,14 @@ class NotifyNextcloudTalk(NotifyBase):
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'url_prefix': {
'name': _('URL Prefix'),
'type': 'string',
},
})
# Define any kwargs we're using
template_kwargs = {
'headers': {
@ -112,7 +120,7 @@ class NotifyNextcloudTalk(NotifyBase):
},
}
def __init__(self, targets=None, headers=None, **kwargs):
def __init__(self, targets=None, headers=None, url_prefix=None, **kwargs):
"""
Initialize Nextcloud Talk Object
"""
@ -123,11 +131,12 @@ class NotifyNextcloudTalk(NotifyBase):
self.logger.warning(msg)
raise TypeError(msg)
# Store our targets
self.targets = parse_list(targets)
if len(self.targets) == 0:
msg = 'At least one Nextcloud Talk Room ID must be specified.'
self.logger.warning(msg)
raise TypeError(msg)
# Support URL Prefix
self.url_prefix = '' if not url_prefix \
else url_prefix.strip('/')
self.headers = {}
if headers:
@ -141,6 +150,12 @@ class NotifyNextcloudTalk(NotifyBase):
Perform Nextcloud Talk Notification
"""
if len(self.targets) == 0:
# There were no services to notify
self.logger.warning(
'There were no Nextcloud Talk targets to notify.')
return False
# Prepare our Header
headers = {
'User-Agent': self.app_id,
@ -172,13 +187,14 @@ class NotifyNextcloudTalk(NotifyBase):
}
# Nextcloud Talk URL
notify_url = '{schema}://{host}'\
notify_url = '{schema}://{host}/{url_prefix}'\
'/ocs/v2.php/apps/spreed/api/v1/chat/{target}'
notify_url = notify_url.format(
schema='https' if self.secure else 'http',
host=self.host if not isinstance(self.port, int)
else '{}:{}'.format(self.host, self.port),
url_prefix=self.url_prefix,
target=target,
)
@ -201,7 +217,8 @@ class NotifyNextcloudTalk(NotifyBase):
verify=self.verify_certificate,
timeout=self.request_timeout,
)
if r.status_code != requests.codes.created:
if r.status_code not in (
requests.codes.created, requests.codes.ok):
# We had a problem
status_str = \
NotifyNextcloudTalk.http_response_code_lookup(
@ -241,6 +258,14 @@ class NotifyNextcloudTalk(NotifyBase):
Returns the URL built dynamically based on specified arguments.
"""
# Our default set of parameters
params = self.url_parameters(privacy=privacy, *args, **kwargs)
# Append our headers into our parameters
params.update({'+{}'.format(k): v for k, v in self.headers.items()})
if self.url_prefix:
params['url_prefix'] = self.url_prefix
# Determine Authentication
auth = '{user}:{password}@'.format(
user=NotifyNextcloudTalk.quote(self.user, safe=''),
@ -250,7 +275,7 @@ class NotifyNextcloudTalk(NotifyBase):
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}/{targets}' \
return '{schema}://{auth}{hostname}{port}/{targets}?{params}' \
.format(
schema=self.secure_protocol
if self.secure else self.protocol,
@ -262,13 +287,15 @@ class NotifyNextcloudTalk(NotifyBase):
else ':{}'.format(self.port),
targets='/'.join([NotifyNextcloudTalk.quote(x)
for x in self.targets]),
params=NotifyNextcloudTalk.urlencode(params),
)
def __len__(self):
"""
Returns the number of targets associated with this notification
"""
return len(self.targets)
targets = len(self.targets)
return targets if targets else 1
@staticmethod
def parse_url(url):
@ -287,6 +314,12 @@ class NotifyNextcloudTalk(NotifyBase):
results['targets'] = \
NotifyNextcloudTalk.split_path(results['fullpath'])
# Support URL Prefixes
if 'url_prefix' in results['qsd'] \
and len(results['qsd']['url_prefix']):
results['url_prefix'] = \
NotifyNextcloudTalk.unquote(results['qsd']['url_prefix'])
# Add our headers that the user can potentially over-ride if they wish
# to to our returned result set and tidy entries by unquoting them
results['headers'] = {

57
test/test_plugin_nextcloud.py
View File

@ -27,7 +27,8 @@
# POSSIBILITY OF SUCH DAMAGE.
from unittest import mock
from apprise import Apprise
from apprise import NotifyType
import requests
from apprise.plugins.NotifyNextcloud import NotifyNextcloud
from helpers import AppriseURLTester
@ -52,7 +53,10 @@ apprise_url_tests = (
}),
('ncloud://localhost', {
# No user specified
'instance': TypeError,
'instance': NotifyNextcloud,
# Since there are no targets specified we expect a False return on
# send()
'notify_response': False,
}),
('ncloud://user@localhost?to=user1,user2&version=invalid', {
# An invalid version was specified
@ -81,6 +85,12 @@ apprise_url_tests = (
('ncloud://user@localhost?to=user1,user2&version=21', {
'instance': NotifyNextcloud,
}),
('ncloud://user@localhost?to=user1&version=20&url_prefix=/abcd', {
'instance': NotifyNextcloud,
}),
('ncloud://user@localhost?to=user1&version=21&url_prefix=/abcd', {
'instance': NotifyNextcloud,
}),
('ncloud://user:pass@localhost/user1/user2', {
'instance': NotifyNextcloud,
@ -160,3 +170,46 @@ def test_plugin_nextcloud_edge_cases(mock_post):
assert 'shortMessage' in mock_post.call_args_list[0][1]['data']
# The longMessage argument is not set
assert 'longMessage' not in mock_post.call_args_list[0][1]['data']
@mock.patch('requests.post')
def test_plugin_nextcloud_url_prefix(mock_post):
"""
NotifyNextcloud() URL Prefix Testing
"""
response = mock.Mock()
response.content = ''
response.status_code = requests.codes.ok
# Prepare our mock object
mock_post.return_value = response
# instantiate our object (without a batch mode)
obj = Apprise.instantiate(
'ncloud://localhost/admin/?version=20&url_prefix=/abcd')
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
# Not set to batch, so we send 2 different messages
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'http://localhost/abcd/ocs/v2.php/apps/' \
'admin_notifications/api/v1/notifications/admin'
mock_post.reset_mock()
# instantiate our object (without a batch mode)
obj = Apprise.instantiate(
'ncloud://localhost/admin/?version=21&'
'url_prefix=a/longer/path/abcd/')
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
# Not set to batch, so we send 2 different messages
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'http://localhost/a/longer/path/abcd/' \
'ocs/v2.php/apps/notifications/api/v2/admin_notifications/admin'

60
test/test_plugin_nextcloudtalk.py
View File

@ -27,7 +27,8 @@
# POSSIBILITY OF SUCH DAMAGE.
from unittest import mock
from apprise import Apprise
from apprise import NotifyType
import requests
from apprise.plugins.NotifyNextcloudTalk import NotifyNextcloudTalk
from helpers import AppriseURLTester
@ -38,7 +39,7 @@ logging.disable(logging.CRITICAL)
apprise_url_tests = (
##################################
# NotifyNextcloud
# NotifyNextcloudTalk
##################################
('nctalk://:@/', {
'instance': None,
@ -64,7 +65,10 @@ apprise_url_tests = (
}),
('nctalk://user:pass@localhost', {
# No roomid specified
'instance': TypeError,
'instance': NotifyNextcloudTalk,
# Since there are no targets specified we expect a False return on
# send()
'notify_response': False,
}),
('nctalk://user:pass@localhost/roomid1/roomid2', {
'instance': NotifyNextcloudTalk,
@ -77,6 +81,10 @@ apprise_url_tests = (
'instance': NotifyNextcloudTalk,
'requests_response_code': requests.codes.created,
}),
('nctalk://user:pass@localhost:8080/roomid?url_prefix=/prefix', {
'instance': NotifyNextcloudTalk,
'requests_response_code': requests.codes.created,
}),
('nctalks://user:pass@localhost/roomid', {
'instance': NotifyNextcloudTalk,
'requests_response_code': requests.codes.created,
@ -115,7 +123,7 @@ apprise_url_tests = (
def test_plugin_nextcloudtalk_urls():
"""
NotifyNextcloud() Apprise URLs
NotifyNextcloudTalk() Apprise URLs
"""
@ -126,7 +134,7 @@ def test_plugin_nextcloudtalk_urls():
@mock.patch('requests.post')
def test_plugin_nextcloudtalk_edge_cases(mock_post):
"""
NotifyNextcloud() Edge Cases
NotifyNextcloudTalk() Edge Cases
"""
@ -148,3 +156,45 @@ def test_plugin_nextcloudtalk_edge_cases(mock_post):
assert obj.send(body="") is True
assert 'data' in mock_post.call_args_list[0][1]
assert 'message' in mock_post.call_args_list[0][1]['data']
@mock.patch('requests.post')
def test_plugin_nextcloud_talk_url_prefix(mock_post):
"""
NotifyNextcloudTalk() URL Prefix Testing
"""
response = mock.Mock()
response.content = ''
response.status_code = requests.codes.created
# Prepare our mock object
mock_post.return_value = response
# instantiate our object (without a batch mode)
obj = Apprise.instantiate(
'nctalk://user:pass@localhost/admin/?url_prefix=/abcd')
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
# Not set to batch, so we send 2 different messages
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'http://localhost/abcd/ocs/v2.php/apps/spreed/api/v1/chat/admin'
mock_post.reset_mock()
# instantiate our object (without a batch mode)
obj = Apprise.instantiate(
'nctalk://user:pass@localhost/admin/?'
'url_prefix=a/longer/path/abcd/')
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
# Not set to batch, so we send 2 different messages
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'http://localhost/a/longer/path/abcd/' \
'ocs/v2.php/apps/spreed/api/v1/chat/admin'