extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-11-08 09:14:53 +01:00
Code Issues Packages Projects Releases Wiki Activity

Nextcloud Support (#186)

Browse Source
This commit is contained in:
Chris Caron 2019-12-12 19:37:05 -05:00 committed by GitHub
parent 6a45ed997c
commit d1a15978af
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 433 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -50,6 +50,7 @@ The table below identifies the services this tool supports and some example serv
| [Matrix](https://github.com/caronc/apprise/wiki/Notify_matrix) | matrix:// or matrixs:// | (TCP) 80 or 443 | matrix://hostname<br />matrix://user@hostname<br />matrixs://user:pass@hostname:port/#room_alias<br />matrixs://user:pass@hostname:port/!room_id<br />matrixs://user:pass@hostname:port/#room_alias/!room_id/#room2<br />matrixs://token@hostname:port/?webhook=matrix<br />matrix://user:token@hostname/?webhook=slack&format=markdown
| [Mattermost](https://github.com/caronc/apprise/wiki/Notify_mattermost) | mmost:// | (TCP) 8065 | mmost://hostname/authkey<br />mmost://hostname:80/authkey<br />mmost://user@hostname:80/authkey<br />mmost://hostname/authkey?channel=channel<br />mmosts://hostname/authkey<br />mmosts://user@hostname/authkey<br />
| [Microsoft Teams](https://github.com/caronc/apprise/wiki/Notify_msteams) | msteams:// | (TCP) 443 | msteams://TokenA/TokenB/TokenC/
| [Nextcloud](https://github.com/caronc/apprise/wiki/Notify_nextcloud) | ncloud:// or nclouds:// | (TCP) 80 or 443 | ncloud://adminuser:pass@host/User<br/>nclouds://adminuser:pass@host/User1/User2/UserN
| [Notica](https://github.com/caronc/apprise/wiki/Notify_notica) | notica:// | (TCP) 443 | notica://Token/
| [Notifico](https://github.com/caronc/apprise/wiki/Notify_notifico) | notifico:// | (TCP) 443 | notifico://ProjectID/MessageHook/
| [Prowl](https://github.com/caronc/apprise/wiki/Notify_prowl) | prowl:// | (TCP) 443 | prowl://apikey<br />prowl://apikey/providerkey

294
apprise/plugins/NotifyNextcloud.py Normal file
View File

@ -0,0 +1,294 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CON
import requests
from .NotifyBase import NotifyBase
from ..URLBase import PrivacyMode
from ..common import NotifyType
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
class NotifyNextcloud(NotifyBase):
"""
A wrapper for Nextcloud Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Nextcloud'
# The services URL
service_url = 'https://nextcloud.com/'
# Insecure protocol (for those self hosted requests)
protocol = 'ncloud'
# The default protocol (this is secure for notica)
secure_protocol = 'nclouds'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_nextcloud'
# Nextcloud URL
notify_url = '{schema}://{host}/ocs/v2.php/apps/admin_notifications/' \
'api/v1/notifications/{target}'
# Nextcloud does not support a title
title_maxlen = 255
# Defines the maximum allowable characters per message.
body_maxlen = 4000
# Define object templates
templates = (
'{schema}://{user}:{password}@{host}/{targets}',
'{schema}://{user}:{password}@{host}:{port}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'user': {
'name': _('Username'),
'type': 'string',
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
},
'target_user': {
'name': _('Target User'),
'type': 'string',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
'required': True,
},
})
# Define any kwargs we're using
template_kwargs = {
'headers': {
'name': _('HTTP Header'),
'prefix': '+',
},
}
def __init__(self, targets=None, headers=None, **kwargs):
"""
Initialize Nextcloud Object
"""
super(NotifyNextcloud, self).__init__(**kwargs)
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.headers = {}
if headers:
# Store our extra headers
self.headers.update(headers)
return
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Nextcloud Notification
"""
# Prepare our Header
headers = {
'User-Agent': self.app_id,
'OCS-APIREQUEST': 'true',
}
# Apply any/all header over-rides defined
headers.update(self.headers)
# error tracking (used for function return)
has_error = False
# Create a copy of the targets list
targets = list(self.targets)
while len(targets):
target = targets.pop(0)
# Prepare our Payload
payload = {
'shortMessage': title if title else self.app_desc,
}
if body:
# Only store the longMessage if a body was defined; nextcloud
# doesn't take kindly to empty longMessage entries.
payload['longMessage'] = body
auth = None
if self.user:
auth = (self.user, self.password)
notify_url = self.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),
target=target,
)
self.logger.debug('Nextcloud POST URL: %s (cert_verify=%r)' % (
notify_url, self.verify_certificate,
))
self.logger.debug('Nextcloud Payload: %s' % str(payload))
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.post(
notify_url,
data=payload,
headers=headers,
auth=auth,
verify=self.verify_certificate,
)
if r.status_code != requests.codes.ok:
# We had a problem
status_str = \
NotifyNextcloud.http_response_code_lookup(
r.status_code)
self.logger.warning(
'Failed to send Nextcloud notification:'
'{}{}error={}.'.format(
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug(
'Response Details:\r\n{}'.format(r.content))
# track our failure
has_error = True
continue
else:
self.logger.info('Sent Nextcloud notification.')
except requests.RequestException as e:
self.logger.warning(
'A Connection error occured sending Nextcloud '
'notification.',
)
self.logger.debug('Socket Exception: %s' % str(e))
# track our failure
has_error = True
continue
return not has_error
def url(self, privacy=False, *args, **kwargs):
"""
Returns the URL built dynamically based on specified arguments.
"""
# Define any arguments set
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
}
# Append our headers into our args
args.update({'+{}'.format(k): v for k, v in self.headers.items()})
# Determine Authentication
auth = ''
if self.user and self.password:
auth = '{user}:{password}@'.format(
user=NotifyNextcloud.quote(self.user, safe=''),
password=self.pprint(
self.password, privacy, mode=PrivacyMode.Secret, safe=''),
)
elif self.user:
auth = '{user}@'.format(
user=NotifyNextcloud.quote(self.user, safe=''),
)
default_port = 443 if self.secure else 80
return '{schema}://{auth}{hostname}{port}/{targets}?{args}' \
.format(
schema=self.secure_protocol
if self.secure else self.protocol,
auth=auth,
hostname=NotifyNextcloud.quote(self.host, safe=''),
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
targets='/'.join([NotifyNextcloud.quote(x)
for x in self.targets]),
args=NotifyNextcloud.urlencode(args),
)
@staticmethod
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to substantiate this object.
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
# Fetch our targets
results['targets'] = \
NotifyNextcloud.split_path(results['fullpath'])
# The 'to' makes it easier to use yaml configuration
if 'to' in results['qsd'] and len(results['qsd']['to']):
results['targets'] += \
NotifyNextcloud.parse_list(results['qsd']['to'])
# Add our headers that the user can potentially over-ride if they
# wish to to our returned result set
results['headers'] = results['qsd-']
results['headers'].update(results['qsd+'])
return results

9
packaging/redhat/python-apprise.spec
View File

@ -49,10 +49,11 @@ it easy to access:
Boxcar, ClickSend, Discord, E-Mail, Emby, Faast, Flock, Gitter, Gotify, Growl,
IFTTT, Join, KODI, Kumulos, Mailgun, MatterMost, Matrix, Microsoft Windows
Notifications, Microsoft Teams, MessageBird, MSG91, Nexmo, Notica, Notifico,
Notify MyAndroid, Prowl, Pushalot, PushBullet, Pushjet, Pushover, PushSafer,
Rocket.Chat, SendGrid, SimplePush, Slack, Super Toasty, Stride, Syslog,
Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC, XMPP, Webex Teams}
Notifications, Microsoft Teams, MessageBird, MSG91, Nexmo, Nextcloud, Notica,
Notifico, Notify MyAndroid, Prowl, Pushalot, PushBullet, Pushjet, Pushover,
PushSafer, Rocket.Chat, SendGrid, SimplePush, Slack, Super Toasty, Stride,
Syslog, Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC, XMPP,
Webex Teams}
Name: python-%{pypi_name}
Version: 0.8.2

8
setup.py
View File

@ -72,10 +72,10 @@ setup(
keywords='Push Notifications Alerts Email AWS SNS Boxcar ClickSend '
'Discord Dbus Emby Faast Flock Gitter Gnome Gotify Growl IFTTT Join '
'KODI Kumulos Mailgun Matrix Mattermost MessageBird MSG91 Nexmo '
'Notica, Notifico Prowl PushBullet Pushjet Pushed Pushover PushSafer '
'Rocket.Chat Ryver SendGrid SimplePush Slack Stride Syslog Techulus '
'Push Telegram Twilio Twist Twitter XBMC Microsoft MSTeams Windows '
'Webex CLI API',
'Nextcloud Notica, Notifico Prowl PushBullet Pushjet Pushed Pushover '
'PushSafer Rocket.Chat Ryver SendGrid SimplePush Slack Stride Syslog '
'Techulus Push Telegram Twilio Twist Twitter XBMC Microsoft MSTeams '
'Windows Webex CLI API',
author='Chris Caron',
author_email='lead2gold@gmail.com',
packages=find_packages(),

63
test/test_nextcloud_plugin.py Normal file
View File

@ -0,0 +1,63 @@
# -*- coding: utf-8 -*-
#
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import six
import mock
import requests
from apprise import plugins
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
@mock.patch('requests.post')
def test_nextcloud_empty_body(mock_post):
"""
API: NotifyNextcloud() empty body
"""
# Disable Throttling to speed testing
plugins.NotifyBase.request_rate_per_sec = 0
# A response
robj = mock.Mock()
robj.content = ''
robj.status_code = requests.codes.ok
# Prepare Mock
mock_post.return_value = robj
# Variation Initializations
obj = plugins.NotifyNextcloud(
host="localhost", user="admin", password="pass", targets="user")
assert isinstance(obj, plugins.NotifyNextcloud) is True
assert isinstance(obj.url(), six.string_types) is True
# An empty body
assert obj.send(body="") is True
assert 'data' in mock_post.call_args_list[0][1]
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']

66
test/test_rest_plugins.py
View File

@ -1737,6 +1737,72 @@ TEST_URLS = (
'test_requests_exceptions': True,
}),
##################################
# NotifyNextcloud
##################################
('ncloud://:@/', {
'instance': None,
}),
('ncloud://', {
'instance': None,
}),
('nclouds://', {
# No hostname
'instance': None,
}),
('ncloud://localhost', {
# No user specified
'instance': TypeError,
}),
('ncloud://localhost/admin', {
'instance': plugins.NotifyNextcloud,
}),
('ncloud://user@localhost/admin', {
'instance': plugins.NotifyNextcloud,
}),
('ncloud://user@localhost?to=user1,user2', {
'instance': plugins.NotifyNextcloud,
}),
('ncloud://user:pass@localhost/user1/user2', {
'instance': plugins.NotifyNextcloud,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'ncloud://user:****@localhost/user1/user2',
}),
('ncloud://user:pass@localhost:8080/admin', {
'instance': plugins.NotifyNextcloud,
}),
('nclouds://user:pass@localhost/admin', {
'instance': plugins.NotifyNextcloud,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'nclouds://user:****@localhost/admin',
}),
('nclouds://user:pass@localhost:8080/admin/', {
'instance': plugins.NotifyNextcloud,
}),
('ncloud://localhost:8080/admin?-HeaderKey=HeaderValue', {
'instance': plugins.NotifyNextcloud,
}),
('ncloud://user:pass@localhost:8081/admin', {
'instance': plugins.NotifyNextcloud,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
}),
('ncloud://user:pass@localhost:8082/admin', {
'instance': plugins.NotifyNextcloud,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('ncloud://user:pass@localhost:8083/user1/user2/user3', {
'instance': plugins.NotifyNextcloud,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
##################################
# NotifyProwl
##################################