diff --git a/KEYWORDS b/KEYWORDS index b23a1981..5ffb7af6 100644 --- a/KEYWORDS +++ b/KEYWORDS @@ -36,6 +36,7 @@ KODI Kumulos LaMetric Line +LunaSea MacOSX Mailgun Mastodon diff --git a/README.md b/README.md index 33573deb..5fc5a633 100644 --- a/README.md +++ b/README.md @@ -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
lametric://apikey@hostname:port
lametric://client_id@client_secret | [Line](https://github.com/caronc/apprise/wiki/Notify_line) | line:// | (TCP) 443 | line://Token@User
line://Token/User1/User2/UserN +| [LunaSea](https://github.com/caronc/apprise/wiki/Notify_lunasea) | lunasea:// | (TCP) 80 or 443 | lunasea://user:pass@+FireBaseDevice/
lunasea://user:pass@FireBaseUser/
lunasea://user:pass@hostname/+FireBaseDevice/
lunasea://user:pass@hostname/@FireBaseUser/ | [Mailgun](https://github.com/caronc/apprise/wiki/Notify_mailgun) | mailgun:// | (TCP) 443 | mailgun://user@hostname/apikey
mailgun://user@hostname/apikey/email
mailgun://user@hostname/apikey/email1/email2/emailN
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
mastodon://access_key@hostname/@user
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
matrix://user@hostname
matrixs://user:pass@hostname:port/#room_alias
matrixs://user:pass@hostname:port/!room_id
matrixs://user:pass@hostname:port/#room_alias/!room_id/#room2
matrixs://token@hostname:port/?webhook=matrix
matrix://user:token@hostname/?webhook=slack&format=markdown diff --git a/apprise/plugins/NotifyLunaSea.py b/apprise/plugins/NotifyLunaSea.py new file mode 100644 index 00000000..51d82091 --- /dev/null +++ b/apprise/plugins/NotifyLunaSea.py @@ -0,0 +1,440 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2024, Chris Caron +# +# 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 diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 87611164..5598af03 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -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, diff --git a/test/test_plugin_lunasea.py b/test/test_plugin_lunasea.py new file mode 100644 index 00000000..3b43cd84 --- /dev/null +++ b/test/test_plugin_lunasea.py @@ -0,0 +1,280 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2024, Chris Caron +# +# 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()