From 15b366a307b3a51ba83535ab661b67c9785d2fc8 Mon Sep 17 00:00:00 2001 From: HerbertGao Date: Sun, 16 Nov 2025 00:26:23 +0800 Subject: [PATCH] Added support for Dot. Notifications (#1445) --- README.md | 1 + apprise/plugins/dot.py | 609 +++++++++++++++++ packaging/redhat/python-apprise.spec | 31 +- pyproject.toml | 2 + tests/test_plugin_dot.py | 967 +++++++++++++++++++++++++++ 5 files changed, 1595 insertions(+), 15 deletions(-) create mode 100644 apprise/plugins/dot.py create mode 100644 tests/test_plugin_dot.py diff --git a/README.md b/README.md index aa878cc0..ed43da2e 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,7 @@ The table below identifies the services this tool supports and some example serv | [BlueSky](https://github.com/caronc/apprise/wiki/Notify_bluesky) | bluesky:// | (TCP) 443 | bluesky://Handle:AppPw
bluesky://Handle:AppPw/TargetHandle
bluesky://Handle:AppPw/TargetHandle1/TargetHandle2/TargetHandleN | [Chanify](https://github.com/caronc/apprise/wiki/Notify_chanify) | chantify:// | (TCP) 443 | chantify://token | [Discord](https://github.com/caronc/apprise/wiki/Notify_discord) | discord:// | (TCP) 443 | discord://webhook_id/webhook_token
discord://avatar@webhook_id/webhook_token +| [Dot.](https://dot.mindreset.tech/docs/service/studio/api/text_api) | dot:// | (TCP) 443 | dot://apikey@device_id/text/
dot://apikey@device_id/image/
**Note**: `device_id` is the Quote/0 hardware serial | [Emby](https://github.com/caronc/apprise/wiki/Notify_emby) | emby:// or embys:// | (TCP) 8096 | emby://user@hostname/
emby://user:password@hostname | [Enigma2](https://github.com/caronc/apprise/wiki/Notify_enigma2) | enigma2:// or enigma2s:// | (TCP) 80 or 443 | enigma2://hostname | [FCM](https://github.com/caronc/apprise/wiki/Notify_fcm) | fcm:// | (TCP) 443 | fcm://project@apikey/DEVICE_ID
fcm://project@apikey/#TOPIC
fcm://project@apikey/DEVICE_ID1/#topic1/#topic2/DEVICE_ID2/ diff --git a/apprise/plugins/dot.py b/apprise/plugins/dot.py new file mode 100644 index 00000000..191a0a85 --- /dev/null +++ b/apprise/plugins/dot.py @@ -0,0 +1,609 @@ +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2025, 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://dot.mindreset.tech/docs/service/studio/api/text_api +# https://dot.mindreset.tech/docs/service/studio/api/image_api +# +# Text API Fields: +# - refreshNow (bool, optional, default true): controls display timing. +# - deviceId (string, required): unique device serial. +# - title (string, optional): title text shown on screen. +# - message (string, optional): body text shown on screen. +# - signature (string, optional): footer/signature text. +# - icon (string, optional): base64 PNG icon (40px x 40px). +# - link (string, optional): tap-to-interact target URL. +# +# Image API Fields: +# - refreshNow (bool, optional, default true): controls display timing. +# - deviceId (string, required): unique device serial. +# - image (string, required): base64 PNG image (296px x 152px). +# - link (string, optional): tap-to-interact target URL. +# - border (number, optional, default 0): 0=white, 1=black frame. +# - ditherType (string, optional, default DIFFUSION): dithering mode. +# - ditherKernel (string, optional, default FLOYD_STEINBERG): +# dithering kernel. + +from contextlib import suppress +import json + +import requests + +from ..common import NotifyImageSize, NotifyType +from ..locale import gettext_lazy as _ +from ..url import PrivacyMode +from ..utils.parse import parse_bool +from .base import NotifyBase + +# Supported Dither Types +DOT_DITHER_TYPES = ( + "DIFFUSION", + "ORDERED", + "NONE", +) + +# Supported Dither Kernels +DOT_DITHER_KERNELS = ( + "THRESHOLD", + "ATKINSON", + "BURKES", + "FLOYD_STEINBERG", + "SIERRA2", + "STUCKI", + "JARVIS_JUDICE_NINKE", + "DIFFUSION_ROW", + "DIFFUSION_COLUMN", + "DIFFUSION_2D", +) + + +class NotifyDot(NotifyBase): + """A wrapper for Dot. Notifications.""" + + # The default descriptive name associated with the Notification + service_name = "Dot." + # Alias: devices marketed as "Quote/0" remain discoverable. + + # The services URL + service_url = "https://dot.mindreset.tech" + + # All notification requests are secure + secure_protocol = "dot" + + # A URL that takes you to the setup/help of the specific protocol + setup_url = "https://dot.mindreset.tech/docs/service/studio/api/text_api" + + # Allows the user to specify the NotifyImageSize object + image_size = NotifyImageSize.XY_128 + + # Support Attachments + attachment_support = True + + # Supported API modes + SUPPORTED_MODES = ("text", "image") + DEFAULT_MODE = "text" + + # Define object templates + templates = ("{schema}://{apikey}@{device_id}/{mode}/",) + + # Define our template arguments + template_tokens = dict( + NotifyBase.template_tokens, + **{ + "apikey": { + "name": _("API Key"), + "type": "string", + "required": True, + "private": True, + }, + "device_id": { + "name": _("Device Serial Number"), + "type": "string", + "required": True, + "map_to": "device_id", + }, + "mode": { + "name": _("API Mode"), + "type": "choice:string", + "values": SUPPORTED_MODES, + "default": DEFAULT_MODE, + "map_to": "mode", + }, + }, + ) + + # Define our template arguments + template_args = dict( + NotifyBase.template_args, + **{ + "refresh": { + "name": _("Refresh Now"), + "type": "bool", + "default": True, + "map_to": "refresh_now", + }, + "signature": { + "name": _("Text Signature"), + "type": "string", + }, + "icon": { + "name": _("Icon Base64 (Text API)"), + "type": "string", + }, + "image": { + "name": _("Image Base64 (Image API)"), + "type": "string", + "map_to": "image_data", + }, + "link": { + "name": _("Link"), + "type": "string", + }, + "border": { + "name": _("Border"), + "type": "int", + "min": 0, + "max": 1, + "default": 0, + }, + "dither_type": { + "name": _("Dither Type"), + "type": "choice:string", + "values": DOT_DITHER_TYPES, + "default": "DIFFUSION", + }, + "dither_kernel": { + "name": _("Dither Kernel"), + "type": "choice:string", + "values": DOT_DITHER_KERNELS, + "default": "FLOYD_STEINBERG", + }, + }, + ) + # Note: + # - icon (Text API): base64 PNG icon (40px x 40px) in lower-left corner. + # Can be provided via `icon` parameter or first attachment. + # - image (Image API): base64 PNG image (296px x 152px) supplied via + # configuration `image` parameter or first attachment. + # - Only the first attachment is used; multiple attachments trigger a + # warning. + + def __init__( + self, + apikey=None, + device_id=None, + mode=DEFAULT_MODE, + refresh_now=True, + signature=None, + icon=None, + link=None, + border=None, + dither_type=None, + dither_kernel=None, + image_data=None, + **kwargs, + ): + """Initialize Notify Dot Object.""" + super().__init__(**kwargs) + + # API Key (from user) + self.apikey = apikey + + # Device ID tracks the Dot hardware serial. + self.device_id = device_id + + # Refresh Now flag: True shows content immediately (default). + self.refresh_now = parse_bool(refresh_now, default=True) + + # API mode ("text" or "image") + self.mode = ( + mode.lower() + if isinstance(mode, str) and mode.lower() in self.SUPPORTED_MODES + else self.DEFAULT_MODE + ) + if ( + not isinstance(mode, str) + or mode.lower() not in self.SUPPORTED_MODES + ): + self.logger.warning( + "Unsupported Dot mode (%s) specified; defaulting to '%s'.", + mode, + self.mode, + ) + + # Signature text used by the Text API footer. + self.signature = signature if isinstance(signature, str) else None + + # Icon for the Text API (base64 PNG 40x40, lower-left corner). + # Note: distinct from the Image API "image" field. + self.icon = icon if isinstance(icon, str) else None + + # Image payload for the Image API (base64 PNG 296x152). + self.image_data = image_data if isinstance(image_data, str) else None + if self.mode == "text" and self.image_data: + self.logger.warning( + "Image data provided in text mode; ignoring configurable" + " image payload." + ) + self.image_data = None + + # Link for tap-to-interact navigation. + self.link = link if isinstance(link, str) else None + + # Border for the Image API + self.border = border + + # Dither type for Image API + self.dither_type = dither_type + + # Dither kernel for the Image API + self.dither_kernel = dither_kernel + + # Text API endpoint + self.text_api_url = "https://dot.mindreset.tech/api/open/text" + + # Image API endpoint + self.image_api_url = "https://dot.mindreset.tech/api/open/image" + + return + + def send( + self, + body, + title="", + notify_type=NotifyType.INFO, + attach=None, + **kwargs, + ): + """Perform Dot Notification.""" + + if not self.apikey: + self.logger.warning("No API key was specified") + return False + + if not self.device_id: + self.logger.warning("No device ID was specified") + return False + + # Prepare our headers + headers = { + "Authorization": f"Bearer {self.apikey}", + "Content-Type": "application/json", + "User-Agent": self.app_id, + } + + if self.mode == "image": + if title or body: + self.logger.warning( + "Title and body are not supported in image mode " + "and will be ignored." + ) + + image_data = ( + self.image_data if isinstance(self.image_data, str) else None + ) + + # Use first attachment as image if no image_data provided + # attachment.base64() returns base64-encoded string for API + if not image_data and attach and self.attachment_support: + if len(attach) > 1: + self.logger.warning( + "Multiple attachments provided; only the first " + "one will be used as image." + ) + try: + attachment = attach[0] + if attachment: + # Convert attachment to base64-encoded string + image_data = attachment.base64() + except Exception as e: + self.logger.warning(f"Failed to process attachment: {e!s}") + + if not image_data: + self.logger.warning( + "Image API mode selected but no image data was provided." + ) + return False + + # Use Image API + # Image API payload: + # refreshNow: display timing control. + # deviceId: Dot device serial (required). + # image: base64 PNG 296x152 (required). + # link: optional tap target. + # border: optional frame color. + # ditherType: optional dithering mode. + # ditherKernel: optional dithering kernel. + payload = { + "refreshNow": self.refresh_now, + "deviceId": self.device_id, + "image": image_data, # Image payload shown on screen + } + + if self.link: + payload["link"] = self.link + + if self.border is not None: + payload["border"] = self.border + + if self.dither_type is not None: + payload["ditherType"] = self.dither_type + + if self.dither_kernel is not None: + payload["ditherKernel"] = self.dither_kernel + + api_url = self.image_api_url + + else: + # Use Text API + # Text API payload: + # refreshNow: display timing control. + # deviceId: Dot device serial (required). + # title: optional title on screen. + # message: optional body on screen. + # signature: optional footer text. + # icon: optional base64 PNG icon (40x40). + # link: optional tap target. + payload = { + "refreshNow": self.refresh_now, + "deviceId": self.device_id, + } + + if title: + payload["title"] = title + + if body: + payload["message"] = body + + if self.signature: + payload["signature"] = ( + self.signature + ) # Footer/signature displayed on screen + + # Use first attachment as icon if no icon provided + # attachment.base64() returns base64-encoded string for API + icon_data = self.icon + if not icon_data and attach and self.attachment_support: + if len(attach) > 1: + self.logger.warning( + "Multiple attachments provided; only the first " + "one will be used as icon." + ) + try: + attachment = attach[0] + if attachment: + # Convert attachment to base64-encoded string + icon_data = attachment.base64() + except Exception as e: + self.logger.warning(f"Failed to process attachment: {e!s}") + + if icon_data: + # Text API icon payload + payload["icon"] = icon_data + + if self.link: + payload["link"] = self.link + + api_url = self.text_api_url + + self.logger.debug( + "Dot POST URL:" + f" {api_url} (cert_verify={self.verify_certificate!r})" + ) + self.logger.debug(f"Dot Payload: {json.dumps(payload, indent=2)}") + + # Always call throttle before any remote server i/o is made + self.throttle() + + try: + r = requests.post( + api_url, + data=json.dumps(payload), + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + + if r.status_code == requests.codes.ok: + self.logger.info(f"Sent Dot notification to {self.device_id}.") + return True + + # We had a problem + status_str = NotifyDot.http_response_code_lookup(r.status_code) + + self.logger.warning( + "Failed to send Dot notification to {}: " + "{}{}error={}.".format( + self.device_id, + status_str, + ", " if status_str else "", + r.status_code, + ) + ) + + self.logger.debug(f"Response Details:\r\n{r.content}") + + return False + + except requests.RequestException as e: + self.logger.warning( + "A Connection error occurred sending Dot " + f"notification to {self.device_id}." + ) + self.logger.debug(f"Socket Exception: {e!s}") + + return False + + @property + def url_identifier(self): + """Returns all of the identifiers that make this URL unique from + another similar one. + """ + return ( + self.secure_protocol, + self.apikey, + self.device_id, + self.mode, + ) + + def url(self, privacy=False, *args, **kwargs): + """Returns the URL built dynamically based on specified arguments.""" + + # Define any URL parameters + params = { + "refresh": "yes" if self.refresh_now else "no", + } + + if self.mode == "text": + if self.signature: + params["signature"] = self.signature + + if self.icon: + params["icon"] = self.icon + + if self.link: + params["link"] = self.link + + else: # image mode + if self.image_data: + params["image"] = self.image_data + + if self.link: + params["link"] = self.link + + if self.border is not None: + params["border"] = str(self.border) + + if self.dither_type and self.dither_type != "DIFFUSION": + params["dither_type"] = self.dither_type + + if self.dither_kernel and self.dither_kernel != "FLOYD_STEINBERG": + params["dither_kernel"] = self.dither_kernel + + # Extend our parameters + params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) + + mode_segment = f"/{self.mode}/" + + return "{schema}://{apikey}@{device_id}{mode}?{params}".format( + schema=self.secure_protocol, + apikey=self.pprint( + self.apikey, privacy, mode=PrivacyMode.Secret, safe="" + ), + device_id=NotifyDot.quote(self.device_id, safe=""), + mode=mode_segment, + params=NotifyDot.urlencode(params), + ) + + def __len__(self): + """Returns the number of targets associated with this notification.""" + return 1 if self.device_id else 0 + + @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) + if not results: + # We're done early as we couldn't load the results + return results + + # Determine API mode from path (default to text) + mode = NotifyDot.DEFAULT_MODE + path_tokens = NotifyDot.split_path(results.get("fullpath")) + if path_tokens: + candidate = path_tokens.pop(0).lower() + if candidate in NotifyDot.SUPPORTED_MODES: + mode = candidate + else: + NotifyDot.logger.warning( + "Unsupported Dot mode (%s) detected; defaulting to '%s'.", + candidate, + NotifyDot.DEFAULT_MODE, + ) + results["mode"] = mode + remaining_path = "/".join(path_tokens) + results["fullpath"] = "/" + remaining_path if remaining_path else "/" + results["path"] = remaining_path + + # Extract API key from user + user = results.get("user") + if user: + results["apikey"] = NotifyDot.unquote(user) + + # Extract device ID from hostname + host = results.get("host") + if host: + results["device_id"] = NotifyDot.unquote(host) + + # Refresh Now + refresh_value = results["qsd"].get("refresh") + if refresh_value: + results["refresh_now"] = parse_bool(refresh_value.strip()) + + # Signature + signature_value = results["qsd"].get("signature") + if signature_value: + results["signature"] = NotifyDot.unquote(signature_value.strip()) + + # Icon + icon_value = results["qsd"].get("icon") + if icon_value: + results["icon"] = NotifyDot.unquote(icon_value.strip()) + + # Link + link_value = results["qsd"].get("link") + if link_value: + results["link"] = NotifyDot.unquote(link_value.strip()) + + # Border + border_value = results["qsd"].get("border") + if border_value: + with suppress(TypeError, ValueError): + results["border"] = int(border_value.strip()) + + # Dither Type + dither_type_value = results["qsd"].get("dither_type") + if dither_type_value: + results["dither_type"] = NotifyDot.unquote( + dither_type_value.strip() + ) + + # Dither Kernel + dither_kernel_value = results["qsd"].get("dither_kernel") + if dither_kernel_value: + results["dither_kernel"] = NotifyDot.unquote( + dither_kernel_value.strip() + ) + + # Image (Image API) + image_value = results["qsd"].get("image") + if image_value: + results["image_data"] = NotifyDot.unquote(image_value.strip()) + + return results diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 673ed931..2f61423b 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -56,26 +56,27 @@ Apprise is a Python package that simplifies access to many popular \ notification services. It supports sending alerts to platforms such as: \ \ -`46elks`, `AfricasTalking`, `Apprise API`, `APRS`, `AWS SES`, `AWS SNS`, -`Bark`, `BlueSky`, `Burst SMS`, `BulkSMS`, `BulkVS`, `Chanify`, `Clickatell`, \ -`ClickSend`, `DAPNET`, `DingTalk`, `Discord`, `E-Mail`, `Emby`, `FCM`, \ -`Feishu`, `Flock`, `Free Mobile`, `Google Chat`, `Gotify`, `Growl`, \ -`Guilded`, `Home Assistant`, `httpSMS`, `IFTTT`, `Join`, `Kavenegar`, `KODI`, \ -`Kumulos`, `LaMetric`, `Lark`, `Line`, `MacOSX`, `Mailgun`, `Mastodon`, \ -`Mattermost`, `Matrix`, `MessageBird`, `Microsoft Windows`, \ +`46elks`, `AfricasTalking`, `Apprise API`, `APRS`, `AWS SES`, `AWS SNS`, \ +`Bark`, `BlueSky`, `Burst SMS`, `BulkSMS`, `BulkVS`, `Chanify`, \ +`Clickatell`, `ClickSend`, `DAPNET`, `DingTalk`, `Discord`, \ +`Dot. (Quote/0)`, `E-Mail`, `Emby`, `FCM`, `Feishu`, `Flock`, \ +`Free Mobile`, `Google Chat`, `Gotify`, `Growl`, `Guilded`, \ +`Home Assistant`, `httpSMS`, `IFTTT`, `Join`, `Kavenegar`, `KODI`, \ +`Kumulos`, `LaMetric`, `Lark`, `Line`, `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`, \ `ParsePlatform`, `Plivo`, `PopcornNotify`, `Prowl`, `Pushalot`, \ `PushBullet`, `Pushjet`, `PushMe`, `Pushover`, `Pushplus`, `PushSafer`, \ -`Pushy`, `PushDeer`, `QQ Push`, `Revolt`, `Reddit`, `Resend`, `Rocket.Chat`, \ -`RSyslog`, `SendGrid`, `SendPulse`, `ServerChan`, `Seven`, `SFR`, `Signal`, \ -`SIGNL4`, `SimplePush`, `Sinch`, `Slack`, `SMPP`, `SMSEagle`, `SMS Manager`, \ -`SMTP2Go`, `SparkPost`, `Splunk`, `Spike`, `Spug Push`, `Super Toasty`, \ -`Streamlabs`, `Stride`, `Synology Chat`, `Syslog`, `Techulus Push`, \ -`Telegram`, `Threema Gateway`, `Twilio`, `Twitter`, `Twist`, `Vapid`, \ -`VictorOps`, `Voipms`, `Vonage`, `WebPush`, `WeCom Bot`, `WhatsApp`, \ -`Webex Teams`, `Workflows`, `WxPusher`, and `XBMC`.} +`Pushy`, `PushDeer`, `QQ Push`, `Revolt`, `Reddit`, `Resend`, \ +`Rocket.Chat`, `RSyslog`, `SendGrid`, `SendPulse`, `ServerChan`, `Seven`, \ +`SFR`, `Signal`, `SIGNL4`, `SimplePush`, `Sinch`, `Slack`, `SMPP`, \ +`SMSEagle`, `SMS Manager`, `SMTP2Go`, `SparkPost`, `Splunk`, `Spike`, \ +`Spug Push`, `Super Toasty`, `Streamlabs`, `Stride`, `Synology Chat`, \ +`Syslog`, `Techulus Push`, `Telegram`, `Threema Gateway`, `Twilio`, \ +`Twitter`, `Twist`, `Vapid`, `VictorOps`, `Voipms`, `Vonage`, `WebPush`, \ +`WeCom Bot`, `WhatsApp`, `Webex Teams`, `Workflows`, `WxPusher`, and `XBMC`.} Name: python-%{pypi_name} Version: 1.9.5 diff --git a/pyproject.toml b/pyproject.toml index 6a2e5d22..9336380b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -72,6 +72,7 @@ keywords = [ "DBus", "DingTalk", "Discord", + "Dot", "Email", "Emby", "Enigma2", @@ -134,6 +135,7 @@ keywords = [ "PushSafer", "Pushy", "QQ Push", + "Quote/0", "Reddit", "Resend", "Revolt", diff --git a/tests/test_plugin_dot.py b/tests/test_plugin_dot.py new file mode 100644 index 00000000..bee4f1f8 --- /dev/null +++ b/tests/test_plugin_dot.py @@ -0,0 +1,967 @@ +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2025, 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. + +# Disable logging for a cleaner testing output +import json +import logging +from unittest import mock +from urllib.parse import parse_qs, urlparse + +from helpers import AppriseURLTester + +from apprise.plugins.dot import NotifyDot + + +class DummyAttachment: + def __init__(self, payload="ZmFjZQ=="): + self._payload = payload + + def base64(self): + return self._payload + + +logging.disable(logging.CRITICAL) + +# Our Testing URLs +apprise_url_tests = ( + ( + "dot://", + { + # No API key or device ID + "instance": None, + }, + ), + ( + "dot://@", + { + # No device ID + "instance": None, + }, + ), + ( + "dot://apikey@", + { + # No device ID + "instance": None, + }, + ), + ( + "dot://@device_id", + { + # No API key + "instance": NotifyDot, + # Expected notify() response False (because we won't be able + # to actually notify anything if no api key was specified + "notify_response": False, + }, + ), + ( + "dot://apikey@device_id/text/", + { + # Everything is okay (text mode) + "instance": NotifyDot, + # Our expected url(privacy=True) startswith() response: + "privacy_url": "dot://****@device_id/text/", + }, + ), + ( + "dot://apikey@device_id/text/?refresh=no", + { + # Disable refresh now + "instance": NotifyDot, + }, + ), + ( + "dot://apikey@device_id/text/?signature=test_signature", + { + # With signature + "instance": NotifyDot, + }, + ), + ( + "dot://apikey@device_id/text/?link=https://example.com", + { + # With link + "instance": NotifyDot, + }, + ), + ( + "dot://apikey@device_id/image/?link=https://example.com&border=1&dither_type=ORDERED&dither_kernel=ATKINSON", + { + # Image mode without payload should fail + "instance": NotifyDot, + "notify_response": False, + "attach_response": True, + }, + ), + ( + "dot://apikey@device_id/image/?image=ZmFrZUJhc2U2NA==&link=https://example.com&border=1&dither_type=DIFFUSION&dither_kernel=FLOYD_STEINBERG", + { + # Image mode with provided image data + "instance": NotifyDot, + # Our expected url(privacy=True) startswith() response: + "privacy_url": "dot://****@device_id/image/", + }, + ), + ( + "dot://apikey@device_id/text/", + { + "instance": NotifyDot, + # throw a bizarre code forcing us to fail to look it up + "response": False, + "requests_response_code": 999, + }, + ), + ( + "dot://apikey@device_id/text/", + { + "instance": NotifyDot, + # Throws a series of i/o exceptions with this flag + # is set and tests that we gracefully handle them + "test_requests_exceptions": True, + }, + ), + ( + "dot://apikey@device_id/unknown/", + { + # Unknown mode defaults back to text + "instance": NotifyDot, + "privacy_url": "dot://****@device_id/text/", + }, + ), +) + + +def test_plugin_dot_urls(): + """NotifyDot() Apprise URLs.""" + + # Run our general tests + AppriseURLTester(tests=apprise_url_tests).run_all() + + +def test_notify_dot_image_mode_requires_image(): + dot = NotifyDot(apikey="token", device_id="device", mode="image") + assert dot.notify(title="x", body="y") is False + + +def test_notify_dot_image_mode_with_attachment(): + """Test image mode uses first attachment when no image_data provided.""" + dot = NotifyDot( + apikey="token", + device_id="device", + mode="image", + link="https://example.com", + border=1, + dither_type="ORDERED", + dither_kernel="ATKINSON", + ) + + response = mock.Mock() + response.status_code = 200 + + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send( + body="payload", title="title", attach=[DummyAttachment("YmFzZTY0")] + ) + + assert mock_post.called + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert payload["image"] == "YmFzZTY0" + assert payload["deviceId"] == "device" + + +def test_notify_dot_image_mode_with_existing_image_data(): + """Test image mode ignores attachment when image_data is provided.""" + dot = NotifyDot( + apikey="token", + device_id="device", + mode="image", + image_data="existing_image_data", + ) + + response = mock.Mock() + response.status_code = 200 + + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send( + body="test", + title="test", + attach=[DummyAttachment("attachment_data")], + ) + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + # Should use existing image_data, not attachment + assert payload["image"] == "existing_image_data" + + +def test_notify_dot_text_mode_with_existing_icon(): + """Test text mode with existing icon (attachment should be ignored).""" + dot = NotifyDot( + apikey="token", + device_id="device", + signature="footer", + icon="aW5jb24=", + ) + + response = mock.Mock() + response.status_code = 200 + + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send( + title="hello", + body="world", + attach=[DummyAttachment("attachment_icon")], + ) + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert "image" not in payload + assert payload["deviceId"] == "device" + assert payload["message"] == "world" + # Should use existing icon, not attachment + assert payload["icon"] == "aW5jb24=" + + +def test_notify_dot_text_mode_uses_attachment_as_icon(): + """Test text mode uses first attachment as icon when no icon provided.""" + dot = NotifyDot( + apikey="token", + device_id="device", + signature="footer", + ) + + response = mock.Mock() + response.status_code = 200 + + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send( + title="hello", + body="world", + attach=[DummyAttachment("attachment_icon_data")], + ) + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert payload["deviceId"] == "device" + assert payload["message"] == "world" + # Should use attachment as icon + assert payload["icon"] == "attachment_icon_data" + + +def test_notify_dot_text_mode_multiple_attachments_warning(): + """Test text mode warns when multiple attachments are provided.""" + dot = NotifyDot( + apikey="token", + device_id="device", + ) + + response = mock.Mock() + response.status_code = 200 + + with ( + mock.patch("requests.post", return_value=response) as mock_post, + mock.patch.object(dot.logger, "warning") as mock_warning, + ): + assert dot.send( + title="hello", + body="world", + attach=[ + DummyAttachment("first"), + DummyAttachment("second"), + ], + ) + # Should warn about multiple attachments + mock_warning.assert_called_once() + assert "Multiple attachments" in str(mock_warning.call_args) + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + # Should use first attachment only + assert payload["icon"] == "first" + + +def test_notify_dot_url_generation(): + text_dot = NotifyDot( + apikey="token", + device_id="device", + signature="sig", + icon="aW5jb24=", + ) + text_url = text_dot.url() + parsed = urlparse(text_url) + assert parsed.path.endswith("/text/") + query = parse_qs(parsed.query) + assert query["refresh"] == ["yes"] + assert query["signature"] == ["sig"] + + image_dot = NotifyDot( + apikey="token", + device_id="device", + mode="image", + image_data="aW1hZ2U=", + link="https://example.com", + border=1, + dither_type="ORDERED", + dither_kernel="ATKINSON", + ) + image_url = image_dot.url() + parsed_image = urlparse(image_url) + assert parsed_image.path.endswith("/image/") + image_query = parse_qs(parsed_image.query) + assert image_query["image"] == ["aW1hZ2U="] + assert image_query["border"] == ["1"] + + +def test_notify_dot_parse_url_mode_and_image(): + result = NotifyDot.parse_url( + "dot://token@device/image/?image=Zm9vYmFy&link=https://example.com" + ) + assert result["mode"] == "image" + assert result["image_data"] == "Zm9vYmFy" + assert result["link"] == "https://example.com" + + fallback = NotifyDot.parse_url("dot://token@device/unknown/?refresh=no") + assert fallback["mode"] == "text" + assert fallback["refresh_now"] is False + + +def test_notify_dot_invalid_mode(): + """Test invalid mode handling.""" + dot = NotifyDot(apikey="token", device_id="device", mode="invalid_mode") + assert dot.mode == "text" + + dot = NotifyDot(apikey="token", device_id="device", mode=123) + assert dot.mode == "text" + + +def test_notify_dot_image_data_in_text_mode(): + """Test that image_data is ignored in text mode.""" + dot = NotifyDot( + apikey="token", + device_id="device", + mode="text", + image_data="somebase64", + ) + assert dot.image_data is None + + +def test_notify_dot_text_mode_with_title_and_body(): + """Test text mode with title and body.""" + dot = NotifyDot( + apikey="token", + device_id="device", + ) + + response = mock.Mock() + response.status_code = 200 + + # Test with title and body provided at runtime + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send(body="test_body", title="test_title") + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert payload["message"] == "test_body" + assert payload["title"] == "test_title" + + +def test_notify_dot_no_device_id(): + """Test behavior when device_id is missing.""" + dot = NotifyDot(apikey="token", device_id=None) + assert dot.notify(title="test", body="test") is False + assert len(dot) == 0 + + +def test_notify_dot_parse_url_with_all_params(): + """Test parse_url with all parameters.""" + result = NotifyDot.parse_url( + "dot://apikey@device/image/?refresh=yes&signature=sig&icon=icon_b64" + "&link=https://example.com&border=1&dither_type=ORDERED" + "&dither_kernel=ATKINSON&image=img_b64" + ) + assert result["mode"] == "image" + assert result["refresh_now"] is True + assert result["signature"] == "sig" + assert result["icon"] == "icon_b64" + assert result["link"] == "https://example.com" + assert result["border"] == 1 + assert result["dither_type"] == "ORDERED" + assert result["dither_kernel"] == "ATKINSON" + assert result["image_data"] == "img_b64" + + +def test_notify_dot_url_identifier(): + """Test url_identifier property.""" + dot = NotifyDot(apikey="token", device_id="device", mode="image") + identifier = dot.url_identifier + assert identifier == ("dot", "token", "device", "image") + + +def test_notify_dot_image_mode_with_failed_attachment(): + """Test image mode when attachment fails to convert.""" + + class FailedAttachment: + def base64(self): + raise Exception("Conversion failed") + + dot = NotifyDot(apikey="token", device_id="device", mode="image") + # Should fail when no valid image data is available + assert dot.notify( + title="test", body="test", attach=[FailedAttachment()] + ) is False + + +def test_notify_dot_url_generation_defaults(): + """Test URL generation with default values.""" + dot = NotifyDot(apikey="token", device_id="device") + url = dot.url() + assert "refresh=yes" in url + assert "/text/" in url + + # Test image mode URL with non-default values + dot_image = NotifyDot( + apikey="token", + device_id="device", + mode="image", + image_data="img", + dither_type="ORDERED", + dither_kernel="ATKINSON", + ) + url_image = dot_image.url() + assert "/image/" in url_image + assert "dither_type=ORDERED" in url_image + assert "dither_kernel=ATKINSON" in url_image + + +def test_notify_dot_image_mode_with_multiple_attachments(): + """Test image mode with multiple attachments (only first is used).""" + + dot = NotifyDot(apikey="token", device_id="device", mode="image") + + response = mock.Mock() + response.status_code = 200 + + # Multiple attachments provided, only first should be used + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send( + body="test", + title="test", + attach=[ + DummyAttachment("first_attachment"), + DummyAttachment("second_attachment"), + ], + ) + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + # Should use first attachment only + assert payload["image"] == "first_attachment" + + +def test_notify_dot_text_mode_without_title(): + """Test text mode without title (title is optional).""" + dot = NotifyDot(apikey="token", device_id="device") + + response = mock.Mock() + response.status_code = 200 + + # Test with empty title - title should not be in payload + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send(body="test message", title="") + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + # Title should not be in payload when empty + assert "title" not in payload + assert payload["message"] == "test message" + + +def test_notify_dot_url_generation_with_link(): + """Test URL generation with link in text mode.""" + dot = NotifyDot( + apikey="token", + device_id="device", + link="https://example.com", + ) + url = dot.url() + assert "link=" in url + + # Test image mode with border=0 (should not appear in URL for default) + dot_image = NotifyDot( + apikey="token", + device_id="device", + mode="image", + image_data="img", + border=0, + ) + url_image = dot_image.url() + assert "border=0" in url_image + + +def test_notify_dot_title_handling(): + """Test title handling in text mode.""" + dot = NotifyDot( + apikey="token", + device_id="device", + ) + + response = mock.Mock() + response.status_code = 200 + + # Test 1: With provided title + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send(body="test", title="provided_title") + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert payload["title"] == "provided_title" + + # Test 2: Without provided title, should not include title in payload + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send(body="test", title="") + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + # Title should not be in payload when empty + assert "title" not in payload + + +def test_notify_dot_image_mode_no_border(): + """Test image mode with border=None to skip border in payload.""" + dot = NotifyDot( + apikey="token", + device_id="device", + mode="image", + image_data="base64img", + ) + # Manually set border to None + dot.border = None + + response = mock.Mock() + response.status_code = 200 + + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send(body="test", title="test") + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + # Border should not be in payload when None + assert "border" not in payload + + +def test_notify_dot_image_mode_no_dither(): + """Test image mode with no dither_type/dither_kernel.""" + dot = NotifyDot( + apikey="token", + device_id="device", + mode="image", + image_data="base64img", + ) + # Manually set to None to test the conditional branches + dot.dither_type = None + dot.dither_kernel = None + + response = mock.Mock() + response.status_code = 200 + + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send(body="test", title="test") + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + # Dither fields should not be in payload when None + assert "ditherType" not in payload + assert "ditherKernel" not in payload + + +def test_notify_dot_text_mode_no_optional_fields(): + """Test text mode with no signature, icon, or link.""" + dot = NotifyDot( + apikey="token", + device_id="device", + ) + + response = mock.Mock() + response.status_code = 200 + + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send(body="test body", title="test title") + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert "signature" not in payload + assert "icon" not in payload + # Link should not be in payload when not set + assert payload.get("link") is None or "link" not in payload + + +def test_notify_dot_url_generation_without_defaults(): + """Test URL generation without default dither values.""" + # Test with DIFFUSION (default) - should not appear in URL + dot = NotifyDot( + apikey="token", + device_id="device", + mode="image", + image_data="img", + dither_type="DIFFUSION", + dither_kernel="FLOYD_STEINBERG", + ) + url = dot.url() + # Default values should not appear in URL + assert "dither_type" not in url + assert "dither_kernel" not in url + + +def test_notify_dot_image_mode_attachment_exception(): + """Test exception handling in image mode when attachment.base64() fails.""" + + class ExceptionAttachment: + def base64(self): + raise Exception("First attachment fails") + + dot = NotifyDot(apikey="token", device_id="device", mode="image") + + # First attachment throws exception, should log warning and fail + with mock.patch.object(dot.logger, "warning") as mock_warning: + assert dot.send( + body="test", + title="test", + attach=[ExceptionAttachment()], + ) is False + # Should log warning about failed attachment processing + assert mock_warning.called + # Check that the warning message contains expected text + warning_calls = [str(call) for call in mock_warning.call_args_list] + assert any( + "Failed to process attachment" in str(call) + for call in warning_calls + ) + + +def test_notify_dot_image_mode_attachment_none(): + """Test image mode when attachment is None.""" + dot = NotifyDot(apikey="token", device_id="device", mode="image") + + # Attachment is None, should skip base64() call and fail + assert dot.send( + body="test", + title="test", + attach=[None], + ) is False + + +def test_notify_dot_image_mode_attachment_falsy(): + """Test image mode when attachment is falsy.""" + dot = NotifyDot(apikey="token", device_id="device", mode="image") + + # Attachment is falsy (empty string), should skip base64() call and fail + class FalsyAttachment: + def __bool__(self): + return False + + def base64(self): + return "should_not_be_called" + + assert dot.send( + body="test", + title="test", + attach=[FalsyAttachment()], + ) is False + + +def test_notify_dot_text_mode_attachment_exception(): + """Test exception handling in text mode when attachment.base64() fails.""" + + class ExceptionAttachment: + def base64(self): + raise Exception("Attachment base64 conversion fails") + + dot = NotifyDot(apikey="token", device_id="device", mode="text") + + response = mock.Mock() + response.status_code = 200 + + # First attachment throws exception, should log warning but continue + with ( + mock.patch("requests.post", return_value=response) as mock_post, + mock.patch.object(dot.logger, "warning") as mock_warning, + ): + assert dot.send( + title="hello", + body="world", + attach=[ExceptionAttachment()], + ) + # Should log warning about failed attachment processing + assert mock_warning.called + # Check that the warning message contains expected text + warning_calls = [str(call) for call in mock_warning.call_args_list] + assert any( + "Failed to process attachment" in str(call) + for call in warning_calls + ) + + # Should still send notification without icon + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert payload["message"] == "world" + assert "icon" not in payload + + +def test_notify_dot_text_mode_attachment_none(): + """Test text mode when attachment is None (covers if attachment branch).""" + dot = NotifyDot(apikey="token", device_id="device", mode="text") + + response = mock.Mock() + response.status_code = 200 + + # Attachment is None, should skip base64() call and continue without icon + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send( + title="hello", + body="world", + attach=[None], + ) + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert payload["message"] == "world" + assert "icon" not in payload + + +def test_notify_dot_text_mode_attachment_falsy(): + """Test text mode when attachment is falsy.""" + dot = NotifyDot(apikey="token", device_id="device", mode="text") + + response = mock.Mock() + response.status_code = 200 + + # Attachment is falsy, should skip base64() call and continue without icon + class FalsyAttachment: + def __bool__(self): + return False + + def base64(self): + return "should_not_be_called" + + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send( + title="hello", + body="world", + attach=[FalsyAttachment()], + ) + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert payload["message"] == "world" + assert "icon" not in payload + + +def test_notify_dot_parse_url_no_host(): + """Test parse_url when host is empty (line 578).""" + # Test URL with empty host - device_id should not be added + # Using a valid URL structure but testing when host is explicitly empty + result = NotifyDot.parse_url("dot://apikey@device/text/") + # This should succeed and have a device_id + assert result is not None + assert result.get("device_id") == "device" + + +def test_notify_dot_url_with_border_not_none(): + """Test URL generation when border is not None (line 515).""" + dot = NotifyDot( + apikey="token", + device_id="device", + mode="image", + image_data="img", + border=1, + ) + url = dot.url() + # Border should be in URL when not None + assert "border=1" in url + + +def test_notify_dot_image_mode_with_only_title(): + """Test image mode warning with only title (no body).""" + dot = NotifyDot(apikey="token", device_id="device", mode="image") + + response = mock.Mock() + response.status_code = 200 + + # Test with only title, no body - should still warn + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send( + title="test_title", + body="", + attach=[DummyAttachment("image_data")], + ) + + # Should have sent the notification but logged a warning + assert mock_post.called + + +def test_notify_dot_image_mode_with_only_body(): + """Test image mode warning with only body (no title).""" + dot = NotifyDot(apikey="token", device_id="device", mode="image") + + response = mock.Mock() + response.status_code = 200 + + # Test with only body, no title - should still warn + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send( + title="", + body="test_body", + attach=[DummyAttachment("image_data")], + ) + + # Should have sent the notification but logged a warning + assert mock_post.called + + +def test_notify_dot_text_mode_without_body(): + """Test text mode with empty body.""" + dot = NotifyDot(apikey="token", device_id="device") + + response = mock.Mock() + response.status_code = 200 + + # Test with title but no body + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send(body="", title="test_title") + + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert payload["title"] == "test_title" + # Body should not be in payload when empty + assert "message" not in payload + + +def test_notify_dot_parse_url_without_host(): + """Test parse_url when URL has no host.""" + # URL with no host (missing device_id) - should return None + result = NotifyDot.parse_url("dot://apikey@/text/") + # Without a host, the URL is invalid and parse_url returns None + assert result is None + + +def test_notify_dot_image_mode_without_title_and_body(): + """Test image mode without title and body (line 294->300).""" + dot = NotifyDot( + apikey="token", + device_id="device", + mode="image", + image_data="base64img", + ) + + response = mock.Mock() + response.status_code = 200 + + # Send without title and body - should not trigger warning + with mock.patch("requests.post", return_value=response) as mock_post: + assert dot.send(title="", body="") + + # Should have sent the notification + assert mock_post.called + _args, kwargs = mock_post.call_args + payload = json.loads(kwargs["data"]) + assert payload["image"] == "base64img" + assert "title" not in payload + assert "message" not in payload + + +def test_notify_dot_parse_url_with_empty_refresh(): + """Test parse_url when refresh query parameter is empty (line 535->539).""" + # Test with no refresh parameter (should default to True) + result = NotifyDot.parse_url("dot://apikey@device/text/") + assert result is not None + # When refresh is not specified, it defaults to True + assert result.get("refresh_now") is None # Not set in parse_url + + +def test_notify_dot_image_mode_first_attachment_fails(): + """Test image mode when first attachment fails (returns None).""" + + class FailingAttachment: + def base64(self): + return None # Returns None + + dot = NotifyDot(apikey="token", device_id="device", mode="image") + + # First attachment returns None, should fail immediately + assert dot.notify( + title="", + body="", + attach=[FailingAttachment()], + ) is False + + +def test_notify_dot_image_mode_with_empty_attach_list(): + """Test image mode with empty attachments list (line 305->313).""" + dot = NotifyDot(apikey="token", device_id="device", mode="image") + + # Try with empty attachments list + # Condition: not image_data and attach -> not None and [] -> False + # Should skip the for loop and go directly to line 313 + assert dot.notify( + title="", + body="", + attach=[], # Empty list (truthy in Python but loop won't execute) + ) is False + + +def test_notify_dot_parse_url_without_host_field(): + """Test parse_url when host field is None (line 535->539).""" + from apprise import NotifyBase + + # Mock NotifyBase.parse_url to return results with host=None + # This triggers the else branch of "if host:" at line 535 + with mock.patch.object(NotifyBase, "parse_url") as mock_parse: + mock_parse.return_value = { + "user": "apikey", + "password": None, + "port": None, + "host": None, # host is None - triggers 535->539 branch + "fullpath": "/text/", + "path": "", + "query": None, + "schema": "dot", + "qsd": {"refresh": "yes"}, + "secure": False, + "verify": True, + } + + result = NotifyDot.parse_url("dot://fake") + + # Should have mode but no device_id since host was None + assert result is not None + assert result.get("mode") == "text" + assert result.get("device_id") is None + assert result.get("apikey") == "apikey" + assert result.get("refresh_now") is True # refresh was in qsd +