NotificationAPI Support - wip

apprise/plugins/notificationapi.py

@@ -0,0 +1,790 @@
+# BSD 2-Clause License
+#
+# Apprise - Push Notification Library.
+# Copyright (c) 2025, 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.
+
+# Simple API Reference:
+#  - https://www.notificationapi.com/docs/reference/server#send
+
+from __future__ import annotations
+
+import base64
+from email.utils import formataddr
+from itertools import chain
+from json import dumps, loads
+import re
+
+import requests
+
+from ..common import NotifyImageSize, NotifyType
+from ..locale import gettext_lazy as _
+from ..utils.parse import (
+    is_email,
+    is_phone_no,
+    parse_emails,
+    parse_list,
+    validate_regex,
+)
+from .base import NotifyBase
+
+# Valid targets:
+# id:sms
+# id:email
+# id
+IS_VALID_TARGET_RE = re.compile(
+    r"^\s*(((?P<id>[\w_-]+)\s*:)?(?P<target>.+)|(?P<id2>[\w_-]+))", re.I)
+
+
+class NotificationAPIRegion:
+    """Regions."""
+
+    CA = "ca"
+    US = "us"
+    EU = "eu"
+
+
+# NotificationAPI endpoints
+NOTIFICATIONAPI_API_LOOKUP = {
+    NotificationAPIRegion.US: "https://api.notificationapi.com",
+    NotificationAPIRegion.CA: "https://api.ca.notificationapi.com",
+    NotificationAPIRegion.EU: "https://api.eu.notificationapi.com",
+}
+
+# A List of our regions we can use for verification
+NOTIFICATIONAPI_REGIONS = (
+    NotificationAPIRegion.US,
+    NotificationAPIRegion.CA,
+    NotificationAPIRegion.EU,
+)
+
+
+class NotificationAPIChannel:
+    """Channels"""
+
+    AUTO = "AUTO"
+    EMAIL = "EMAIL"
+    SMS = "SMS"
+
+
+# A List of our channels we can use for verification
+NOTIFICATIONAPI_CHANNELS: frozenset[str] = frozenset([
+    NotificationAPIChannel.AUTO,
+    NotificationAPIChannel.EMAIL,
+    NotificationAPIChannel.SMS,
+])
+
+
+class NotifyNotificationAPI(NotifyBase):
+    """
+    A wrapper for NotificationAPI Notifications
+    """
+
+    # The default descriptive name associated with the Notification
+    service_name = "NotificationAPI"
+
+    # The services URL
+    service_url = "https://www.notificationapi.com/"
+
+    # The default secure protocol
+    secure_protocol = ("napi", "notificationapi")
+
+    # A URL that takes you to the setup/help of the specific protocol
+    setup_url = "https://github.com/caronc/apprise/wiki/Notify_notificationapi"
+
+    # If no NotificationAPI Message Type is specified, then the following is
+    # used
+    default_message_type = "apprise"
+
+    # Allow 300 requests per minute.
+    # 60/300 = 0.2
+    request_rate_per_sec = 0.2
+
+    # Allows the user to specify the NotifyImageSize object
+    image_size = NotifyImageSize.XY_72
+
+    # Define object templates
+    templates = (
+        "{schema}://{client_id}/{client_secret}/{targets}",
+        "{schema}://{type}@{client_id}/{client_secret}/{targets}",
+    )
+
+    # Explicit URL tokens we care about (all others from base are ignored)
+    template_tokens = dict(NotifyBase.template_tokens, **{
+        "type": {
+            "name": _("Message Type"),
+            "type": "string",
+            "regex": (r"^[A-Z0-9_-]+$", "i"),
+            "required": True,
+            "map_to": "message_type",
+        },
+        "client_id": {
+            "name": _("Client ID"),
+            "type": "string",
+            "required": True,
+        },
+        "client_secret": {
+            "name": _("Client Secret"),
+            "type": "string",
+            "required": True,
+            "private": True,
+        },
+        "target_email": {
+            "name": _("Target Email"),
+            "type": "string",
+            "map_to": "targets",
+        },
+        "target_id": {
+            "name": _("Target ID"),
+            "type": "string",
+            "map_to": "targets",
+        },
+        "target_sms": {
+            "name": _("Target SMS"),
+            "type": "string",
+            "map_to": "targets",
+        },
+        "targets": {
+            "name": _("Targets"),
+            "type": "list:string",
+        },
+    })
+
+    # Supported query args
+    template_args = dict(NotifyBase.template_args, **{
+        "type": {
+            "alias_of": "type",
+        },
+        "channel": {
+            "name": _("Channel"),
+            "type": "choice:string",
+            "values": NOTIFICATIONAPI_CHANNELS,
+            "default": NotificationAPIChannel.AUTO,
+        },
+        "region": {
+            "name": _("Region Name"),
+            "type": "choice:string",
+            "values": NOTIFICATIONAPI_REGIONS,
+            "default": NotificationAPIRegion.US,
+        },
+        "base_url": {
+            "name": _("Base URL Override"),
+            "type": "string",
+        },
+        "to": {
+            "alias_of": "targets",
+        },
+        # Email Values
+        "cc": {
+            "name": _("Carbon Copy"),
+            "type": "list:string",
+        },
+        "bcc": {
+            "name": _("Blind Carbon Copy"),
+            "type": "list:string",
+        },
+        "from": {
+            "name": _("From Email"),
+            "type": "string",
+            "map_to": "from_addr",
+        },
+        "id": {
+            "alias_of": "client_id",
+        },
+        "secret": {
+            "alias_of": "client_secret",
+        }
+    })
+
+    # Define our token control
+    template_kwargs = {
+        "tokens": {
+            "name": _("Template Tokens"),
+            "prefix": ":",
+        },
+    }
+
+    def __init__(self, client_id, client_secret, message_type=None,
+                 targets=None, cc=None, bcc=None, channel=None, region=None,
+                 from_addr=None, tokens=None, **kwargs):
+        """
+        Initialize Notify NotificationAPI Object
+        """
+        super().__init__(**kwargs)
+
+        # Client ID
+        self.client_id = validate_regex(client_id)
+        if not self.client_id:
+            msg = "An invalid NotificationAPI Client ID " \
+                  "({}) was specified.".format(client_id)
+            self.logger.warning(msg)
+            raise TypeError(msg)
+
+        # Client Secret
+        self.client_secret = validate_regex(client_secret)
+        if not self.client_secret:
+            msg = "An invalid NotificationAPI Client Secret " \
+                  "({}) was specified.".format(client_secret)
+            self.logger.warning(msg)
+            raise TypeError(msg)
+
+        # For tracking our email -> name lookups
+        self.names = {}
+
+        # Temporary from_addr to work with for parsing
+        _from_addr = [self.app_id, ""]
+        self.from_addr = None
+        if isinstance(from_addr, str):
+            result = is_email(from_addr)
+            if result:
+                _from_addr = (
+                    result["name"] if result["name"] else _from_addr[0],
+                    result["full_email"])
+            else:
+                # Only update the string but use the already detected info
+                _from_addr[0] = from_addr
+
+                # Store our lookup
+            self.from_addr = _from_addr[1]
+        self.names[_from_addr[1]] = _from_addr[0]
+
+        # Our Targets; must be formatted as
+        #  id:email or id:phone_no
+        self.email_targets = []
+        self.sms_targets = []
+        self.id_targets = []
+
+        if not message_type:
+            # Assign a default message type
+            self.message_type = self.default_message_type
+
+        else:
+            # Validate information
+            self.message_type = validate_regex(
+                message_type, *self.template_tokens["type"]["regex"])
+            if not self.message_type:
+                msg = "An invalid NotificationAPI Message Type " \
+                      "({}) was specified.".format(message_type)
+                self.logger.warning(msg)
+                raise TypeError(msg)
+
+        # Precompute auth header
+        # Ruby/docs show POST "/{client_id}/sender" with:
+        #      Basic base64(client_id:client_secret)
+        # https://www.notificationapi.com/docs/reference/server
+        token = base64.b64encode(
+            f"{self.client_id}:{self.client_secret}".encode()).decode("ascii")
+        self.auth_header = f"Basic {token}"
+
+        # Acquire Carbon Copies
+        self.cc = set()
+
+        # Acquire Blind Carbon Copies
+        self.bcc = set()
+
+        # Store our region
+        try:
+            self.region = (
+                self.template_args["region"]["default"]
+                if region is None else region.lower()
+            )
+
+            if self.region not in NOTIFICATIONAPI_REGIONS:
+                # allow the outer except to handle this common response
+                raise IndexError()
+
+        except (AttributeError, IndexError, TypeError):
+            # Invalid region specified
+            msg = \
+                f"The NotificationAPI region specified ({region}) is invalid."
+            self.logger.warning(msg)
+            raise TypeError(msg) from None
+
+        if targets:
+            # Validate recipients (to:) and drop bad ones:
+            for entry in parse_list(targets):
+                result = IS_VALID_TARGET_RE.match(entry)
+                if not result:
+                    msg = (
+                        "The NotificationAPI target specified "
+                        f"({entry}) is invalid.")
+                    self.logger.warning(msg)
+                    continue
+
+                if result.group("id2"):
+                    self.id_targets.append({
+                        "id": result.group("id2"),
+                    })
+                    continue
+
+                # Store our content
+                uid, recipient = result.group("id"), result.group("target")
+
+                result = is_email(recipient)
+                if result:
+                    self.email_targets.append({
+                        "id": uid,
+                        "email": result["full_email"],
+                    })
+                    continue
+
+                result = is_phone_no(recipient)
+                if result:
+                    self.sms_targets.append({
+                        "id": uid,
+                        "number": result["full"],
+                    })
+                    continue
+
+                msg = (
+                    "The NotificationAPI target specified "
+                    f"({entry}) is invalid.")
+                self.logger.warning(msg)
+
+        if channel is None:
+            self.channel = NotificationAPIChannel.AUTO
+
+        else:
+            # Store our channel
+            try:
+                self.channel = (
+                    self.template_args["channel"]["default"]
+                    if channel is None else channel.upper()
+                )
+
+                if self.channel not in NOTIFICATIONAPI_CHANNELS:
+                    # allow the outer except to handle this common response
+                    raise IndexError()
+
+            except (AttributeError, IndexError, TypeError):
+                # Invalid channel specified
+                msg = (
+                    "The NotificationAPI channel specified "
+                    f"({channel}) is invalid.")
+                self.logger.warning(msg)
+                raise TypeError(msg) from None
+
+        # Validate recipients (cc:) and drop bad ones:
+        for recipient in parse_emails(cc):
+            result = is_email(recipient)
+            if result:
+                self.cc.add(result["full_email"])
+                if result["name"]:
+                    self.names[result["full_email"]] = result["name"]
+                continue
+
+            self.logger.warning(
+                "Dropped invalid Carbon Copy email "
+                "({}) specified.".format(recipient),
+            )
+
+        # Validate recipients (bcc:) and drop bad ones:
+        for recipient in parse_emails(bcc):
+
+            result = is_email(recipient)
+            if result:
+                self.bcc.add(result["full_email"])
+                if result["name"]:
+                    self.names[result["full_email"]] = result["name"]
+                continue
+
+            self.logger.warning(
+                "Dropped invalid Blind Carbon Copy email "
+                "({}) specified.".format(recipient),
+            )
+
+        # Template functionality
+        self.tokens = {}
+        if isinstance(tokens, dict):
+            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
+
+    @property
+    def url_identifier(self):
+        """
+        Returns all of the identifiers that make this URL unique from
+        another simliar one. Targets or end points should never be identified
+        here.
+        """
+        return (self.secure_protocol[0], self.client_id, self.client_secret)
+
+    def url(self, privacy=False, *args, **kwargs):
+        """
+        Returns the URL built dynamically based on specified arguments.
+        """
+
+        # Our URL parameters
+        params = self.url_parameters(privacy=privacy, *args, **kwargs)
+
+        if len(self.cc) > 0:
+            # Handle our Carbon Copy Addresses
+            params["cc"] = ",".join([
+                formataddr(
+                    (self.names.get(e, False), e),
+                    # Swap comma for it's escaped url code (if detected) since
+                    # we're using that as a delimiter
+                    charset="utf-8").replace(",", "%2C")
+                for e in self.cc])
+
+        if len(self.bcc) > 0:
+            # Handle our Blind Carbon Copy Addresses
+            params["bcc"] = ",".join([
+                formataddr(
+                    (self.names.get(e, False), e),
+                    # Swap comma for it's escaped url code (if detected) since
+                    # we're using that as a delimiter
+                    charset="utf-8").replace(",", "%2C")
+                for e in self.bcc])
+
+        if self.channel != self.template_args["channel"]["default"]:
+            # Prepare our default channel
+            params["channel"] = self.channel.lower()
+
+        if self.region != self.template_args["region"]["default"]:
+            # Prepare our default region
+            params["region"] = self.region
+
+        # handle from=
+        if self.from_addr and self.names[self.from_addr] != self.app_id:
+            params["from"] = self.names[self.from_addr]
+
+        # Extend our parameters
+        params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
+        # Store any template entries if specified
+        params.update({f":{k}": v for k, v in self.tokens.items()})
+
+        return "{schema}://{mtype}@{cid}/{secret}/{targets}?{params}".format(
+            schema=self.secure_protocol[0],
+            mtype=self.message_type,
+            cid=self.pprint(self.client_id, privacy, safe=""),
+            secret=self.pprint(self.client_secret, privacy, safe=""),
+            targets="" if not (
+                self.sms_targets or self.email_targets) else "/".join(
+                chain(
+                    [
+                        NotifyNotificationAPI.quote(
+                            "{}:{}".format(x["id"], x["email"]), safe="")
+                        for x in self.email_targets],
+                    [
+                        NotifyNotificationAPI.quote(
+                            "{}:{}".format(
+                                x["id"], x["number"]), safe="")
+                        for x in self.sms_targets],
+                    [
+                        NotifyNotificationAPI.quote(x, safe="")
+                        for x in self.id_targets],
+                    )),
+            params=NotifyNotificationAPI.urlencode(params),
+        )
+
+    def __len__(self):
+        """
+        Returns the number of targets associated with this notification
+        """
+
+        return max(
+            1,
+            len(self.email_targets) + len(self.sms_targets)
+            + len(self.id_targets))
+
+    def gen_payload(self, body, title="", notify_type=NotifyType.INFO,
+                    **kwargs):
+        """
+        generates our NotificationAPI payload
+        """
+
+        # Take a copy of our token dictionary
+        parameters = self.tokens.copy()
+
+        # Apply some defaults template values
+        parameters["appBody"] = body
+        parameters["appTitle"] = title
+        parameters["appType"] = notify_type.value
+        parameters["appId"] = self.app_id
+        parameters["appDescription"] = self.app_desc
+        parameters["appColor"] = self.color(notify_type)
+        parameters["appImageUrl"] = self.image_url(notify_type)
+        parameters["appUrl"] = self.app_url
+
+        # A Simple Email Payload Template
+        _payload = {
+            "type": self.message_type,
+            "parameters": {**parameters},
+        }
+
+        if self.channel != NotificationAPIChannel.AUTO:
+            _payload["forceChannels"] = [self.channel]
+
+        # Copy our list to work with
+        targets = list(self.email_targets) + \
+            list(self.sms_targets) + list(self.id_targets)
+
+        if self.from_addr:
+            _payload.update({
+                "options": {
+                    "email": {
+                        "fromAddress": self.names[self.from_addr],
+                        "fromName": self.from_addr}}})
+
+        elif self.cc or self.bcc:
+            # Set up shell
+            _payload.update({"options": {"email": {}}})
+
+        while len(targets) > 0:
+            target = targets.pop(0)
+
+            # Create a copy of our template
+            payload = _payload.copy()
+
+            # the cc, bcc, to field must be unique or SendMail will fail,
+            # the below code prepares this by ensuring the target isn't in
+            # the cc list or bcc list. It also makes sure the cc list does
+            # not contain any of the bcc entries
+            if "email" in target:
+                cc = (self.cc - self.bcc - {target["email"]})
+                bcc = (self.bcc - {target["email"]})
+
+            else:
+                # Assume defaults
+                cc = self.cc
+                bcc = self.bcc
+
+            #
+            # Prepare our 'to'
+            #
+            payload["to"] = {**target}
+
+            # Support cc/bcc
+            if len(cc):
+                payload["options"]["email"]["ccAddresses"] = list(cc)
+            if len(bcc):
+                payload["options"]["email"]["bccAddresses"] = list(bcc)
+
+            yield payload
+
+    def send(self, body, title="", notify_type=NotifyType.INFO, **kwargs):
+        """
+        Perform NotificationAPI Notification
+        """
+
+        # error tracking (used for function return)
+        has_error = False
+
+        if not (self.email_targets or self.sms_targets):
+            # There is no one to email or send an sms message to; we're done
+            self.logger.warning(
+                "There are no NotificationAPI recipients to notify"
+            )
+            return False
+
+        # Prepare our URL
+        url = (
+            f"{NOTIFICATIONAPI_API_LOOKUP[self.region]}/"
+            f"{self.client_id}/sender")
+
+        headers = {
+            "User-Agent": self.app_id,
+            "Content-Type": "application/json",
+            "Authorization": self.auth_header,
+        }
+
+        for payload in self.gen_payload(
+                body, title="", notify_type=NotifyType.INFO, **kwargs):
+            # Perform our post
+            self.logger.debug(
+                "NotifiationAPI POST URL: {} (cert_verify={!r})".format(
+                    url, self.verify_certificate))
+            self.logger.debug(
+                "NotifiationAPI Payload: %s", str(payload["to"]["id"]))
+
+            # TODO: Prepare our default response
+            # response = {}
+
+            # Always call throttle before any remote server i/o is made
+            self.throttle()
+            try:
+                r = requests.post(
+                    url,
+                    data=dumps(payload),
+                    headers=headers,
+                    verify=self.verify_certificate,
+                    timeout=self.request_timeout,
+                )
+
+                try:
+                    # TODO: Use response variable to assist in status
+                    # handling
+                    # response = loads(r.content)
+                    loads(r.content)
+
+                except (AttributeError, TypeError, ValueError):
+                    # This gets thrown if we can't parse our JSON Response
+                    #  - ValueError = r.content is Unparsable
+                    #  - TypeError = r.content is None
+                    #  - AttributeError = r is None
+                    self.logger.warning(
+                        "Invalid response from NotifiationAPI server.")
+                    self.logger.debug(
+                        "Response Details:\r\n{}".format(r.content))
+
+                    # Record our failure
+                    has_error = True
+                    continue
+
+                # Reference status code
+                status_code = r.status_code
+
+                if status_code not in (
+                        requests.codes.ok, requests.codes.accepted):
+                    # We had a problem
+                    status_str = \
+                        NotifyNotificationAPI.http_response_code_lookup(
+                            status_code)
+
+                    self.logger.warning(
+                        "Failed to send NotifiationAPI notification to %s: "
+                        "%s%serror=%d",
+                        payload["to"]["id"],
+                        status_str,
+                        ", " if status_str else "",
+                        status_code)
+                    self.logger.debug(
+                        "Response Details:\r\n%s", str(r.content))
+
+                    # Record our failure
+                    has_error = True
+
+                else:
+                    self.logger.info(
+                        "Sent NotifiationAPI notification to %s.",
+                        payload["to"]["id"])
+
+
+            except requests.RequestException as e:
+                self.logger.warning(
+                    "A Connection error occurred sending NotifiationAPI "
+                    "notification to %s.", payload["to"]["id"])
+                self.logger.debug("Socket Exception: {}".format(str(e)))
+
+                # Record our failure
+                has_error = True
+
+        return not has_error
+
+    @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
+
+        # Define our minimum requirements; defining them now saves us from
+        # having to if/else all kinds of branches below...
+        results["client_id"] = None
+        results["client_secret"] = None
+
+        # Prepare our targets (starting with our host)
+        results["targets"] = [
+            NotifyNotificationAPI.unquote(results["host"])]
+
+        # For tracking email sources
+        results["from_addr"] = None
+        if "from" in results["qsd"] and len(results["qsd"]["from"]):
+            results["from_addr"] = \
+                NotifyNotificationAPI.unquote(results["qsd"]["from"].rstrip())
+
+        # First 2 elements are the client_id and client_secret
+        # Following are targets
+        results["targets"] += \
+            NotifyNotificationAPI.split_path(results["fullpath"])
+        # check for our client id
+        if "id" in results["qsd"] and len(results["qsd"]["id"]):
+            # Store our Client ID
+            results["client_id"] = \
+                NotifyNotificationAPI.unquote(results["qsd"]["id"])
+
+        elif results["targets"]:
+            # Store our Client ID
+            results["client_id"] = results["targets"].pop(0)
+
+        if "secret" in results["qsd"] and len(results["qsd"]["secret"]):
+            # Store our Client Secret
+            results["client_secret"] = \
+                NotifyNotificationAPI.unquote(results["qsd"]["secret"])
+
+        elif results["targets"]:
+            # Store our Client Secret
+            results["client_secret"] = results["targets"].pop(0)
+
+        if "region" in results["qsd"] and len(results["qsd"]["region"]):
+            results["region"] = \
+                NotifyNotificationAPI.unquote(results["qsd"]["region"])
+
+        if "channel" in results["qsd"] and len(results["qsd"]["channel"]):
+            results["channel"] = \
+                NotifyNotificationAPI.unquote(results["qsd"]["channel"])
+
+        if "type" in results["qsd"] and len(results["qsd"]["type"]):
+            results["message_type"] = \
+                NotifyNotificationAPI.unquote(results["qsd"]["type"])
+
+        elif results["user"]:
+            # Pull from user
+            results["message_type"] = \
+                NotifyNotificationAPI.unquote(results["user"])
+
+        # The 'to' makes it easier to use yaml configuration
+        if "to" in results["qsd"] and len(results["qsd"]["to"]):
+            results["targets"].append(
+                NotifyNotificationAPI.unquote(results["qsd"]["to"]))
+
+        # Handle Carbon Copy Addresses
+        if "cc" in results["qsd"] and len(results["qsd"]["cc"]):
+            results["cc"] = \
+                NotifyNotificationAPI.unquote(results["qsd"]["cc"])
+
+        # Handle Blind Carbon Copy Addresses
+        if "bcc" in results["qsd"] and len(results["qsd"]["bcc"]):
+            results["bcc"] = \
+                NotifyNotificationAPI.unquote(results["qsd"]["bcc"])
+
+        # Store our tokens
+        results["tokens"] = results["qsd:"]
+
+        return results

