test coverage

pull/1399/head
Chris Caron 2025-09-06 13:23:56 -04:00
parent e763f3b71e
commit 3064eb06db
7 changed files with 360 additions and 67 deletions

View File

@ -92,6 +92,7 @@ The table below identifies the services this tool supports and some example serv
| [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 | [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
| [NextcloudTalk](https://github.com/caronc/apprise/wiki/Notify_nextcloudtalk) | nctalk:// or nctalks:// | (TCP) 80 or 443 | nctalk://user:pass@host/RoomId<br/>nctalks://user:pass@host/RoomId1/RoomId2/RoomIdN | [NextcloudTalk](https://github.com/caronc/apprise/wiki/Notify_nextcloudtalk) | nctalk:// or nctalks:// | (TCP) 80 or 443 | nctalk://user:pass@host/RoomId<br/>nctalks://user:pass@host/RoomId1/RoomId2/RoomIdN
| [Notica](https://github.com/caronc/apprise/wiki/Notify_notica) | notica:// | (TCP) 443 | notica://Token/ | [Notica](https://github.com/caronc/apprise/wiki/Notify_notica) | notica:// | (TCP) 443 | notica://Token/
| [NotificationAPI](https://github.com/caronc/apprise/wiki/Notify_notificationapi) | napi:// | (TCP) 443 | napi://ClientID/ClientSecret/Target<br />napi://ClientID/ClientSecret/Target1/Target2/TargetN<br />napi://MessageType@ClientID/ClientSecret/Target
| [Notifiarr](https://github.com/caronc/apprise/wiki/Notify_notifiarr) | notifiarr:// | (TCP) 443 | notifiarr://apikey/#channel<br />notifiarr://apikey/#channel1/#channel2/#channeln | [Notifiarr](https://github.com/caronc/apprise/wiki/Notify_notifiarr) | notifiarr:// | (TCP) 443 | notifiarr://apikey/#channel<br />notifiarr://apikey/#channel1/#channel2/#channeln
| [Notifico](https://github.com/caronc/apprise/wiki/Notify_notifico) | notifico:// | (TCP) 443 | notifico://ProjectID/MessageHook/ | [Notifico](https://github.com/caronc/apprise/wiki/Notify_notifico) | notifico:// | (TCP) 443 | notifico://ProjectID/MessageHook/
| [ntfy](https://github.com/caronc/apprise/wiki/Notify_ntfy) | ntfy:// | (TCP) 80 or 443 | ntfy://topic/<br/>ntfys://topic/ | [ntfy](https://github.com/caronc/apprise/wiki/Notify_ntfy) | ntfy:// | (TCP) 80 or 443 | ntfy://topic/<br/>ntfys://topic/

View File

@ -666,7 +666,7 @@ class ConfigBase(URLBase):
valid_line_re = re.compile( valid_line_re = re.compile(
r"^\s*(?P<line>([;#]+(?P<comment>.*))|" r"^\s*(?P<line>([;#]+(?P<comment>.*))|"
r"(\s*(?P<tags>[a-z0-9, \t_-]+)\s*=|=)?\s*" r"(\s*(?P<tags>[a-z0-9, \t_-]+)\s*=|=)?\s*"
r"((?P<url>[a-z0-9]{1,12}://.*)|(?P<assign>[a-z0-9, \t_-]+))|" r"((?P<url>[a-z0-9]{1,32}://.*)|(?P<assign>[a-z0-9, \t_-]+))|"
r"include\s+(?P<config>.+))?\s*$", r"include\s+(?P<config>.+))?\s*$",
re.I, re.I,
) )

View File

@ -32,6 +32,7 @@ from __future__ import annotations
import base64 import base64
from email.utils import formataddr from email.utils import formataddr
from itertools import chain
from json import dumps, loads from json import dumps, loads
import re import re
@ -51,7 +52,7 @@ from .base import NotifyBase
# Used to detect ID # Used to detect ID
IS_VALID_ID_RE = re.compile( IS_VALID_ID_RE = re.compile(
r"^\s*(@|%40)?(?P<id>[\w_-]+)\s*", re.I) r"^\s*(@|%40)?(?P<id>[\w_-]+)\s*$", re.I)
class NotificationAPIRegion: class NotificationAPIRegion:
@ -85,7 +86,7 @@ class NotificationAPIChannel:
INAPP = "inapp" INAPP = "inapp"
WEB_PUSH = "web_push" WEB_PUSH = "web_push"
MOBILE_PUSH = "mobile_push" MOBILE_PUSH = "mobile_push"
SLACK = "mobile_push" SLACK = "slack"
# A List of our channels we can use for verification # A List of our channels we can use for verification
@ -209,10 +210,6 @@ class NotifyNotificationAPI(NotifyBase):
"type": "choice:string", "type": "choice:string",
"values": NOTIFICATIONAPI_MODES, "values": NOTIFICATIONAPI_MODES,
}, },
"base_url": {
"name": _("Base URL Override"),
"type": "string",
},
"to": { "to": {
"alias_of": "targets", "alias_of": "targets",
}, },
@ -328,7 +325,6 @@ class NotifyNotificationAPI(NotifyBase):
self.message_type = self.default_message_type self.message_type = self.default_message_type
else: else:
# Validate information
self.message_type = validate_regex( self.message_type = validate_regex(
message_type, *self.template_tokens["type"]["regex"]) message_type, *self.template_tokens["type"]["regex"])
if not self.message_type: if not self.message_type:
@ -382,6 +378,9 @@ class NotifyNotificationAPI(NotifyBase):
raise TypeError(msg) from None raise TypeError(msg) from None
self.channels.add(channel) self.channels.add(channel)
# Used for URL generation afterwards only
self._invalid_targets = []
if targets: if targets:
current_target = {} current_target = {}
for entry in parse_list(targets, sort=False): for entry in parse_list(targets, sort=False):
@ -417,7 +416,7 @@ class NotifyNotificationAPI(NotifyBase):
if result: if result:
if "number" not in current_target: if "number" not in current_target:
current_target["number"] = \ current_target["number"] = \
('+' if entry[0] == '+' else '') + result["full"] ("+" if entry[0] == "+" else "") + result["full"]
if not self.channels: if not self.channels:
self.channels.add(NotificationAPIChannel.SMS) self.channels.add(NotificationAPIChannel.SMS)
self.logger.info( self.logger.info(
@ -448,17 +447,17 @@ class NotifyNotificationAPI(NotifyBase):
current_target["id"] = result.group("id") current_target["id"] = result.group("id")
continue continue
elif "id" in current_target: # Store id in next target and move on
# Store and move on self.targets.append(current_target)
self.targets.append(current_target) current_target = {
current_target = { "id": result.group("id")
"id": result.group("id") }
} continue
continue
self.logger.warning( self.logger.warning(
"Ignoring invalid NotificationAPI target " "Dropped invalid NotificationAPI target "
f"({entry}) specified") f"({entry}) specified")
self._invalid_targets.append(entry)
continue continue
if "id" in current_target: if "id" in current_target:
@ -509,14 +508,6 @@ class NotifyNotificationAPI(NotifyBase):
if isinstance(tokens, dict): if isinstance(tokens, dict):
self.tokens.update(tokens) self.tokens.update(tokens)
elif tokens:
msg = (
"The specified NotificationAPI Template Tokens "
f"({tokens}) are not identified as a dictionary."
)
self.logger.warning(msg)
raise TypeError(msg)
return return
@property @property
@ -572,7 +563,7 @@ class NotifyNotificationAPI(NotifyBase):
if self.channels: if self.channels:
# Prepare our default channel # Prepare our default channel
params["channels"] = self.channels params["channels"] = ",".join(self.channels)
if self.region != self.template_args["region"]["default"]: if self.region != self.template_args["region"]["default"]:
# Prepare our default region # Prepare our default region
@ -589,8 +580,8 @@ class NotifyNotificationAPI(NotifyBase):
targets = [] targets = []
for target in self.targets: for target in self.targets:
if "id" in target: # ID is always present
targets.append(f"@{target['id']}") targets.append(f"@{target['id']}")
if "number" in target: if "number" in target:
targets.append(f"{target['number']}") targets.append(f"{target['number']}")
if "email" in target: if "email" in target:
@ -603,7 +594,8 @@ class NotifyNotificationAPI(NotifyBase):
mtype=mtype, mtype=mtype,
cid=self.pprint(self.client_id, privacy, safe=""), cid=self.pprint(self.client_id, privacy, safe=""),
secret=self.pprint(self.client_secret, privacy, safe=""), secret=self.pprint(self.client_secret, privacy, safe=""),
targets=NotifyNotificationAPI.quote("/".join(targets), safe="/"), targets=NotifyNotificationAPI.quote("/".join(
chain(targets, self._invalid_targets)), safe="/"),
params=NotifyNotificationAPI.urlencode(params), params=NotifyNotificationAPI.urlencode(params),
) )
@ -649,9 +641,9 @@ class NotifyNotificationAPI(NotifyBase):
if self.notify_format == NotifyFormat.HTML else body if self.notify_format == NotifyFormat.HTML else body
for channel in self.channels: for channel in self.channels:
# Python v3.10 supports `match/case` but since Apprise aims to be # Python v3.10 supports `match/case` but since Apprise aims to
# compatible with Python v3.9+, we must use if/else for the time # be compatible with Python v3.9+, we must use if/else for the
# being # time being
if channel == NotificationAPIChannel.SMS: if channel == NotificationAPIChannel.SMS:
_payload.update({ _payload.update({
NotificationAPIChannel.SMS: { NotificationAPIChannel.SMS: {
@ -703,7 +695,7 @@ class NotifyNotificationAPI(NotifyBase):
}, },
}) })
elif channel == NotificationAPIChannel.SLACK: else: # channel == NotificationAPIChannel.SLACK
_payload.update({ _payload.update({
NotificationAPIChannel.SLACK: { NotificationAPIChannel.SLACK: {
"text": (title + "\n" + text_body) "text": (title + "\n" + text_body)
@ -883,8 +875,10 @@ class NotifyNotificationAPI(NotifyBase):
results["client_secret"] = None results["client_secret"] = None
# Prepare our targets (starting with our host) # Prepare our targets (starting with our host)
results["targets"] = [ results["targets"] = []
NotifyNotificationAPI.unquote(results["host"])] if results["host"]:
results["targets"].append(
NotifyNotificationAPI.unquote(results["host"]))
# For tracking email sources # For tracking email sources
results["from_addr"] = None results["from_addr"] = None
@ -919,13 +913,9 @@ class NotifyNotificationAPI(NotifyBase):
results["region"] = \ results["region"] = \
NotifyNotificationAPI.unquote(results["qsd"]["region"]) NotifyNotificationAPI.unquote(results["qsd"]["region"])
if "channel" in results["qsd"] and len(results["qsd"]["channel"]): if "channels" in results["qsd"] and len(results["qsd"]["channels"]):
results["channel"] = \ results["channels"] = \
NotifyNotificationAPI.unquote(results["qsd"]["channel"]) NotifyNotificationAPI.unquote(results["qsd"]["channels"])
if "type" in results["qsd"] and len(results["qsd"]["type"]):
results["message_type"] = \
NotifyNotificationAPI.unquote(results["qsd"]["type"])
if "mode" in results["qsd"] and len(results["qsd"]["mode"]): if "mode" in results["qsd"] and len(results["qsd"]["mode"]):
results["mode"] = \ results["mode"] = \
@ -935,6 +925,11 @@ class NotifyNotificationAPI(NotifyBase):
results["reply_to"] = \ results["reply_to"] = \
NotifyNotificationAPI.unquote(results["qsd"]["reply"]) NotifyNotificationAPI.unquote(results["qsd"]["reply"])
# Handling of Message Type
if "type" in results["qsd"] and len(results["qsd"]["type"]):
results["message_type"] = \
NotifyNotificationAPI.unquote(results["qsd"]["type"])
elif results["user"]: elif results["user"]:
# Pull from user # Pull from user
results["message_type"] = \ results["message_type"] = \

View File

@ -58,14 +58,14 @@ NOTIFY_CUSTOM_DEL_TOKENS = re.compile(r"^-(?P<key>.*)\s*")
NOTIFY_CUSTOM_COLON_TOKENS = re.compile(r"^:(?P<key>.*)\s*") NOTIFY_CUSTOM_COLON_TOKENS = re.compile(r"^:(?P<key>.*)\s*")
# Used for attempting to acquire the schema if the URL can't be parsed. # Used for attempting to acquire the schema if the URL can't be parsed.
GET_SCHEMA_RE = re.compile(r"\s*(?P<schema>[a-z0-9]{1,12})://.*$", re.I) GET_SCHEMA_RE = re.compile(r"\s*(?P<schema>[a-z0-9]{1,32})://.*$", re.I)
# Used for validating that a provided entry is indeed a schema # Used for validating that a provided entry is indeed a schema
# this is slightly different then the GET_SCHEMA_RE above which # this is slightly different then the GET_SCHEMA_RE above which
# insists the schema is only valid with a :// entry. this one # insists the schema is only valid with a :// entry. this one
# extrapolates the individual entries # extrapolates the individual entries
URL_DETAILS_RE = re.compile( URL_DETAILS_RE = re.compile(
r"\s*(?P<schema>[a-z0-9]{1,12})(://(?P<base>.*))?$", re.I r"\s*(?P<schema>[a-z0-9]{1,32})(://(?P<base>.*))?$", re.I
) )
# Regular expression based and expanded from: # Regular expression based and expanded from:
@ -124,7 +124,7 @@ CALL_SIGN_DETECTION_RE = re.compile(
# Regular expression used to destinguish between multiple URLs # Regular expression used to destinguish between multiple URLs
URL_DETECTION_RE = re.compile( URL_DETECTION_RE = re.compile(
r"([a-z0-9]+?:\/\/.*?)(?=$|[\s,]+[a-z0-9]{1,12}?:\/\/)", re.I r"([a-z0-9]+?:\/\/.*?)(?=$|[\s,]+[a-z0-9]{1,32}?:\/\/)", re.I
) )
EMAIL_DETECTION_RE = re.compile( EMAIL_DETECTION_RE = re.compile(
@ -1100,7 +1100,7 @@ def parse_list(*args, cast=None, allow_whitespace=True, sort=True):
) )
) )
return ( return (
[x for x in filter(bool, list(result))] list(filter(bool, list(result)))
if allow_whitespace if allow_whitespace
else [x.strip() for x in filter(bool, list(result)) if x.strip()] else [x.strip() for x in filter(bool, list(result)) if x.strip()]
) )

View File

@ -64,9 +64,9 @@ notification services. It supports sending alerts to platforms such as: \
`Kumulos`, `LaMetric`, `Lark`, `Line`, `MacOSX`, `Mailgun`, `Mastodon`, \ `Kumulos`, `LaMetric`, `Lark`, `Line`, `MacOSX`, `Mailgun`, `Mastodon`, \
`Mattermost`, `Matrix`, `MessageBird`, `Microsoft Windows`, \ `Mattermost`, `Matrix`, `MessageBird`, `Microsoft Windows`, \
`Microsoft Teams`, `Misskey`, `MQTT`, `MSG91`, `MyAndroid`, `Nexmo`, \ `Microsoft Teams`, `Misskey`, `MQTT`, `MSG91`, `MyAndroid`, `Nexmo`, \
`Nextcloud`, `NextcloudTalk`, `Notica`, `Notifiarr`, `Notifico`, `ntfy`, \ `Nextcloud`, `NextcloudTalk`, `Notica`, `NotificationAPI`, `Notifiarr`,
`Office365`, `OneSignal`, `Opsgenie`, `PagerDuty`, `PagerTree`, \ `Notifico`, `ntfy`, \ `Office365`, `OneSignal`, `Opsgenie`, `PagerDuty`, \
`ParsePlatform`, `Plivo`, `PopcornNotify`, `Prowl`, `Pushalot`, \ `PagerTree`, `ParsePlatform`, `Plivo`, `PopcornNotify`, `Prowl`, `Pushalot`, \
`PushBullet`, `Pushjet`, `PushMe`, `Pushover`, `Pushplus`, `PushSafer`, \ `PushBullet`, `Pushjet`, `PushMe`, `Pushover`, `Pushplus`, `PushSafer`, \
`Pushy`, `PushDeer`, `QQ Push`, `Revolt`, `Reddit`, `Resend`, `Rocket.Chat`, \ `Pushy`, `PushDeer`, `QQ Push`, `Revolt`, `Reddit`, `Resend`, `Rocket.Chat`, \
`RSyslog`, `SendGrid`, `SendPulse`, `ServerChan`, `Seven`, `SFR`, `Signal`, \ `RSyslog`, `SendGrid`, `SendPulse`, `ServerChan`, `Seven`, `SFR`, `Signal`, \

View File

@ -107,6 +107,7 @@ keywords = [
"Nextcloud", "Nextcloud",
"NextcloudTalk", "NextcloudTalk",
"Notica", "Notica",
"NoticationAPI",
"Notifiarr", "Notifiarr",
"Notifico", "Notifico",
"Ntfy", "Ntfy",

View File

@ -63,25 +63,103 @@ apprise_url_tests = (
# Just an Email specified, no client_id or client_secret # Just an Email specified, no client_id or client_secret
"instance": TypeError, "instance": TypeError,
}), }),
("napi://user@client_id/cs14/user@example.ca", { ("napi://user@client_id/cs14a/user@example.ca", {
# No id matched # No id matched
"instance": TypeError, "instance": TypeError,
}), }),
("napi://type@client_id/client_secret/id/+15551235553/", { ("napi://user@client_id/cs14b/+15551235553", {
# No id matched
"instance": TypeError,
}),
("napi://user@client_id/cs14c/+15551235553/user@example.ca", {
# No id matched
"instance": TypeError,
}),
("napi://type@client_id/client_secret/id/+15551235553/?mode=invalid", {
# Invalid mode
"instance": TypeError,
}),
("napi://type@client_id/client_secret/id/+15551235553/?region=invalid", {
# Invalid region
"instance": TypeError,
}),
((
"napi://type@client_id/client_secret/id/user@example.ca/"
"user2@example.ca"
), {
# to many emails assigned to id (variation 1)
"instance": TypeError,
}),
((
"napi://type@client_id/client_secret/user@example.ca/"
"user2@example.ca"
), {
# to many emails assigned to id (variation 2)
"instance": TypeError,
}),
((
"napi://type@client_id/client_secret/id/+15551235553/"
"+15551235555"
), {
# to many phone no's assigned to id (variation 1)
"instance": TypeError,
}),
((
"napi://type@client_id/client_secret/+15551235553/"
"+15551235555"
), {
# to many phone no's assigned to id (variation 2)
"instance": TypeError,
}),
("napi://type@client_id/client_secret/id/+15551235553/?mode=invalid", {
# Invalid mode
"instance": TypeError,
}),
("napi://client_id/client_secret/id/+15551231234/?type=*(", {
# Invalid type
"instance": TypeError,
}),
("napi://client_id/client_secret/id/+15551231234/?channels=bad", {
# Invalid channel
"instance": TypeError,
}),
("napi://?secret=cs&to=id,user404@example.com&type=typed", {
# No id found
"instance": TypeError,
}),
("napi://client_id/client_secret/id/g@rb@ge/+15551235553/", {
# g@rb@ge enry ignored
"instance": NotifyNotificationAPI, "instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE, "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
}), }),
("napi://type@cid/secret/id/user1@example.com/", { ("napi://cid/secret/id/user1@example.com/?type=apprise-msg", {
"instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
}),
("notificationapi://cid/secret/id/user1@example.com", {
# Support full schema:// of notificationapi://
"instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
}),
("napi://cid/secret/id/id2/user1@example.com", {
# two id's in a row
"instance": NotifyNotificationAPI, "instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE, "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
}), }),
(("napi://type@cid/secret/id10/user2@example.com/" (("napi://type@cid/secret/id10/user2@example.com/"
"id5/+15551235555/id8/+15551235534"), { "id5/+15551235555/id8/+15551235534"
"?reply=Chris<chris@example.com>"), {
"instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
}),
(("napi://type@cid/secret/abc1/user1@example.com/"
"id5/+15551235555/?from=Chris&reply=Christopher"), {
"instance": NotifyNotificationAPI, "instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE, "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
}), }),
(("napi://type@cid/secret/id/user3@example.com/" (("napi://type@cid/secret/id/user3@example.com/"
"?from=joe@example.ca"), { "?from=joe@example.ca&reply=user@abc.com"), {
# Set from/source # Set from/source
"instance": NotifyNotificationAPI, "instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE, "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
@ -91,18 +169,43 @@ apprise_url_tests = (
# Set from/source # Set from/source
"instance": NotifyNotificationAPI, "instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE, "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
# Our expected url(privacy=True) startswith() response:
"privacy_url": "napi://type@c...d/s...t/",
}), }),
("napi://?id=ci&secret=cs&to=id,user5@example.com&type=type", { ("napi://?id=ci&secret=cs&to=id,user5@example.com&type=typec", {
# use just kwargs # use just kwargs
"instance": NotifyNotificationAPI, "instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE, "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
# Our expected url(privacy=True) startswith() response:
"privacy_url": "napi://typec@c...i/c...s/",
}), }),
("napi://?id=ci&secret=cs&type=test-type", { ("napi://id?secret=cs&to=id,user5@example.com&type=typeb", {
# id is pull from the host
"instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
# Our expected url(privacy=True) startswith() response:
"privacy_url": "napi://typeb@i...d/c...s/",
}),
("napi://secret?id=ci&to=id,user5@example.com&type=typea", {
# id pulled from kwargs still allows secret to be the
# next parsed entry from cli
"instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
# Our expected url(privacy=True) startswith() response:
"privacy_url": "napi://typea@c...i/s...t/",
}),
("napi://?id=ci&secret=cs&type=test-type&region=eu", {
# No targets specified # No targets specified
"instance": NotifyNotificationAPI, "instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE, "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
"notify_response": False, "notify_response": False,
}), }),
("napi://?id=ci&secret=cs&to=id,user5@example.com&type=typec", {
# bad response
"instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_BAD_RESPONSE,
"notify_response": False,
}),
("napi://user@client_id/cs2/id/user6@example.ca" ("napi://user@client_id/cs2/id/user6@example.ca"
"?bcc=invalid", { "?bcc=invalid", {
# A good email with a bad Blind Carbon Copy # A good email with a bad Blind Carbon Copy
@ -115,6 +218,12 @@ apprise_url_tests = (
"instance": NotifyNotificationAPI, "instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE, "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
}), }),
("napi://client_id/cs3/id/user8@example.ca"
"?channels=email,sms,slack,mobile_push,web_push,inapp", {
# A good email with Carbon Copy
"instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
}),
("napi://user@client_id/cs4/id/user9@example.ca" ("napi://user@client_id/cs4/id/user9@example.ca"
"?cc=Chris<l2g@nuxref.com>", { "?cc=Chris<l2g@nuxref.com>", {
# A good email with Carbon Copy # A good email with Carbon Copy
@ -145,7 +254,8 @@ apprise_url_tests = (
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE, "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
}), }),
("napi://user@client_id/cs9/id2/user13@example.ca/" ("napi://user@client_id/cs9/id2/user13@example.ca/"
"id/kris@example.com/id/chris2@example.com/id/+15552341234", { "id/kris@example.com/id/chris2@example.com/id/+15552341234"
"?:token=value", {
# Several emails to notify # Several emails to notify
"instance": NotifyNotificationAPI, "instance": NotifyNotificationAPI,
"requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE, "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
@ -230,8 +340,8 @@ def test_plugin_napi_urls():
@mock.patch("requests.post") @mock.patch("requests.post")
def test_plugin_napi_sms_payloads(mock_post): def test_plugin_napi_template_sms_payloads(mock_post):
"""NotifyNotificationAPI() Testing SMS Payloads.""" """NotifyNotificationAPI() Testing Template SMS Payloads."""
okay_response = requests.Request() okay_response = requests.Request()
okay_response.status_code = requests.codes.ok okay_response.status_code = requests.codes.ok
@ -261,8 +371,7 @@ def test_plugin_napi_sms_payloads(mock_post):
is True is True
) )
# 2 calls were made, one to perform an email lookup, the second # delivery of message
# was the notification itself
assert mock_post.call_count == 1 assert mock_post.call_count == 1
assert ( assert (
mock_post.call_args_list[0][0][0] mock_post.call_args_list[0][0][0]
@ -299,8 +408,8 @@ def test_plugin_napi_sms_payloads(mock_post):
@mock.patch("requests.post") @mock.patch("requests.post")
def test_plugin_napi_email_payloads(mock_post): def test_plugin_napi_template_email_payloads(mock_post):
"""NotifyNotificationAPI() Testing Email Payloads.""" """NotifyNotificationAPI() Testing Template Email Payloads."""
okay_response = requests.Request() okay_response = requests.Request()
okay_response.status_code = requests.codes.ok okay_response.status_code = requests.codes.ok
@ -331,8 +440,7 @@ def test_plugin_napi_email_payloads(mock_post):
is True is True
) )
# 2 calls were made, one to perform an email lookup, the second # delivery of message
# was the notification itself
assert mock_post.call_count == 1 assert mock_post.call_count == 1
assert ( assert (
mock_post.call_args_list[0][0][0] mock_post.call_args_list[0][0][0]
@ -375,3 +483,191 @@ def test_plugin_napi_email_payloads(mock_post):
# Reset our mock object # Reset our mock object
mock_post.reset_mock() mock_post.reset_mock()
@mock.patch("requests.post")
def test_plugin_napi_message_payloads(mock_post):
"""NotifyNotificationAPI() Testing Message Payloads."""
okay_response = requests.Request()
okay_response.status_code = requests.codes.ok
okay_response.content = NOTIFICATIONAPI_GOOD_RESPONSE
# Assign our mock object our return value
mock_post.return_value = okay_response
# Details
client_id = "my_id_abc"
client_secret = "my_secret"
message_type = "apprise-post"
targets = "userid/test@example.ca/+15551239876"
obj = Apprise.instantiate(
f"napi://{message_type}@{client_id}/{client_secret}/"
f"{targets}?from=Chris<chris@example.eu>&bcc=joe@hidden.com"
f"&mode=message")
assert isinstance(obj, NotifyNotificationAPI)
assert isinstance(obj.url(), str)
# No calls made yet
assert mock_post.call_count == 0
# Send our notification
assert (
obj.notify(body="body", title="title", notify_type=NotifyType.INFO)
is True
)
# delivery of message
assert mock_post.call_count == 1
assert (
mock_post.call_args_list[0][0][0]
== f"https://api.notificationapi.com/{client_id}/sender"
)
payload = loads(mock_post.call_args_list[0][1]["data"])
assert payload == {
"type": "apprise-post",
"to": {
"id": "userid",
"email": "test@example.ca",
"number": "+15551239876",
},
"email": {
"subject": "Apprise",
"html": "body",
"senderName": "chris@example.eu",
"senderEmail": "Chris",
},
"options": {
"email": {
"fromAddress": "chris@example.eu",
"fromName": "Chris",
"bccAddresses": ["joe@hidden.com"],
},
},
}
headers = mock_post.call_args_list[0][1]["headers"]
assert headers == {
"User-Agent": "Apprise",
"Content-Type": "application/json",
"Authorization": "Basic bXlfaWRfYWJjOm15X3NlY3JldA=="}
# Reset our mock object
mock_post.reset_mock()
# Reversing the sms with email causes auto-detection channel to
# be sms instead of email
targets = "userid/+15551239876/test@example.ca"
obj = Apprise.instantiate(
f"napi://{client_id}/{client_secret}/"
f"{targets}?from=Chris<chris@example.eu>&bcc=joe@hidden.com")
assert isinstance(obj, NotifyNotificationAPI)
assert isinstance(obj.url(), str)
# No calls made yet
assert mock_post.call_count == 0
# Send our notification
assert (
obj.notify(body="body", title="title", notify_type=NotifyType.INFO)
is True
)
# delivery of message
assert mock_post.call_count == 1
assert (
mock_post.call_args_list[0][0][0]
== f"https://api.notificationapi.com/{client_id}/sender"
)
payload = loads(mock_post.call_args_list[0][1]["data"])
assert payload == {
"type": "apprise",
"to": {
"id": "userid",
"number": "+15551239876",
"email": "test@example.ca",
},
"sms": {"message": "body"},
"options": {
"email": {
"fromAddress": "chris@example.eu",
"fromName": "Chris",
"bccAddresses": ["joe@hidden.com"]},
},
}
headers = mock_post.call_args_list[0][1]["headers"]
assert headers == {
"User-Agent": "Apprise",
"Content-Type": "application/json",
"Authorization": "Basic bXlfaWRfYWJjOm15X3NlY3JldA=="}
# Reset our mock object
mock_post.reset_mock()
# Experiment with fixed channels:
obj = Apprise.instantiate(
f"napi://{message_type}@{client_id}/{client_secret}/"
f"{targets}?from=Chris<chris@example.eu>&bcc=joe@hidden.com"
f"&mode=message&channels=sms,slack")
assert isinstance(obj, NotifyNotificationAPI)
assert isinstance(obj.url(), str)
# No calls made yet
assert mock_post.call_count == 0
# Send our notification
assert (
obj.notify(body="body", title="title", notify_type=NotifyType.INFO)
is True
)
# delivery of message
assert mock_post.call_count == 1
assert (
mock_post.call_args_list[0][0][0]
== f"https://api.notificationapi.com/{client_id}/sender"
)
payload = loads(mock_post.call_args_list[0][1]["data"])
assert payload == {
"type": "apprise-post",
"to": {
"id": "userid",
"email": "test@example.ca",
"number": "+15551239876",
},
"slack": {"text": "body"},
"sms": {"message": "body"},
"options": {
"email": {
"fromAddress": "chris@example.eu",
"fromName": "Chris",
"bccAddresses": ["joe@hidden.com"],
},
},
}
headers = mock_post.call_args_list[0][1]["headers"]
assert headers == {
"User-Agent": "Apprise",
"Content-Type": "application/json",
"Authorization": "Basic bXlfaWRfYWJjOm15X3NlY3JldA=="}
def test_plugin_napi_edge_cases():
"""
NotifyNotificationAPI() Edge Cases
"""
client_id = "my_id_abc"
client_secret = "my_secret"
targets = ["userid", "test@example.ca", "+15551239876"]
# Tests case where tokens is == None
obj = NotifyNotificationAPI(client_id, client_secret, targets=targets)
assert isinstance(obj, NotifyNotificationAPI)
assert isinstance(obj.url(), str)