diff --git a/.gitignore b/.gitignore index 4ae5f588..355dc8c0 100644 --- a/.gitignore +++ b/.gitignore @@ -69,3 +69,4 @@ target/ .project .pydevproject .settings +.DS_Store diff --git a/apprise/plugins/NotifyAprs.py b/apprise/plugins/NotifyAprs.py new file mode 100644 index 00000000..981c01da --- /dev/null +++ b/apprise/plugins/NotifyAprs.py @@ -0,0 +1,741 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2023, 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. +# +# To use this plugin, you need to be a licensed ham radio operator +# +# Plugin constraints: +# +# - message length = 67 chars max. +# - message content = ASCII 7 bit +# - APRS messages will be sent without msg ID, meaning that +# ham radio operators cannot acknowledge them +# - Bring your own APRS-IS passcode. If you don't know what +# this is or how to get it, then this plugin is not for you +# - Do NOT change the Device/ToCall ID setting UNLESS this +# module is used outside of Apprise. This identifier helps +# the ham radio community with determining the software behind +# a given APRS message. +# - With great (ham radio) power comes great responsibility; do +# not use this plugin for spamming other ham radio operators + +# +# In order to digest text input which is not in plain English, +# users can install the optional 'unidecode' package as part +# of their venv environment. Details: see plugin description +# + +# +# You're done at this point, you only need to know your user/pass that +# you signed up with. + +# The following URLs would be accepted by Apprise: +# - aprs://{user}:{password}@{callsign} +# - aprs://{user}:{password}@{callsign1}/{callsign2} + +# Optional parameters: +# - locale --> APRS-IS target server to connect with +# Default: EURO --> 'euro.aprs2.net' +# Details: https://www.aprs2.net/ + +# +# APRS message format specification: +# http://www.aprs.org/doc/APRS101.PDF +# + +import socket +import sys +from itertools import chain +from .NotifyBase import NotifyBase +from ..AppriseLocale import gettext_lazy as _ +from ..URLBase import PrivacyMode +from ..common import NotifyType +from ..utils import is_call_sign +from ..utils import parse_call_sign +from .. import __version__ +import re + +# Fixed APRS-IS server locales +# Default is 'EURO' +# See https://www.aprs2.net/ for details +# Select the rotating server in case you +# don"t care about a specific locale +APRS_LOCALES = { + "NOAM": "noam.aprs2.net", + "SOAM": "soam.aprs2.net", + "EURO": "euro.aprs2.net", + "ASIA": "asia.aprs2.net", + "AUNZ": "aunz.aprs2.net", + "ROTA": "rotate.aprs2.net", +} + +# Identify all unsupported characters +APRS_BAD_CHARMAP = { + r"Ä": "Ae", + r"Ö": "Oe", + r"Ü": "Ue", + r"ä": "ae", + r"ö": "oe", + r"ü": "ue", + r"ß": "ss", +} + +# Our compiled mapping of bad characters +APRS_COMPILED_MAP = re.compile( + r'(' + '|'.join(APRS_BAD_CHARMAP.keys()) + r')') + + +class NotifyAprs(NotifyBase): + """ + A wrapper for APRS Notifications via APRS-IS + """ + + # The default descriptive name associated with the Notification + service_name = "Aprs" + + # The services URL + service_url = "https://www.aprs2.net/" + + # The default secure protocol + secure_protocol = "aprs" + + # A URL that takes you to the setup/help of the specific protocol + setup_url = "https://github.com/caronc/apprise/wiki/Notify_aprs" + + # APRS default port, supported by all core servers + # Details: https://www.aprs-is.net/Connecting.aspx + notify_port = 10152 + + # The maximum length of the APRS message body + body_maxlen = 67 + + # Apprise APRS Device ID / TOCALL ID + # This is a FIXED value which is associated with this plugin. + # Its value MUST NOT be changed. If you use this APRS plugin + # code OUTSIDE of Apprise, please request your own TOCALL ID. + # Details: see https://github.com/aprsorg/aprs-deviceid + # + # Do NOT use the generic "APRS" TOCALL ID !!!!! + # + device_id = "APPRIS" + + # A title can not be used for APRS Messages. Setting this to zero will + # cause any title (if defined) to get placed into the message body. + title_maxlen = 0 + + # Helps to reduce the number of login-related errors where the + # APRS-IS server "isn't ready yet". If we try to receive the rx buffer + # without this grace perid in place, we may receive "incomplete" responses + # where the login response lacks information. In case you receive too many + # "Rx: APRS-IS msg is too short - needs to have at least two lines" error + # messages, you might want to increase this value to a larger time span + # Per previous experience, do not use values lower than 0.5 (seconds) + request_rate_per_sec = 0.8 + + # Encoding of retrieved content + aprs_encoding = 'latin-1' + + # Define object templates + templates = ("{schema}://{user}:{password}@{targets}",) + + # Define our template tokens + template_tokens = dict( + NotifyBase.template_tokens, + **{ + "user": { + "name": _("User Name"), + "type": "string", + "required": True, + }, + "password": { + "name": _("Password"), + "type": "string", + "private": True, + "required": True, + }, + "target_callsign": { + "name": _("Target Callsign"), + "type": "string", + "regex": ( + r"^[a-z0-9]{2,5}(-[a-z0-9]{1,2})?$", + "i", + ), + "map_to": "targets", + }, + "targets": { + "name": _("Targets"), + "type": "list:string", + "required": True, + }, + } + ) + + # Define our template arguments + template_args = dict( + NotifyBase.template_args, + **{ + "to": { + "name": _("Target Callsign"), + "type": "string", + "map_to": "targets", + }, + "locale": { + "name": _("Locale"), + "type": "choice:string", + "values": APRS_LOCALES, + "default": "EURO", + }, + } + ) + + def __init__(self, targets=None, locale=None, **kwargs): + """ + Initialize APRS Object + """ + super().__init__(**kwargs) + + # Our (future) socket sobject + self.sock = None + + # Parse our targets + self.targets = list() + + """ + Check if the user has provided credentials + """ + if not (self.user and self.password): + msg = "An APRS user/pass was not provided." + self.logger.warning(msg) + raise TypeError(msg) + + """ + Check if the user tries to use a read-only access + to APRS-IS. We need to send content, meaning that + read-only access will not work + """ + if self.password == "-1": + msg = "APRS read-only passwords are not supported." + self.logger.warning(msg) + raise TypeError(msg) + + """ + Check if the password is numeric + """ + if not self.password.isnumeric(): + msg = "Invalid APRS-IS password" + self.logger.warning(msg) + raise TypeError(msg) + + """ + Convert given user name (FROM callsign) and + device ID to to uppercase + """ + self.user = self.user.upper() + self.device_id = self.device_id.upper() + + """ + Check if the user has provided a locale for the + APRS-IS-server and validate it, if necessary + """ + if locale: + if locale.upper() not in APRS_LOCALES: + msg = ( + "Unsupported APRS-IS server locale. " + "Received: {}. Valid: {}".format( + locale, ", ".join(str(x) for x in APRS_LOCALES.keys()) + ) + ) + self.logger.warning(msg) + raise TypeError(msg) + + # Set the transmitter group + self.locale = \ + NotifyAprs.template_args["locale"]["default"] \ + if not locale else locale.upper() + + # Used for URL generation afterwards only + self.invalid_targets = list() + + for target in parse_call_sign(targets): + # Validate targets and drop bad ones + # We just need to know if the call sign (including SSID, if + # provided) is valid and can then process the input as is + result = is_call_sign(target) + if not result: + self.logger.warning( + "Dropping invalid Amateur radio call sign ({}).".format( + target + ), + ) + self.invalid_targets.append(target.upper()) + continue + + # Store entry + self.targets.append(target.upper()) + + return + + def socket_close(self): + """ + Closes the socket connection whereas present + """ + if self.sock: + try: + self.sock.close() + + except Exception: + # No worries if socket exception thrown on close() + pass + + self.sock = None + + def socket_open(self): + """ + Establishes the connection to the APRS-IS + socket server + """ + self.logger.debug( + "Creating socket connection with APRS-IS {}:{}".format( + APRS_LOCALES[self.locale], self.notify_port + ) + ) + + try: + self.sock = socket.create_connection( + (APRS_LOCALES[self.locale], self.notify_port), + self.socket_connect_timeout, + ) + + except ConnectionError as e: + self.logger.debug("Socket Exception socket_open: %s", str(e)) + self.sock = None + return False + + except socket.gaierror as e: + self.logger.debug("Socket Exception socket_open: %s", str(e)) + self.sock = None + return False + + except socket.timeout as e: + self.logger.debug( + "Socket Timeout Exception socket_open: %s", str(e)) + self.sock = None + return False + + except Exception as e: + self.logger.debug("General Exception socket_open: %s", str(e)) + self.sock = None + return False + + # We are connected. + # getpeername() is not supported by every OS. Therefore, + # we MAY receive an exception even though we are + # connected successfully. + try: + # Get the physical host/port of the server + host, port = self.sock.getpeername() + # and create debug info + self.logger.debug("Connected to {}:{}".format(host, port)) + + except ValueError: + # Seens as if we are running on an operating + # system that does not support getpeername() + # Create a minimal log file entry + self.logger.debug("Connected to APRS-IS") + + # Return success + return True + + def aprsis_login(self): + """ + Generate the APRS-IS login string, send it to the server + and parse the response + + Returns True/False wrt whether the login was successful + """ + self.logger.debug("socket_login: init") + + # Check if we are connected + if not self.sock: + self.logger.warning("socket_login: Not connected to APRS-IS") + return False + + # APRS-IS login string, see https://www.aprs-is.net/Connecting.aspx + login_str = "user {0} pass {1} vers apprise {2}\r\n".format( + self.user, self.password, __version__ + ) + + # Send the data & abort in case of error + if not self.socket_send(login_str): + self.logger.warning( + "socket_login: Login to APRS-IS unsuccessful," + " exception occurred" + ) + self.socket_close() + return False + + rx_buf = self.socket_receive(len(login_str) + 100) + # Abort the remaining process in case an error has occurred + if not rx_buf: + self.logger.warning( + "socket_login: Login to APRS-IS " + "unsuccessful, exception occurred" + ) + self.socket_close() + return False + + # APRS-IS sends at least two lines of data + # The data that we need is in line #2 so + # let's split the content and see what we have + rx_lines = rx_buf.splitlines() + if len(rx_lines) < 2: + self.logger.warning( + "socket_login: APRS-IS msg is too short" + " - needs to have at least two lines" + ) + self.socket_close() + return False + + # Now split the 2nd line's content and extract + # both call sign and login status + try: + _, _, callsign, status, _ = rx_lines[1].split(" ", 4) + + except ValueError: + # ValueError is returned if there were not enough elements to + # populate the response + self.logger.warning( + "socket_login: " "received invalid response from APRS-IS" + ) + self.socket_close() + return False + + if callsign != self.user: + self.logger.warning( + "socket_login: " "call signs differ: %s" % callsign + ) + self.socket_close() + return False + + if status.startswith("unverified"): + self.logger.warning( + "socket_login: " + "invalid APRS-IS password for given call sign" + ) + self.socket_close() + return False + + # all validations are successful; we are connected + return True + + def socket_send(self, tx_data): + """ + Generic "Send data to a socket" + """ + self.logger.debug("socket_send: init") + + # Check if we are connected + if not self.sock: + self.logger.warning("socket_send: Not connected to APRS-IS") + return False + + # Encode our data if we are on Python3 or later + payload = ( + tx_data.encode("utf-8") if sys.version_info[0] >= 3 else tx_data + ) + + # Always call throttle before any remote server i/o is made + self.throttle() + + # Try to open the socket + # Send the content to APRS-IS + try: + self.sock.setblocking(True) + self.sock.settimeout(self.socket_connect_timeout) + self.sock.sendall(payload) + + except socket.gaierror as e: + self.logger.warning("Socket Exception socket_send: %s" % str(e)) + self.sock = None + return False + + except socket.timeout as e: + self.logger.warning( + "Socket Timeout Exception " "socket_send: %s" % str(e) + ) + self.sock = None + return False + + except Exception as e: + self.logger.warning( + "General Exception " "socket_send: %s" % str(e) + ) + self.sock = None + return False + + self.logger.debug("socket_send: successful") + + # mandatory on several APRS-IS servers + # helps to reduce the number of errors where + # the server only returns an abbreviated message + return True + + def socket_reset(self): + """ + Resets the socket's buffer + """ + self.logger.debug("socket_reset: init") + _ = self.socket_receive(0) + self.logger.debug("socket_reset: successful") + return True + + def socket_receive(self, rx_len): + """ + Generic "Receive data from a socket" + """ + self.logger.debug("socket_receive: init") + + # Check if we are connected + if not self.sock: + self.logger.warning("socket_receive: not connected to APRS-IS") + return False + + # len is zero in case we intend to + # reset the socket + if rx_len > 0: + self.logger.debug("socket_receive: Receiving data from APRS-IS") + + # Receive content from the socket + try: + self.sock.setblocking(False) + self.sock.settimeout(self.socket_connect_timeout) + rx_buf = self.sock.recv(rx_len) + + except socket.gaierror as e: + self.logger.warning( + "Socket Exception socket_receive: %s" % str(e) + ) + self.sock = None + return False + + except socket.timeout as e: + self.logger.warning( + "Socket Timeout Exception " "socket_receive: %s" % str(e) + ) + self.sock = None + return False + + except Exception as e: + self.logger.warning( + "General Exception " "socket_receive: %s" % str(e) + ) + self.sock = None + return False + + rx_buf = ( + rx_buf.decode(self.aprs_encoding) + if sys.version_info[0] >= 3 else rx_buf + ) + + # There will be no data in case we reset the socket + if rx_len > 0: + self.logger.debug("Received content: {}".format(rx_buf)) + + self.logger.debug("socket_receive: successful") + + return rx_buf.rstrip() + + def send(self, body, title="", notify_type=NotifyType.INFO, **kwargs): + """ + Perform APRS Notification + """ + + if not self.targets: + # There is no one to notify; we're done + self.logger.warning( + "There are no amateur radio call signs to notify" + ) + return False + + # prepare payload + payload = body + + # sock object is "None" if we were unable to establish a connection + # In case of errors, the error message has already been sent + # to the logger object + if not self.socket_open(): + return False + + # We have established a successful connection + # to the socket server. Now send the login information + if not self.aprsis_login(): + return False + + # Login & authorization confirmed + # reset what is in our buffer + self.socket_reset() + + # error tracking (used for function return) + has_error = False + + # Create a copy of the targets list + targets = list(self.targets) + + self.logger.debug("Starting Payload setup") + + # Prepare the outgoing message + # Due to APRS's contraints, we need to do + # a lot of filtering before we can send + # the actual message + # + # First remove all characters from the + # payload that would break APRS + # see https://www.aprs.org/doc/APRS101.PDF pg. 71 + payload = re.sub("[{}|~]+", "", payload) + + payload = ( + APRS_COMPILED_MAP.sub( + lambda x: APRS_BAD_CHARMAP[x.group()], payload) + ) + + # Finally, constrain output string to 67 characters as + # APRS messages are limited in length + payload = payload[:67] + + # Our outgoing message MUST end with a CRLF so + # let's amend our payload respectively + payload = payload.rstrip("\r\n") + "\r\n" + + self.logger.debug("Payload setup complete: {}".format(payload)) + + # send the message to our target call sign(s) + for index in range(0, len(targets)): + # prepare the output string + # Format: + # Device ID/TOCALL - our call sign - target call sign - body + buffer = "{}>{}::{:9}:{}".format( + self.user, self.device_id, targets[index], payload + ) + + # and send the content to the socket + # Note that there will be no response from APRS and + # that all exceptions are handled within the 'send' method + self.logger.debug("Sending APRS message: {}".format(buffer)) + + # send the content + if not self.socket_send(buffer): + has_error = True + break + + # Finally, reset our socket buffer + # we DO NOT read from the socket as we + # would simply listen to the default APRS-IS stream + self.socket_reset() + + self.logger.debug("Closing socket.") + self.socket_close() + self.logger.info( + "Sent %d/%d APRS-IS notification(s)", index + 1, len(targets)) + return not has_error + + def url(self, privacy=False, *args, **kwargs): + """ + Returns the URL built dynamically based on specified arguments. + """ + + # Define any URL parameters + params = {} + + if self.locale != NotifyAprs.template_args["locale"]["default"]: + # Store our locale if not default + params['locale'] = self.locale + + # Extend our parameters + params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) + + # Setup Authentication + auth = "{user}:{password}@".format( + user=NotifyAprs.quote(self.user, safe=""), + password=self.pprint( + self.password, privacy, mode=PrivacyMode.Secret, safe="" + ), + ) + + return "{schema}://{auth}{targets}?{params}".format( + schema=self.secure_protocol, + auth=auth, + targets="/".join(chain( + [self.pprint(x, privacy, safe="") for x in self.targets], + [self.pprint(x, privacy, safe="") + for x in self.invalid_targets], + )), + params=NotifyAprs.urlencode(params), + ) + + def __len__(self): + """ + Returns the number of targets associated with this notification + """ + targets = len(self.targets) + return targets if targets > 0 else 1 + + def __del__(self): + """ + Ensure we close any lingering connections + """ + self.socket_close() + + @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 + + # All elements are targets + results["targets"] = [NotifyAprs.unquote(results["host"])] + + # All entries after the hostname are additional targets + results["targets"].extend(NotifyAprs.split_path(results["fullpath"])) + + # Support the 'to' variable so that we can support rooms this way too + # The 'to' makes it easier to use yaml configuration + if "to" in results["qsd"] and len(results["qsd"]["to"]): + results["targets"] += NotifyAprs.parse_list(results["qsd"]["to"]) + + # Set our APRS-IS server locale's key value and convert it to uppercase + if "locale" in results["qsd"] and len(results["qsd"]["locale"]): + results["locale"] = NotifyAprs.unquote( + results["qsd"]["locale"] + ).upper() + + return results diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 51855961..ffa04798 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -39,18 +39,19 @@ Apprise is a Python package for simplifying access to all of the different notification services that are out there. Apprise opens the door and makes it easy to access: -Apprise API, AWS SES, AWS SNS, Bark, Boxcar, Burst SMS, BulkSMS, ClickSend, -DAPNET, DingTalk, Discord, E-Mail, Emby, Faast, FCM, Flock, Google Chat, -Gotify, Growl, Guilded, Home Assistant, IFTTT, Join, Kavenegar, KODI, Kumulos, -LaMetric, 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, PopcornNotify, -Prowl, Pushalot, PushBullet, Pushjet, PushMe, Pushover, PushSafer, Pushy, -PushDeer, Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, Signal, -SimplePush, Sinch, Slack, SMSEagle, SMTP2Go, Spontit, SparkPost, Super Toasty, -Streamlabs, Stride, Synology Chat, Syslog, Techulus Push, Telegram, Threema -Gateway, Twilio, Twitter, Twist, XBMC, Voipms, Vonage, WhatsApp, Webex Teams} +Apprise API, APRS, AWS SES, AWS SNS, Bark, Boxcar, Burst SMS, BulkSMS, +ClickSend, DAPNET, DingTalk, Discord, E-Mail, Emby, Faast, FCM, Flock, +Google Chat, Gotify, Growl, Guilded, Home Assistant, IFTTT, Join, Kavenegar, +KODI, Kumulos, LaMetric, 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, PopcornNotify, Prowl, Pushalot, PushBullet, Pushjet, PushMe, +Pushover, PushSafer, Pushy, PushDeer, Reddit, Rocket.Chat, RSyslog, SendGrid, +ServerChan, Signal, SimplePush, Sinch, Slack, SMSEagle, SMTP2Go, Spontit, +SparkPost, Super Toasty, Streamlabs, Stride, Synology Chat, Syslog, +Techulus Push, Telegram, Threema Gateway, Twilio, Twitter, Twist, XBMC, +Voipms, Vonage, WhatsApp, Webex Teams} Name: python-%{pypi_name} Version: 1.6.0 diff --git a/test/test_plugin_aprs.py b/test/test_plugin_aprs.py new file mode 100644 index 00000000..af48723e --- /dev/null +++ b/test/test_plugin_aprs.py @@ -0,0 +1,355 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2023, 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. + +from unittest import mock +import socket +import apprise +from apprise.plugins.NotifyAprs import NotifyAprs + +# Disable logging for a cleaner testing output +import logging + +logging.disable(logging.CRITICAL) + + +@mock.patch('socket.create_connection') +def test_plugin_aprs_urls(mock_create_connection): + """ + NotifyAprs() Apprise URLs + + """ + # A socket object + sobj = mock.Mock() + sobj.return_value = 1 + sobj.getpeername.return_value = ('localhost', 1234) + sobj.socket_close.return_value = None + sobj.setblocking.return_value = True + sobj.recv.return_value = \ + 'ping\npong pong DF1JSL-15 verified pong'.encode('latin-1') + sobj.sendall.return_value = True + sobj.settimeout.return_value = True + + # Prepare Mock + mock_create_connection.return_value = sobj + + # Test invalid URLs + assert apprise.Apprise.instantiate("aprs://") is None + assert apprise.Apprise.instantiate("aprs://:@/") is None + + # No call-sign specified + assert apprise.Apprise.instantiate("aprs://DF1JSL-15:12345") is None + + # Garbage + assert NotifyAprs.parse_url(None) is None + + # Valid call-sign but no password + assert apprise.Apprise.instantiate( + "aprs://DF1JSL-15:@DF1ABC") is None + assert apprise.Apprise.instantiate( + "aprs://DF1JSL-15@DF1ABC") is None + # Password of -1 not supported + assert apprise.Apprise.instantiate( + "aprs://DF1JSL-15:-1@DF1ABC") is None + # Alpha Password not supported + assert apprise.Apprise.instantiate( + "aprs://DF1JSL-15:abcd@DF1ABC") is None + + # Valid instances + instance = apprise.Apprise.instantiate( + "aprs://DF1JSL-15:12345@DF1ABC") + assert isinstance(instance, NotifyAprs) + assert instance.url(privacy=True).startswith( + 'aprs://DF1JSL-15:****@D...C?') + assert instance.notify('test') is True + + instance = apprise.Apprise.instantiate( + "aprs://DF1JSL-15:12345@DF1ABC/DF1DEF") + assert isinstance(instance, NotifyAprs) + assert instance.url(privacy=True).startswith( + 'aprs://DF1JSL-15:****@D...C/D...F?') + assert instance.notify('test') is True + + instance = apprise.Apprise.instantiate( + "aprs://DF1JSL-15:12345@DF1ABC-1/DF1ABC/DF1ABC-15") + assert isinstance(instance, NotifyAprs) + assert instance.url(privacy=True).startswith( + 'aprs://DF1JSL-15:****@D...1/D...C/D...5?') + assert instance.notify('test') is True + + instance = apprise.Apprise.instantiate( + "aprs://DF1JSL-15:12345@?to=DF1ABC,DF1DEF") + assert isinstance(instance, NotifyAprs) + assert instance.url(privacy=True).startswith( + 'aprs://DF1JSL-15:****@D...C/D...F?') + assert instance.notify('test') is True + + # Test Locale settings + instance = apprise.Apprise.instantiate( + "aprs://DF1JSL-15:12345@DF1ABC?locale=EURO") + assert isinstance(instance, NotifyAprs) + assert instance.url(privacy=True).startswith( + 'aprs://DF1JSL-15:****@D...C?') + # we used the default locale, so no setting + assert 'locale=' not in instance.url(privacy=True) + assert instance.notify('test') is True + + instance = apprise.Apprise.instantiate( + "aprs://DF1JSL-15:12345@DF1ABC?locale=NOAM") + assert isinstance(instance, NotifyAprs) + assert instance.url(privacy=True).startswith( + 'aprs://DF1JSL-15:****@D...C?') + # locale is set in URL + assert 'locale=NOAM' in instance.url(privacy=True) + assert instance.notify('test') is True + + # Invalid locale + assert apprise.Apprise.instantiate( + "aprs://DF1JSL-15:12345@DF1ABC?locale=invalid") is None + + # Invalid call signs + instance = apprise.Apprise.instantiate( + "aprs://DF1JSL-15:12345@abcdefghi/a") + + # We still instantiate + assert isinstance(instance, NotifyAprs) + + # We still load our bad entries + assert instance.url(privacy=True).startswith( + "aprs://DF1JSL-15:****@A...I/A...A?") + + # But with only bad entries, we have nothing to notify + assert instance.notify('test') is False + + # Enforces a close + del instance + + +@mock.patch('socket.create_connection') +def test_plugin_aprs_edge_cases(mock_create_connection): + """ + NotifyAprs() Edge Cases + """ + + # A socket object + sobj = mock.Mock() + sobj.return_value = 1 + sobj.getpeername.return_value = ('localhost', 1234) + sobj.socket_close.return_value = None + sobj.setblocking.return_value = True + sobj.recv.return_value = \ + 'ping\npong pong DF1JSL-15 verified pong'.encode('latin-1') + sobj.sendall.return_value = True + sobj.settimeout.return_value = True + + # Prepare Mock + mock_create_connection.return_value = sobj + + # Valid instances + instance = apprise.Apprise.instantiate( + "aprs://DF1JSL-15:12345@DF1ABC/DF1DEF") + assert isinstance(instance, NotifyAprs) + + # Objects read + assert len(instance) == 2 + + # Bad data + sobj.recv.return_value = 'one line'.encode('latin-1') + assert instance.notify(body='body', title='title') is False + sobj.recv.return_value = '\n\n\n'.encode('latin-1') + assert instance.notify(body='body', title='title') is False + sobj.recv.return_value = ''.encode('latin-1') + assert instance.notify(body='body', title='title') is False + sobj.recv.return_value = '\ndata'.encode('latin-1') + assert instance.notify(body='body', title='title') is False + # Different Call-Sign then what we logged in as + sobj.recv.return_value = \ + 'ping\npong pong DF1JSL-14 verified, pong'.encode('latin-1') + assert instance.notify(body='body', title='title') is False + # Unverified + sobj.recv.return_value = \ + 'ping\npong pong DF1JSL-15 unverified, pong'.encode('latin-1') + assert instance.notify(body='body', title='title') is False + + # + # Test Login edge cases + # + sobj.return_value = False + assert instance.aprsis_login() is False + sobj.return_value = 1 + sobj.recv.return_value = ''.encode('latin-1') + assert instance.aprsis_login() is False + sobj.recv.return_value = \ + 'ping\npong pong DF1JSL-15 verified pong'.encode('latin-1') + + # + # Test Socket Send Exceptions + # + sobj.sendall.return_value = None + sobj.sendall.side_effect = socket.gaierror('gaierror') + # No connection + assert instance.socket_send('data') is False + # Ensure we have a connection before calling socket_send() + assert instance.socket_open() is True + assert instance.socket_send('data') is False + sobj.sendall.side_effect = socket.timeout('timeout') + assert instance.socket_open() is True + assert instance.socket_send('data') is False + assert instance.socket_open() is True + sobj.sendall.side_effect = socket.error('error') + assert instance.socket_send('data') is False + + # Login is impacted by socket_send + sobj.return_value = 1 + assert instance.socket_open() is True + assert instance.aprsis_login() is False + + # Return some of our + sobj.sendall.side_effect = None + sobj.sendall.return_value = True + + assert instance.socket_open() is True + sobj.close.return_value = None + sobj.close.side_effect = socket.gaierror('gaierror') + instance.socket_close() + sobj.close.side_effect = socket.timeout('timeout') + instance.socket_close() + sobj.close.side_effect = socket.error('error') + instance.socket_close() + sobj.return_value = None + instance.socket_close() + # Socket isn't open; so we can't get content + assert instance.socket_receive(100) is False + sobj.close.side_effect = None + sobj.close.return_value = None + # Double close test + instance.socket_close() + + sobj.return_value = 1 + mock_create_connection.return_value = None + mock_create_connection.side_effect = socket.gaierror('gaierror') + assert instance.socket_open() is False + assert instance.notify('test') is False + mock_create_connection.side_effect = socket.timeout('timeout') + assert instance.socket_open() is False + assert instance.notify('test') is False + mock_create_connection.side_effect = socket.error('error') + assert instance.socket_open() is False + assert instance.notify('test') is False + mock_create_connection.side_effect = ConnectionError('ConnectionError') + assert instance.socket_open() is False + assert instance.notify('test') is False + + # Restore our good connection + mock_create_connection.return_value = sobj + mock_create_connection.side_effect = None + + # Functionality has been restored + assert instance.socket_open() is True + + # Now play with getpeername + sobj.getpeername.return_value = None + sobj.getpeername.side_effect = ValueError('getpeername ValueError') + assert instance.socket_open() is True + + sobj.getpeername.return_value = ('localhost', 1234) + assert instance.socket_open() is True + # Test different receive settings + assert instance.socket_receive(0) + assert instance.socket_receive(-1) + assert instance.socket_receive(100) + + sobj.recv.side_effect = socket.gaierror('gaierror') + assert instance.socket_open() is True + assert instance.socket_receive(100) is False + sobj.recv.side_effect = socket.timeout('timeout') + assert instance.socket_open() is True + assert instance.socket_receive(100) is False + sobj.recv.side_effect = socket.error('error') + assert instance.socket_open() is True + assert instance.socket_receive(100) is False + + # Restore + sobj.recv.side_effect = None + sobj.recv.return_value = \ + 'ping\npong pong DF1JSL-15 verified pong'.encode('latin-1') + + # Simulate a successful connection, but a failed notification + # To do this we need to have a login succeed, but the second call to send + # to fail + sobj.sendall.return_value = True + assert instance.notify('test') is True + + sobj.sendall.return_value = None + sobj.sendall.side_effect = (True, socket.gaierror('gaierror')) + assert instance.notify('test') is False + + sobj.sendall.return_value = True + sobj.sendall.side_effect = None + del sobj + + +def test_plugin_aprs_config_files(): + """ + NotifyAprs() Config File Cases + """ + content = """ + urls: + - aprs://DF1JSL-15:12345@DF1ABC": + - locale: NOAM + + - aprs://DF1JSL-15:12345@DF1ABC: + - locale: SOAM + + - aprs://DF1JSL-15:12345@DF1ABC: + - locale: EURO + + - aprs://DF1JSL-15:12345@DF1ABC: + - locale: ASIA + + - aprs://DF1JSL-15:12345@DF1ABC: + - locale: AUNZ + + - aprs://DF1JSL-15:12345@DF1ABC: + - locale: ROTA + + # This will fail to load because the locale is bad + - aprs://DF1JSL-15:12345@DF1ABC: + - locale: aprs_invalid + """ + + # Create ourselves a config object + ac = apprise.AppriseConfig() + assert ac.add_config(content=content) is True + + aobj = apprise.Apprise() + + # Add our configuration + aobj.add(ac) + + assert len(ac.servers()) == 6 + assert len(aobj) == 6