tests/helpers/rest.py

@@ -449,7 +449,8 @@ class AppriseURLTester:
                     body=self.body, title=self.title, notify_type=notify_type
                 )
                 if _resp != notify_response:
-                    raise AssertionError()
+                    raise AssertionError(
+                        f"notify() call; notify_response={_resp} on {url}")
 
                 if notify_response:
                     # If we successfully got a response, there must have been

tests/test_plugin_notificationapi.py

@@ -0,0 +1,223 @@
+# BSD 2-Clause License
+#
+# Apprise - Push Notification Library.
+# Copyright (c) 2025, 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.
+
+from json import dumps
+
+# Disable logging for a cleaner testing output
+import logging
+import os
+
+from helpers import AppriseURLTester
+import requests
+
+from apprise.plugins.notificationapi import NotifyNotificationAPI
+
+logging.disable(logging.CRITICAL)
+
+NOTIFICATIONAPI_GOOD_RESPONSE = dumps({})
+
+NOTIFICATIONAPI_BAD_RESPONSE = "{"
+
+# Attachment Directory
+TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), "var")
+
+# Our Testing URLs
+apprise_url_tests = (
+    ("napi://", {
+        "instance": TypeError,
+    }),
+    ("napi://:@/", {
+        "instance": TypeError,
+    }),
+    ("napi://abcd", {
+        # invalid from email
+        "instance": TypeError,
+    }),
+    ("napi://abcd@host.com", {
+        # Just an Email specified, no client_id or client_secret
+        "instance": TypeError,
+    }),
+    ("napi://type@client_id/client_secret/id:+15551235555/", {
+        "instance": NotifyNotificationAPI,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+    }),
+    ("napi://type@cid/secret/id:user@example.com/", {
+        "instance": NotifyNotificationAPI,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+    }),
+    (("napi://type@cid/secret/id:user@example.com/"
+      "id:+15551235555/id:+15551235534"), {
+        "instance": NotifyNotificationAPI,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+    }),
+    (("napi://type@cid/secret/id:user@example.com/"
+      "?from=joe@example.ca"), {
+        # Set from/source
+        "instance": NotifyNotificationAPI,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+    }),
+    (("napi://type@cid/secret/id:user@example.com/"
+      "?from=joe@example.ca&bcc=user1@yahoo.ca&cc=user2@yahoo.ca"), {
+        # Set from/source
+        "instance": NotifyNotificationAPI,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+    }),
+    ("napi://?id=ci&secret=cs&to=id:user@example.com&type=type", {
+        # use just kwargs
+        "instance": NotifyNotificationAPI,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+    }),
+    ("napi://?id=ci&secret=cs&type=test-type", {
+        # No targets specified
+        "instance": NotifyNotificationAPI,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+        "notify_response": False,
+    }),
+    ("napi://user@client_id/cs2/id:user@example.ca"
+     "?bcc=invalid", {
+         # A good email with a bad Blind Carbon Copy
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs3/id:user@example.ca"
+     "?cc=l2g@nuxref.com", {
+         # A good email with Carbon Copy
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs4/id:user@example.ca"
+     "?cc=Chris<l2g@nuxref.com>", {
+         # A good email with Carbon Copy
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs5/id:user@example.ca"
+     "?cc=invalid", {
+         # A good email with Carbon Copy
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs6/id:user@example.ca"
+     "?to=invalid", {
+         # an invalid to email
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+     ("napi://user@client_id/cs7/id:chris@example.com", {
+        # An email with a designated to email
+        "instance": NotifyNotificationAPI,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+    }),
+    ("napi://user@client_id/cs8/id:user@example.ca"
+     "?to=id:Chris<chris@example.com>", {
+         # An email with a full name in in To field
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs9/id:user@example.ca"
+     "id:chris@example.com/id:chris2@example.com/id:+15552341234", {
+         # Several emails to notify
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs10/id:user@example.ca"
+     "?cc=Chris<chris@example.com>", {
+         # An email with a full name in cc
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs11/id:user@example.ca"
+     "?cc=chris@example.com", {
+         # An email with a full name in cc
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs12/id:user@example.ca"
+     "?bcc=Chris<chris@example.com>", {
+         # An email with a full name in bcc
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs13/id:user@example.ca"
+     "?bcc=chris@example.com", {
+         # An email with a full name in bcc
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs14/id:user@example.ca"
+     "?to=Chris<chris@example.com>", {
+         # An email with a full name in bcc
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs15/id:user@example.ca"
+     "?to=chris@example.com", {
+         # An email with a full name in bcc
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+     }),
+    ("napi://user@client_id/cs16/id:user@example.ca"
+     "?template=1234&+sub=value&+sub2=value2", {
+         # A good email with a template + substitutions
+         "instance": NotifyNotificationAPI,
+         "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+
+         # Our expected url(privacy=True) startswith() response:
+         "privacy_url": "napi://user@c...d/c...6/",
+     }),
+    ("napi://user@client_id/cs17/id:user@example.ca", {
+        "instance": NotifyNotificationAPI,
+        # force a failure
+        "response": False,
+        "requests_response_code": requests.codes.internal_server_error,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+    }),
+    ("napi://user@client_id/cs18/id:user@example.ca", {
+        "instance": NotifyNotificationAPI,
+        # throw a bizzare code forcing us to fail to look it up
+        "response": False,
+        "requests_response_code": 999,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+    }),
+    ("napi://user@client_id/cs19/id:user@example.ca", {
+        "instance": NotifyNotificationAPI,
+        # Throws a series of connection and transfer exceptions when this flag
+        # is set and tests that we gracfully handle them
+        "test_requests_exceptions": True,
+        "requests_response_text": NOTIFICATIONAPI_GOOD_RESPONSE,
+    }),
+)
+
+
+def test_plugin_napi_urls():
+    """
+    NotifyNotificationAPI() Apprise URLs
+
+    """
+
+    # Run our general tests
+    AppriseURLTester(tests=apprise_url_tests).run_all()