mirror of
https://github.com/caronc/apprise.git
synced 2024-11-24 09:03:29 +01:00
Added LunaSea Support (#1072)
This commit is contained in:
Chris Caron
GitHub
parent
de1cb522f7
commit
108da1e288
@ -77,6 +77,7 @@ The table below identifies the services this tool supports and some example serv
|
||||
| [Kumulos](https://github.com/caronc/apprise/wiki/Notify_kumulos) | kumulos:// | (TCP) 443 | kumulos://apikey/serverkey
|
||||
| [LaMetric Time](https://github.com/caronc/apprise/wiki/Notify_lametric) | lametric:// | (TCP) 443 | lametric://apikey@device_ipaddr<br/>lametric://apikey@hostname:port<br/>lametric://client_id@client_secret
|
||||
| [Line](https://github.com/caronc/apprise/wiki/Notify_line) | line:// | (TCP) 443 | line://Token@User<br/>line://Token/User1/User2/UserN
|
||||
| [LunaSea](https://github.com/caronc/apprise/wiki/Notify_lunasea) | lunasea:// | (TCP) 80 or 443 | lunasea://user:pass@+FireBaseDevice/<br/>lunasea://user:pass@FireBaseUser/<br/>lunasea://user:pass@hostname/+FireBaseDevice/<br/>lunasea://user:pass@hostname/@FireBaseUser/
|
||||
| [Mailgun](https://github.com/caronc/apprise/wiki/Notify_mailgun) | mailgun:// | (TCP) 443 | mailgun://user@hostname/apikey<br />mailgun://user@hostname/apikey/email<br />mailgun://user@hostname/apikey/email1/email2/emailN<br />mailgun://user@hostname/apikey/?name="From%20User"
|
||||
| [Mastodon](https://github.com/caronc/apprise/wiki/Notify_mastodon) | mastodon:// or mastodons://| (TCP) 80 or 443 | mastodon://access_key@hostname<br />mastodon://access_key@hostname/@user<br />mastodon://access_key@hostname/@user1/@user2/@userN
|
||||
| [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
|
||||
|
||||
440
apprise/plugins/NotifyLunaSea.py
Normal file
440
apprise/plugins/NotifyLunaSea.py
Normal file
@ -0,0 +1,440 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# BSD 2-Clause License
|
||||
#
|
||||
# Apprise - Push Notification Library.
|
||||
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
|
||||
#
|
||||
# Redistribution and use in source and binary forms, with or without
|
||||
# modification, are permitted provided that the following conditions are met:
|
||||
#
|
||||
# 1. Redistributions of source code must retain the above copyright notice,
|
||||
# this list of conditions and the following disclaimer.
|
||||
#
|
||||
# 2. Redistributions in binary form must reproduce the above copyright notice,
|
||||
# this list of conditions and the following disclaimer in the documentation
|
||||
# and/or other materials provided with the distribution.
|
||||
#
|
||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
|
||||
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
||||
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
||||
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
|
||||
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
|
||||
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
|
||||
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
|
||||
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
|
||||
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
|
||||
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
||||
# POSSIBILITY OF SUCH DAMAGE.
|
||||
#
|
||||
# API:
|
||||
# https://docs.lunasea.app/lunasea/notifications/custom-notifications
|
||||
#
|
||||
import re
|
||||
import requests
|
||||
from json import dumps
|
||||
|
||||
from .NotifyBase import NotifyBase
|
||||
from ..common import NotifyType
|
||||
from ..common import NotifyImageSize
|
||||
from ..utils import parse_list
|
||||
from ..utils import is_hostname
|
||||
from ..utils import is_ipaddr
|
||||
from ..utils import parse_bool
|
||||
from ..AppriseLocale import gettext_lazy as _
|
||||
from ..URLBase import PrivacyMode
|
||||
|
||||
|
||||
class LunaSeaMode:
|
||||
"""
|
||||
Define LunaSea Notification Modes
|
||||
"""
|
||||
# App posts upstream to the developer API on LunaSea's website
|
||||
CLOUD = "cloud"
|
||||
|
||||
# Running a dedicated private ntfy Server
|
||||
PRIVATE = "private"
|
||||
|
||||
|
||||
LUNASEA_MODES = (
|
||||
LunaSeaMode.CLOUD,
|
||||
LunaSeaMode.PRIVATE,
|
||||
)
|
||||
|
||||
|
||||
class NotifyLunaSea(NotifyBase):
|
||||
"""
|
||||
A wrapper for LunaSea Notifications
|
||||
"""
|
||||
|
||||
# The default descriptive name associated with the Notification
|
||||
service_name = 'LunaSea'
|
||||
|
||||
# The services URL
|
||||
service_url = 'https://luasea.app'
|
||||
|
||||
# The default insecure protocol
|
||||
protocol = ('lunasea', 'lsea')
|
||||
|
||||
# The default secure protocol
|
||||
secure_protocol = ('lunaseas', 'lseas')
|
||||
|
||||
# A URL that takes you to the setup/help of the specific protocol
|
||||
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_lunasea'
|
||||
|
||||
# Allows the user to specify the NotifyImageSize object
|
||||
image_size = NotifyImageSize.XY_256
|
||||
|
||||
# LunaSea Notification Details
|
||||
cloud_notify_url = 'https://notify.lunasea.app'
|
||||
notify_user_path = '/v1/custom/user/{}'
|
||||
notify_device_path = '/v1/custom/device/{}'
|
||||
|
||||
# if our hostname matches the following we automatically enforce
|
||||
# cloud mode
|
||||
__auto_cloud_host = re.compile(r'(notify\.)?lunasea\.app', re.IGNORECASE)
|
||||
|
||||
# Define object templates
|
||||
templates = (
|
||||
'{schema}://{targets}',
|
||||
'{schema}://{host}/{targets}',
|
||||
'{schema}://{host}:{port}/{targets}',
|
||||
'{schema}://{user}@{host}/{targets}',
|
||||
'{schema}://{user}@{host}:{port}/{targets}',
|
||||
'{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',
|
||||
},
|
||||
'port': {
|
||||
'name': _('Port'),
|
||||
'type': 'int',
|
||||
'min': 1,
|
||||
'max': 65535,
|
||||
},
|
||||
'user': {
|
||||
'name': _('Username'),
|
||||
'type': 'string',
|
||||
},
|
||||
'password': {
|
||||
'name': _('Password'),
|
||||
'type': 'string',
|
||||
'private': True,
|
||||
},
|
||||
'token': {
|
||||
'name': _('Token'),
|
||||
'type': 'string',
|
||||
'private': True,
|
||||
},
|
||||
'target_user': {
|
||||
'name': _('Target User'),
|
||||
'type': 'string',
|
||||
'prefix': '@',
|
||||
'map_to': 'targets',
|
||||
},
|
||||
'target_device': {
|
||||
'name': _('Target Device'),
|
||||
'type': 'string',
|
||||
'prefix': '+',
|
||||
'map_to': 'targets',
|
||||
},
|
||||
'targets': {
|
||||
'name': _('Targets'),
|
||||
'type': 'list:string',
|
||||
'required': True,
|
||||
},
|
||||
})
|
||||
|
||||
# Define our template arguments
|
||||
template_args = dict(NotifyBase.template_args, **{
|
||||
'to': {
|
||||
'alias_of': 'targets',
|
||||
},
|
||||
'image': {
|
||||
'name': _('Include Image'),
|
||||
'type': 'bool',
|
||||
'default': False,
|
||||
'map_to': 'include_image',
|
||||
},
|
||||
'mode': {
|
||||
'name': _('Mode'),
|
||||
'type': 'choice:string',
|
||||
'values': LUNASEA_MODES,
|
||||
'default': LunaSeaMode.PRIVATE,
|
||||
},
|
||||
})
|
||||
|
||||
def __init__(self, targets=None, mode=None, token=None,
|
||||
include_image=False, **kwargs):
|
||||
"""
|
||||
Initialize LunaSea Object
|
||||
"""
|
||||
super().__init__(**kwargs)
|
||||
|
||||
# Show image associated with notification
|
||||
self.include_image = \
|
||||
self.template_args['image']['default'] \
|
||||
if include_image is None else include_image
|
||||
|
||||
# Prepare our mode
|
||||
self.mode = mode.strip().lower() \
|
||||
if isinstance(mode, str) \
|
||||
else self.template_args['mode']['default']
|
||||
|
||||
if self.mode not in LUNASEA_MODES:
|
||||
msg = 'An invalid LunaSea mode ({}) was specified.'.format(mode)
|
||||
self.logger.warning(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
self.targets = []
|
||||
for target in parse_list(targets):
|
||||
if len(target) < 4:
|
||||
self.logger.warning(
|
||||
'A specified target ({}) is invalid and will be '
|
||||
'ignored'.format(target))
|
||||
continue
|
||||
|
||||
if target[0] == '+':
|
||||
# Device
|
||||
self.targets.append(('+', target[1:]))
|
||||
|
||||
elif target[0] == '@':
|
||||
# User
|
||||
self.targets.append(('@', target[1:]))
|
||||
|
||||
else:
|
||||
# User
|
||||
self.targets.append(('@', target))
|
||||
|
||||
return
|
||||
|
||||
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
|
||||
"""
|
||||
Perform LunaSea Notification
|
||||
"""
|
||||
|
||||
# error tracking (used for function return)
|
||||
has_error = False
|
||||
|
||||
if not len(self.targets):
|
||||
# We have nothing to notify; we're done
|
||||
self.logger.warning('There are no LunaSea targets to notify')
|
||||
return False
|
||||
|
||||
# Prepare our headers
|
||||
headers = {
|
||||
'User-Agent': self.app_id,
|
||||
'Content-Type': 'application/json',
|
||||
'Accept': 'application/json',
|
||||
}
|
||||
|
||||
# prepare payload
|
||||
payload = {
|
||||
'title': title if title else self.app_desc,
|
||||
'body': body,
|
||||
}
|
||||
|
||||
# Acquire image_url
|
||||
image_url = None if not self.include_image \
|
||||
else self.image_url(notify_type)
|
||||
|
||||
if image_url:
|
||||
payload['image'] = image_url
|
||||
|
||||
# Prepare our Authentication (if defined)
|
||||
if self.user and self.password:
|
||||
auth = (self.user, self.password)
|
||||
|
||||
else:
|
||||
# No Auth
|
||||
auth = None
|
||||
|
||||
if self.mode == LunaSeaMode.CLOUD:
|
||||
# Cloud Service
|
||||
notify_url = self.cloud_notify_url
|
||||
|
||||
else:
|
||||
# Local Hosting
|
||||
schema = 'https' if self.secure else 'http'
|
||||
|
||||
notify_url = '%s://%s' % (schema, self.host)
|
||||
if isinstance(self.port, int):
|
||||
notify_url += ':%d' % self.port
|
||||
|
||||
# Create a copy of the targets list
|
||||
targets = list(self.targets)
|
||||
while len(targets):
|
||||
target = targets.pop(0)
|
||||
|
||||
if target[0] == '+':
|
||||
url = notify_url + self.notify_device_path.format(target[1])
|
||||
|
||||
else:
|
||||
url = notify_url + self.notify_user_path.format(target[1])
|
||||
|
||||
self.logger.debug('LunaSea POST URL: %s (cert_verify=%r)' % (
|
||||
url, self.verify_certificate,
|
||||
))
|
||||
self.logger.debug('LunaSea Payload: %s' % str(payload))
|
||||
|
||||
# Always call throttle before any remote server i/o is made
|
||||
self.throttle()
|
||||
|
||||
try:
|
||||
r = requests.post(
|
||||
url,
|
||||
data=dumps(payload),
|
||||
headers=headers,
|
||||
auth=auth,
|
||||
verify=self.verify_certificate,
|
||||
timeout=self.request_timeout,
|
||||
)
|
||||
|
||||
if r.status_code not in (
|
||||
requests.codes.ok, requests.codes.no_content):
|
||||
# We had a problem
|
||||
status_str = \
|
||||
NotifyLunaSea.http_response_code_lookup(r.status_code)
|
||||
|
||||
self.logger.warning(
|
||||
'Failed to deliver payload to LunaSea:'
|
||||
'{}{}error={}.'.format(
|
||||
status_str,
|
||||
', ' if status_str else '',
|
||||
r.status_code))
|
||||
|
||||
self.logger.debug(
|
||||
'Response Details:\r\n{}'.format(r.content))
|
||||
|
||||
has_error = True
|
||||
|
||||
# otherwise we were successful
|
||||
continue
|
||||
|
||||
except requests.RequestException as e:
|
||||
self.logger.warning(
|
||||
'A Connection error occurred communicating with LunaSea.')
|
||||
self.logger.debug('Socket Exception: %s' % str(e))
|
||||
|
||||
has_error = True
|
||||
|
||||
return not has_error
|
||||
|
||||
def url(self, privacy=False, *args, **kwargs):
|
||||
"""
|
||||
Returns the URL built dynamically based on specified arguments.
|
||||
"""
|
||||
|
||||
params = {
|
||||
'mode': self.mode,
|
||||
'image': 'yes' if self.include_image else 'no',
|
||||
}
|
||||
|
||||
# Our URL parameters
|
||||
params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
|
||||
|
||||
auth = ''
|
||||
if self.user and self.password:
|
||||
auth = '{user}:{password}@'.format(
|
||||
user=NotifyLunaSea.quote(self.user, safe=''),
|
||||
password=self.pprint(
|
||||
self.password, privacy, mode=PrivacyMode.Secret,
|
||||
safe=''),
|
||||
)
|
||||
elif self.user:
|
||||
auth = '{user}@'.format(
|
||||
user=NotifyLunaSea.quote(self.user, safe=''),
|
||||
)
|
||||
|
||||
if self.mode == LunaSeaMode.PRIVATE:
|
||||
default_port = 443 if self.secure else 80
|
||||
return '{schema}://{auth}{host}{port}/{targets}?{params}'.format(
|
||||
schema=self.secure_protocol[0]
|
||||
if self.secure else self.protocol[0],
|
||||
auth=auth,
|
||||
host=self.host,
|
||||
port='' if self.port is None or self.port == default_port
|
||||
else ':{}'.format(self.port),
|
||||
targets='/'.join(
|
||||
[NotifyLunaSea.quote(x[0] + x[1], safe='@+')
|
||||
for x in self.targets]),
|
||||
params=NotifyLunaSea.urlencode(params)
|
||||
)
|
||||
|
||||
else: # Cloud mode
|
||||
return '{schema}://{auth}{targets}?{params}'.format(
|
||||
schema=self.protocol[0],
|
||||
auth=auth,
|
||||
targets='/'.join(
|
||||
[NotifyLunaSea.quote(x[0] + x[1], safe='@+')
|
||||
for x in self.targets]),
|
||||
params=NotifyLunaSea.urlencode(params)
|
||||
)
|
||||
|
||||
def __len__(self):
|
||||
"""
|
||||
Returns the number of targets associated with this notification
|
||||
"""
|
||||
# always return 1
|
||||
return 1 if not self.targets else len(self.targets)
|
||||
|
||||
@staticmethod
|
||||
def parse_url(url):
|
||||
"""
|
||||
Parses the URL and returns enough arguments that can allow
|
||||
us to re-instantiate this object.
|
||||
|
||||
"""
|
||||
results = NotifyBase.parse_url(url, verify_host=False)
|
||||
if not results:
|
||||
# We're done early as we couldn't load the results
|
||||
return results
|
||||
|
||||
# Fetch our targets
|
||||
results['targets'] = NotifyLunaSea.split_path(results['fullpath'])
|
||||
|
||||
# Boolean to include an image or not
|
||||
results['include_image'] = parse_bool(results['qsd'].get(
|
||||
'image', NotifyLunaSea.template_args['image']['default']))
|
||||
|
||||
# The 'to' makes it easier to use yaml configuration
|
||||
if 'to' in results['qsd'] and len(results['qsd']['to']):
|
||||
results['targets'] += \
|
||||
NotifyLunaSea.parse_list(results['qsd']['to'])
|
||||
|
||||
# Mode override
|
||||
if 'mode' in results['qsd'] and results['qsd']['mode']:
|
||||
results['mode'] = NotifyLunaSea.unquote(
|
||||
results['qsd']['mode'].strip().lower())
|
||||
|
||||
else:
|
||||
# We can try to detect the mode based on the validity of the
|
||||
# hostname.
|
||||
#
|
||||
# This isn't a surfire way to do things though; it's best to
|
||||
# specify the mode= flag
|
||||
results['mode'] = LunaSeaMode.PRIVATE \
|
||||
if ((is_hostname(results['host'])
|
||||
or is_ipaddr(results['host'])) and results['targets']) \
|
||||
else LunaSeaMode.CLOUD
|
||||
|
||||
if results['mode'] == LunaSeaMode.CLOUD:
|
||||
# Store first entry as it can be a topic too in this case
|
||||
# But only if we also rule it out not being the words
|
||||
# lunasea.app itself, something that starts wiht an non-alpha
|
||||
# numeric character:
|
||||
if not NotifyLunaSea.__auto_cloud_host.search(results['host']):
|
||||
# Add it to the front of the list for consistency
|
||||
results['targets'].insert(0, results['host'])
|
||||
|
||||
elif results['mode'] == LunaSeaMode.PRIVATE and \
|
||||
not (is_hostname(results['host'] or
|
||||
is_ipaddr(results['host']))):
|
||||
# Invalid Host for LunaSeaMode.PRIVATE
|
||||
return None
|
||||
|
||||
return results
|
||||
@ -42,7 +42,7 @@ it easy to access:
|
||||
Apprise API, APRS, AWS SES, AWS SNS, Bark, Boxcar, Burst SMS, BulkSMS, BulkVS,
|
||||
ClickSend, DAPNET, DingTalk, Discord, E-Mail, Emby, Faast, FCM, Flock,
|
||||
Google Chat, Gotify, Growl, Guilded, Home Assistant, httpSMS, IFTTT, Join,
|
||||
Kavenegar, KODI, Kumulos, LaMetric, Line, MacOSX, Mailgun, Mastodon,
|
||||
Kavenegar, KODI, Kumulos, LaMetric, Line, LunaSea, MacOSX, Mailgun, Mastodon,
|
||||
Mattermost,Matrix, MessageBird, Microsoft Windows, Microsoft Teams, Misskey,
|
||||
MQTT, MSG91, MyAndroid, Nexmo, Nextcloud, NextcloudTalk, Notica, Notifiarr,
|
||||
Notifico, ntfy, Office365, OneSignal, Opsgenie, PagerDuty, PagerTree,
|
||||
|
||||
280
test/test_plugin_lunasea.py
Normal file
280
test/test_plugin_lunasea.py
Normal file
@ -0,0 +1,280 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# BSD 2-Clause License
|
||||
#
|
||||
# Apprise - Push Notification Library.
|
||||
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
|
||||
#
|
||||
# Redistribution and use in source and binary forms, with or without
|
||||
# modification, are permitted provided that the following conditions are met:
|
||||
#
|
||||
# 1. Redistributions of source code must retain the above copyright notice,
|
||||
# this list of conditions and the following disclaimer.
|
||||
#
|
||||
# 2. Redistributions in binary form must reproduce the above copyright notice,
|
||||
# this list of conditions and the following disclaimer in the documentation
|
||||
# and/or other materials provided with the distribution.
|
||||
#
|
||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
|
||||
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
||||
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
||||
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
|
||||
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
|
||||
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
|
||||
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
|
||||
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
|
||||
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
|
||||
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
||||
# POSSIBILITY OF SUCH DAMAGE.
|
||||
|
||||
import os
|
||||
from unittest import mock
|
||||
from json import loads
|
||||
|
||||
import requests
|
||||
from helpers import AppriseURLTester
|
||||
|
||||
from apprise.plugins.NotifyLunaSea import NotifyLunaSea
|
||||
|
||||
# Disable logging for a cleaner testing output
|
||||
import logging
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
# Attachment Directory
|
||||
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')
|
||||
|
||||
# Our Testing URLs
|
||||
apprise_url_tests = (
|
||||
('lunasea://', {
|
||||
# Initializes okay (as cloud mode) but has no targets to notify
|
||||
'instance': NotifyLunaSea,
|
||||
# invalid targets specified (nothing to notify)
|
||||
# as a result the response type will be false
|
||||
'response': False,
|
||||
}),
|
||||
('lunaseas://44$$$$%3012/?mode=private', {
|
||||
# Private mode initialization with a horrible hostname
|
||||
'instance': None
|
||||
}),
|
||||
('lunasea://:@/', {
|
||||
# Initializes okay (as cloud mode) but has no targets to notify
|
||||
'instance': NotifyLunaSea,
|
||||
# invalid targets specified (nothing to notify)
|
||||
# as a result the response type will be false
|
||||
'response': False,
|
||||
}),
|
||||
# No targets
|
||||
('lunasea://user:pass@localhost?mode=private', {
|
||||
'instance': NotifyLunaSea,
|
||||
# invalid targets specified (nothing to notify)
|
||||
# as a result the response type will be false
|
||||
'response': False,
|
||||
}),
|
||||
# No valid targets
|
||||
('lunasea://user:pass@localhost/#/!/@', {
|
||||
'instance': NotifyLunaSea,
|
||||
# invalid targets specified (nothing to notify)
|
||||
# as a result the response type will be false
|
||||
'response': False,
|
||||
}),
|
||||
# user/pass combos
|
||||
('lunasea://user@localhost/@user/', {
|
||||
'instance': NotifyLunaSea,
|
||||
# Our expected url(privacy=True) startswith() response:
|
||||
'privacy_url': 'lunasea://user@localhost/@user',
|
||||
}),
|
||||
# LunaSea cloud mode (enforced)
|
||||
('lunasea://lunasea.app/@user/+device/', {
|
||||
'instance': NotifyLunaSea,
|
||||
}),
|
||||
# No user/pass combo
|
||||
('lunasea://localhost/@user/@user2/?image=True', {
|
||||
'instance': NotifyLunaSea,
|
||||
}),
|
||||
# Enforce image but not otherwise find one
|
||||
('lunasea://localhost/+device/?image=True', {
|
||||
'instance': NotifyLunaSea,
|
||||
'include_image': False,
|
||||
}),
|
||||
# No images
|
||||
('lunasea://localhost/+device/?image=False', {
|
||||
'instance': NotifyLunaSea,
|
||||
}),
|
||||
('lunaseas://user:pass@localhost?to=+device', {
|
||||
'instance': NotifyLunaSea,
|
||||
# The response text is expected to be the following on a success
|
||||
}),
|
||||
('https://just/a/random/host/that/means/nothing', {
|
||||
# Nothing transpires from this
|
||||
'instance': None
|
||||
}),
|
||||
# Several targets
|
||||
('lunasea://user:pass@+device/user/@user2/?mode=cloud', {
|
||||
'instance': NotifyLunaSea,
|
||||
# The response text is expected to be the following on a success
|
||||
}),
|
||||
# Several targets (but do not add lunasea.app)
|
||||
('lunasea://user:pass@lunasea.app/user1/user2/?mode=cloud', {
|
||||
'instance': NotifyLunaSea,
|
||||
# The response text is expected to be the following on a success
|
||||
}),
|
||||
('lunaseas://user:web/token@localhost/user/?mode=invalid', {
|
||||
# Invalid mode
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('lunasea://user:pass@localhost:8089/+device/user1', {
|
||||
'instance': NotifyLunaSea,
|
||||
# force a failure using basic mode
|
||||
'response': False,
|
||||
'requests_response_code': requests.codes.internal_server_error,
|
||||
}),
|
||||
('lunasea://user:pass@localhost:8082/+device', {
|
||||
'instance': NotifyLunaSea,
|
||||
# throw a bizzare code forcing us to fail to look it up
|
||||
'response': False,
|
||||
'requests_response_code': 999,
|
||||
}),
|
||||
('lunasea://user:pass@localhost:8083/user1/user2/', {
|
||||
'instance': NotifyLunaSea,
|
||||
# Throws a series of connection and transfer exceptions when this flag
|
||||
# is set and tests that we gracfully handle them
|
||||
'test_requests_exceptions': True,
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
def test_plugin_lunasea_urls():
|
||||
"""
|
||||
NotifyLunaSea() Apprise URLs
|
||||
|
||||
"""
|
||||
|
||||
# Run our general tests
|
||||
AppriseURLTester(tests=apprise_url_tests).run_all()
|
||||
|
||||
|
||||
@mock.patch('requests.post')
|
||||
def test_plugin_custom_lunasea_edge_cases(mock_post):
|
||||
"""
|
||||
NotifyLunaSea() Edge Cases
|
||||
|
||||
"""
|
||||
|
||||
# Prepare our response
|
||||
response = requests.Request()
|
||||
response.status_code = requests.codes.ok
|
||||
response.content = ''
|
||||
|
||||
# Prepare Mock
|
||||
mock_post.return_value = response
|
||||
|
||||
# Prepare a URL with some garbage in it that gets parsed out anyway
|
||||
# key take away is we provided userA and device1
|
||||
results = NotifyLunaSea.parse_url('lsea://user:pass@@userA,+device1,~~,,')
|
||||
|
||||
assert isinstance(results, dict)
|
||||
assert results['user'] == 'user'
|
||||
assert results['password'] == 'pass'
|
||||
assert results['port'] is None
|
||||
assert results['host'] == 'userA,+device1,~~,,'
|
||||
assert results['fullpath'] is None
|
||||
assert results['path'] is None
|
||||
assert results['query'] is None
|
||||
assert results['schema'] == 'lsea'
|
||||
assert results['url'] == 'lsea://user:pass@userA,+device1,~~,,'
|
||||
assert isinstance(results['qsd:'], dict) is True
|
||||
|
||||
instance = NotifyLunaSea(**results)
|
||||
assert isinstance(instance, NotifyLunaSea)
|
||||
assert len(instance.targets) == 2
|
||||
assert ('@', 'userA') in instance.targets
|
||||
assert ('+', 'device1') in instance.targets
|
||||
|
||||
assert instance.notify("test") is True
|
||||
|
||||
# 1 call to user, and second to device
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
url = mock_post.call_args_list[0][0][0]
|
||||
assert url == 'https://notify.lunasea.app/v1/custom/device/device1'
|
||||
payload = loads(mock_post.call_args_list[0][1]['data'])
|
||||
assert 'title' in payload
|
||||
assert 'body' in payload
|
||||
assert 'image' not in payload
|
||||
assert payload['body'] == 'test'
|
||||
assert payload['title'] == 'Apprise Notifications'
|
||||
|
||||
url = mock_post.call_args_list[1][0][0]
|
||||
assert url == 'https://notify.lunasea.app/v1/custom/user/userA'
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
assert 'title' in payload
|
||||
assert 'body' in payload
|
||||
assert 'image' not in payload
|
||||
assert payload['body'] == 'test'
|
||||
assert payload['title'] == 'Apprise Notifications'
|
||||
|
||||
assert '@userA' in instance.url()
|
||||
assert '+device1' in instance.url()
|
||||
|
||||
# Test using a locally hosted instance now:
|
||||
mock_post.reset_mock()
|
||||
|
||||
results = NotifyLunaSea.parse_url(
|
||||
'lseas://user:pass@myhost:3222/@userA,+device1,~~,,')
|
||||
|
||||
assert isinstance(results, dict)
|
||||
assert results['user'] == 'user'
|
||||
assert results['password'] == 'pass'
|
||||
assert results['port'] == 3222
|
||||
assert results['host'] == 'myhost'
|
||||
assert (
|
||||
results['fullpath'] == '/%40userA%2C%2Bdevice1%2C~~%2C%2C' or
|
||||
# Compatible with RHEL8 (Python v3.6.8)
|
||||
results['fullpath'] == '/%40userA%2C%2Bdevice1%2C%7E%7E%2C%2C'
|
||||
)
|
||||
assert results['path'] == '/'
|
||||
assert (
|
||||
results['query'] == '%40userA%2C%2Bdevice1%2C~~%2C%2C' or
|
||||
# Compatible with RHEL8 (Python v3.6.8)
|
||||
results['query'] == '%40userA%2C%2Bdevice1%2C%7E%7E%2C%2C'
|
||||
)
|
||||
assert results['schema'] == 'lseas'
|
||||
assert (
|
||||
results['url'] ==
|
||||
'lseas://user:pass@myhost:3222/%40userA%2C%2Bdevice1%2C~~%2C%2C' or
|
||||
# Compatible with RHEL8 (Python v3.6.8)
|
||||
results['url'] ==
|
||||
'lseas://user:pass@myhost:3222/%40userA%2C%2Bdevice1%2C%7E%7E%2C%2C'
|
||||
)
|
||||
assert isinstance(results['qsd:'], dict) is True
|
||||
|
||||
instance = NotifyLunaSea(**results)
|
||||
assert isinstance(instance, NotifyLunaSea)
|
||||
assert len(instance.targets) == 2
|
||||
assert ('@', 'userA') in instance.targets
|
||||
assert ('+', 'device1') in instance.targets
|
||||
|
||||
assert instance.notify("test") is True
|
||||
|
||||
# 1 call to user, and second to device
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
url = mock_post.call_args_list[0][0][0]
|
||||
assert url == 'https://myhost:3222/v1/custom/device/device1'
|
||||
payload = loads(mock_post.call_args_list[0][1]['data'])
|
||||
assert 'title' in payload
|
||||
assert 'body' in payload
|
||||
assert 'image' not in payload
|
||||
assert payload['body'] == 'test'
|
||||
assert payload['title'] == 'Apprise Notifications'
|
||||
|
||||
url = mock_post.call_args_list[1][0][0]
|
||||
assert url == 'https://myhost:3222/v1/custom/user/userA'
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
assert 'title' in payload
|
||||
assert 'body' in payload
|
||||
assert 'image' not in payload
|
||||
assert payload['body'] == 'test'
|
||||
assert payload['title'] == 'Apprise Notifications'
|
||||
|
||||
assert '@userA' in instance.url()
|
||||
assert '+device1' in instance.url()
|
||||
Loading…
Reference in New Issue
Block a user