From 51f91972c26f231d9d995d38725d74b7fa8c33e4 Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sat, 15 May 2021 20:29:37 -0400 Subject: [PATCH] Added SMTP2Go (with Attachment) Support (#389) --- README.md | 1 + apprise/plugins/NotifySMTP2Go.py | 584 +++++++++++++++++++++++++++ packaging/redhat/python-apprise.spec | 5 +- setup.py | 6 +- test/test_rest_plugins.py | 103 +++++ test/test_smtp2go.py | 134 ++++++ 6 files changed, 828 insertions(+), 5 deletions(-) create mode 100644 apprise/plugins/NotifySMTP2Go.py create mode 100644 test/test_smtp2go.py diff --git a/README.md b/README.md index fff15e8642..23cdaf1c11 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,7 @@ The table below identifies the services this tool supports and some example serv | [SendGrid](https://github.com/caronc/apprise/wiki/Notify_sendgrid) | sendgrid:// | (TCP) 443 | sendgrid://APIToken:FromEmail/
sendgrid://APIToken:FromEmail/ToEmail
sendgrid://APIToken:FromEmail/ToEmail1/ToEmail2/ToEmailN/ | [SimplePush](https://github.com/caronc/apprise/wiki/Notify_simplepush) | spush:// | (TCP) 443 | spush://apikey
spush://salt:password@apikey
spush://apikey?event=Apprise | [Slack](https://github.com/caronc/apprise/wiki/Notify_slack) | slack:// | (TCP) 443 | slack://TokenA/TokenB/TokenC/
slack://TokenA/TokenB/TokenC/Channel
slack://botname@TokenA/TokenB/TokenC/Channel
slack://user@TokenA/TokenB/TokenC/Channel1/Channel2/ChannelN +| [SMTP2Go](https://github.com/caronc/apprise/wiki/Notify_smtp2go) | smtp2go:// | (TCP) 443 | smtp2go://user@hostname/apikey
smtp2go://user@hostname/apikey/email
smtp2go://user@hostname/apikey/email1/email2/emailN
smtp2go://user@hostname/apikey/?name="From%20User" | [SparkPost](https://github.com/caronc/apprise/wiki/Notify_sparkpost) | sparkpost:// | (TCP) 443 | sparkpost://user@hostname/apikey
sparkpost://user@hostname/apikey/email
sparkpost://user@hostname/apikey/email1/email2/emailN
sparkpost://user@hostname/apikey/?name="From%20User" | [Spontit](https://github.com/caronc/apprise/wiki/Notify_spontit) | spontit:// | (TCP) 443 | spontit://UserID@APIKey/
spontit://UserID@APIKey/Channel
spontit://UserID@APIKey/Channel1/Channel2/ChannelN | [Syslog](https://github.com/caronc/apprise/wiki/Notify_syslog) | syslog:// | n/a | syslog://
syslog://Facility diff --git a/apprise/plugins/NotifySMTP2Go.py b/apprise/plugins/NotifySMTP2Go.py new file mode 100644 index 0000000000..341ad8a621 --- /dev/null +++ b/apprise/plugins/NotifySMTP2Go.py @@ -0,0 +1,584 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2021 Chris Caron +# All rights reserved. +# +# This code is licensed under the MIT License. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files(the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions : +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. + +# Signup @ https://smtp2go.com (free accounts available) +# +# From your dashboard, you can generate an API Key if you haven't already +# at https://app.smtp2go.com/settings/apikeys/ + +# The API Key from here which will look something like: +# api-60F0DD0AB5BA11ABA421F23C91C88EF4 +# +# Knowing this, you can buid your smtp2go url as follows: +# smtp2go://{user}@{domain}/{apikey} +# smtp2go://{user}@{domain}/{apikey}/{email} +# +# You can email as many addresses as you want as: +# smtp2go://{user}@{domain}/{apikey}/{email1}/{email2}/{emailN} +# +# The {user}@{domain} effectively assembles the 'from' email address +# the email will be transmitted from. If no email address is specified +# then it will also become the 'to' address as well. +# +import base64 +import requests +from json import dumps +from email.utils import formataddr +from .NotifyBase import NotifyBase +from ..common import NotifyType +from ..common import NotifyFormat +from ..utils import parse_emails +from ..utils import parse_bool +from ..utils import is_email +from ..utils import validate_regex +from ..AppriseLocale import gettext_lazy as _ + +SMTP2GO_HTTP_ERROR_MAP = { + 429: 'To many requests.', +} + + +class NotifySMTP2Go(NotifyBase): + """ + A wrapper for SMTP2Go Notifications + """ + + # The default descriptive name associated with the Notification + service_name = 'SMTP2Go' + + # The services URL + service_url = 'https://www.smtp2go.com/' + + # All notification requests are secure + secure_protocol = 'smtp2go' + + # SMTP2Go advertises they allow 300 requests per minute. + # 60/300 = 0.2 + request_rate_per_sec = 0.20 + + # A URL that takes you to the setup/help of the specific protocol + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_smtp2go' + + # Notify URL + notify_url = 'https://api.smtp2go.com/v3/email/send' + + # Default Notify Format + notify_format = NotifyFormat.HTML + + # The maximum amount of emails that can reside within a single + # batch transfer + default_batch_size = 100 + + # Define object templates + templates = ( + '{schema}://{user}@{host}:{apikey}/', + '{schema}://{user}@{host}:{apikey}/{targets}', + ) + + # Define our template tokens + template_tokens = dict(NotifyBase.template_tokens, **{ + 'user': { + 'name': _('User Name'), + 'type': 'string', + 'required': True, + }, + 'host': { + 'name': _('Domain'), + 'type': 'string', + 'required': True, + }, + 'apikey': { + 'name': _('API Key'), + 'type': 'string', + 'private': True, + 'required': True, + }, + 'targets': { + 'name': _('Target Emails'), + 'type': 'list:string', + }, + }) + + # Define our template arguments + template_args = dict(NotifyBase.template_args, **{ + 'name': { + 'name': _('From Name'), + 'type': 'string', + 'map_to': 'from_name', + }, + 'to': { + 'alias_of': 'targets', + }, + 'cc': { + 'name': _('Carbon Copy'), + 'type': 'list:string', + }, + 'bcc': { + 'name': _('Blind Carbon Copy'), + 'type': 'list:string', + }, + 'batch': { + 'name': _('Batch Mode'), + 'type': 'bool', + 'default': False, + }, + }) + + # Define any kwargs we're using + template_kwargs = { + 'headers': { + 'name': _('Email Header'), + 'prefix': '+', + }, + } + + def __init__(self, apikey, targets, cc=None, bcc=None, from_name=None, + headers=None, batch=False, **kwargs): + """ + Initialize SMTP2Go Object + """ + super(NotifySMTP2Go, self).__init__(**kwargs) + + # API Key (associated with project) + self.apikey = validate_regex(apikey) + if not self.apikey: + msg = 'An invalid SMTP2Go API Key ' \ + '({}) was specified.'.format(apikey) + self.logger.warning(msg) + raise TypeError(msg) + + # Validate our username + if not self.user: + msg = 'No SMTP2Go username was specified.' + self.logger.warning(msg) + raise TypeError(msg) + + # Acquire Email 'To' + self.targets = list() + + # Acquire Carbon Copies + self.cc = set() + + # Acquire Blind Carbon Copies + self.bcc = set() + + # For tracking our email -> name lookups + self.names = {} + + self.headers = {} + if headers: + # Store our extra headers + self.headers.update(headers) + + # Prepare Batch Mode Flag + self.batch = batch + + # Get our From username (if specified) + self.from_name = from_name + + # Get our from email address + self.from_addr = '{user}@{host}'.format(user=self.user, host=self.host) + + if not is_email(self.from_addr): + # Parse Source domain based on from_addr + msg = 'Invalid ~From~ email format: {}'.format(self.from_addr) + self.logger.warning(msg) + raise TypeError(msg) + + if targets: + # Validate recipients (to:) and drop bad ones: + for recipient in parse_emails(targets): + result = is_email(recipient) + if result: + self.targets.append( + (result['name'] if result['name'] else False, + result['full_email'])) + continue + + self.logger.warning( + 'Dropped invalid To email ' + '({}) specified.'.format(recipient), + ) + + else: + # If our target email list is empty we want to add ourselves to it + self.targets.append( + (self.from_name if self.from_name else False, self.from_addr)) + + # Validate recipients (cc:) and drop bad ones: + for recipient in parse_emails(cc): + email = is_email(recipient) + if email: + self.cc.add(email['full_email']) + + # Index our name (if one exists) + self.names[email['full_email']] = \ + email['name'] if email['name'] else False + 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): + email = is_email(recipient) + if email: + self.bcc.add(email['full_email']) + + # Index our name (if one exists) + self.names[email['full_email']] = \ + email['name'] if email['name'] else False + continue + + self.logger.warning( + 'Dropped invalid Blind Carbon Copy email ' + '({}) specified.'.format(recipient), + ) + + def send(self, body, title='', notify_type=NotifyType.INFO, attach=None, + **kwargs): + """ + Perform SMTP2Go Notification + """ + + if not self.targets: + # There is no one to email; we're done + self.logger.warning( + 'There are no Email recipients to notify') + return False + + # error tracking (used for function return) + has_error = False + + # Send in batches if identified to do so + batch_size = 1 if not self.batch else self.default_batch_size + + # Prepare our headers + headers = { + 'User-Agent': self.app_id, + 'Accept': 'application/json', + 'Content-Type': 'application/json', + } + + # Track our potential attachments + attachments = [] + + if attach: + for idx, attachment in enumerate(attach): + # Perform some simple error checking + if not attachment: + # We could not access the attachment + self.logger.error( + 'Could not access attachment {}.'.format( + attachment.url(privacy=True))) + return False + + try: + with open(attachment.path, 'rb') as f: + # Output must be in a DataURL format (that's what + # PushSafer calls it): + attachments.append({ + 'filename': attachment.name, + 'fileblob': base64.b64encode(f.read()) + .decode('utf-8'), + 'mimetype': attachment.mimetype, + }) + + except (OSError, IOError) as e: + self.logger.warning( + 'An I/O error occurred while reading {}.'.format( + attachment.name if attachment else 'attachment')) + self.logger.debug('I/O Exception: %s' % str(e)) + return False + + try: + sender = formataddr( + (self.from_name if self.from_name else False, + self.from_addr), charset='utf-8') + + except TypeError: + # Python v2.x Support (no charset keyword) + # Format our cc addresses to support the Name field + sender = formataddr( + (self.from_name if self.from_name else False, + self.from_addr)) + + # Prepare our payload + payload = { + # API Key + 'api_key': self.apikey, + + # Base payload options + 'sender': sender, + 'subject': title, + + # our To array + 'to': [], + } + + if attachments: + payload['attachments'] = attachments + + if self.notify_format == NotifyFormat.HTML: + payload['html_body'] = body + + else: + payload['text_body'] = body + + # Create a copy of the targets list + emails = list(self.targets) + + for index in range(0, len(emails), batch_size): + # Initialize our cc list + cc = (self.cc - self.bcc) + + # Initialize our bcc list + bcc = set(self.bcc) + + # Initialize our to list + to = list() + + for to_addr in self.targets[index:index + batch_size]: + # Strip target out of cc list if in To + cc = (cc - set([to_addr[1]])) + + # Strip target out of bcc list if in To + bcc = (bcc - set([to_addr[1]])) + + try: + # Prepare our to + to.append(formataddr(to_addr, charset='utf-8')) + + except TypeError: + # Python v2.x Support (no charset keyword) + # Format our cc addresses to support the Name field + + # Prepare our to + to.append(formataddr(to_addr)) + + # Prepare our To + payload['to'] = to + + if cc: + try: + # Format our cc addresses to support the Name field + payload['cc'] = [formataddr( + (self.names.get(addr, False), addr), charset='utf-8') + for addr in cc] + + except TypeError: + # Python v2.x Support (no charset keyword) + # Format our cc addresses to support the Name field + payload['cc'] = [formataddr( # pragma: no branch + (self.names.get(addr, False), addr)) + for addr in cc] + + # Format our bcc addresses to support the Name field + if bcc: + # set our bcc variable (convert to list first so it's + # JSON serializable) + payload['bcc'] = list(bcc) + + # Store our header entries if defined into the payload + # in their payload + if self.headers: + payload['custom_headers'] = \ + [{'header': k, 'value': v} + for k, v in self.headers.items()] + + # Some Debug Logging + self.logger.debug('SMTP2Go POST URL: {} (cert_verify={})'.format( + self.notify_url, self.verify_certificate)) + self.logger.debug('SMTP2Go Payload: {}' .format(payload)) + + # For logging output of success and errors; we get a head count + # of our outbound details: + verbose_dest = ', '.join( + [x[1] for x in self.targets[index:index + batch_size]]) \ + if len(self.targets[index:index + batch_size]) <= 3 \ + else '{} recipients'.format( + len(self.targets[index:index + batch_size])) + + # Always call throttle before any remote server i/o is made + self.throttle() + try: + r = requests.post( + self.notify_url, + data=dumps(payload), + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + + if r.status_code != requests.codes.ok: + # We had a problem + status_str = \ + NotifyBase.http_response_code_lookup( + r.status_code, SMTP2GO_HTTP_ERROR_MAP) + + self.logger.warning( + 'Failed to send SMTP2Go notification to {}: ' + '{}{}error={}.'.format( + verbose_dest, + status_str, + ', ' if status_str else '', + r.status_code)) + + self.logger.debug( + 'Response Details:\r\n{}'.format(r.content)) + + # Mark our failure + has_error = True + continue + + else: + self.logger.info( + 'Sent SMTP2Go notification to {}.'.format( + verbose_dest)) + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occurred sending SMTP2Go:%s ' % ( + verbose_dest) + 'notification.' + ) + self.logger.debug('Socket Exception: %s' % str(e)) + + # Mark our failure + has_error = True + continue + + except (OSError, IOError) as e: + self.logger.warning( + 'An I/O error occurred while reading attachments') + self.logger.debug('I/O Exception: %s' % str(e)) + + # Mark our failure + has_error = True + continue + + 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 = { + 'batch': 'yes' if self.batch else 'no', + } + + # Append our headers into our parameters + params.update({'+{}'.format(k): v for k, v in self.headers.items()}) + + # Extend our parameters + params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) + + if self.from_name is not None: + # from_name specified; pass it back on the url + params['name'] = self.from_name + + if self.cc: + # Handle our Carbon Copy Addresses + params['cc'] = ','.join( + ['{}{}'.format( + '' if not e not in self.names + else '{}:'.format(self.names[e]), e) for e in self.cc]) + + if self.bcc: + # Handle our Blind Carbon Copy Addresses + params['bcc'] = ','.join(self.bcc) + + # a simple boolean check as to whether we display our target emails + # or not + has_targets = \ + not (len(self.targets) == 1 + and self.targets[0][1] == self.from_addr) + + return '{schema}://{user}@{host}/{apikey}/{targets}?{params}'.format( + schema=self.secure_protocol, + host=self.host, + user=NotifySMTP2Go.quote(self.user, safe=''), + apikey=self.pprint(self.apikey, privacy, safe=''), + targets='' if not has_targets else '/'.join( + [NotifySMTP2Go.quote('{}{}'.format( + '' if not e[0] else '{}:'.format(e[0]), e[1]), + safe='') for e in self.targets]), + params=NotifySMTP2Go.urlencode(params)) + + @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 + + # Get our entries; split_path() looks after unquoting content for us + # by default + results['targets'] = NotifySMTP2Go.split_path(results['fullpath']) + + # Our very first entry is reserved for our api key + try: + results['apikey'] = results['targets'].pop(0) + + except IndexError: + # We're done - no API Key found + results['apikey'] = None + + if 'name' in results['qsd'] and len(results['qsd']['name']): + # Extract from name to associate with from address + results['from_name'] = \ + NotifySMTP2Go.unquote(results['qsd']['name']) + + # Handle 'to' email address + if 'to' in results['qsd'] and len(results['qsd']['to']): + results['targets'].append(results['qsd']['to']) + + # Handle Carbon Copy Addresses + if 'cc' in results['qsd'] and len(results['qsd']['cc']): + results['cc'] = results['qsd']['cc'] + + # Handle Blind Carbon Copy Addresses + if 'bcc' in results['qsd'] and len(results['qsd']['bcc']): + results['bcc'] = results['qsd']['bcc'] + + # Add our Meta Headers that the user can provide with their outbound + # emails + results['headers'] = {NotifyBase.unquote(x): NotifyBase.unquote(y) + for x, y in results['qsd+'].items()} + + # Get Batch Mode Flag + results['batch'] = \ + parse_bool(results['qsd'].get( + 'batch', NotifySMTP2Go.template_args['batch']['default'])) + + return results diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 0187abcdf0..e8fb09c54c 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -53,8 +53,9 @@ LaMetric, MacOSX, Mailgun, Mattermost, Matrix, Microsoft Windows, Microsoft Teams, MessageBird, MSG91, MyAndroid, Nexmo, Nextcloud, Notica, Notifico, Office365, OneSignal, Opsgenie, ParsePlatform, PopcornNotify, Prowl, Pushalot, PushBullet, Pushjet, Pushover, PushSafer, Reddit, Rocket.Chat, SendGrid, -SimplePush, Sinch, Slack, Spontit, SparkPost, Super Toasty, Stride, Syslog, -Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC, XMPP, Webex Teams} +SimplePush, Sinch, Slack, SMTP2Go, Spontit, SparkPost, Super Toasty, Stride, +Syslog, Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC, XMPP, +Webex Teams} Name: python-%{pypi_name} Version: 0.9.2 diff --git a/setup.py b/setup.py index 8707a23781..2a34fc56b0 100755 --- a/setup.py +++ b/setup.py @@ -75,9 +75,9 @@ 'MacOS Mailgun Matrix Mattermost MessageBird MSG91 Nexmo Nextcloud ' 'Notica Notifico Office365 OneSignal Opsgenie ParsePlatform ' 'PopcornNotify Prowl PushBullet Pushjet Pushed Pushover PushSafer ' - 'Reddit Rocket.Chat Ryver SendGrid SimplePush Sinch Slack SparkPost ' - 'Spontit Stride Syslog Techulus Telegram Twilio Twist Twitter XBMC ' - 'MSTeams Microsoft Windows Webex CLI API', + 'Reddit Rocket.Chat Ryver SendGrid SimplePush Sinch Slack SMTP2Go ' + 'SparkPost Spontit Stride Syslog Techulus Telegram Twilio Twist ' + 'Twitter XBMC MSTeams Microsoft Windows Webex CLI API', author='Chris Caron', author_email='lead2gold@gmail.com', packages=find_packages(), diff --git a/test/test_rest_plugins.py b/test/test_rest_plugins.py index c81bb1ba4a..6dcf5e2f81 100644 --- a/test/test_rest_plugins.py +++ b/test/test_rest_plugins.py @@ -4628,6 +4628,109 @@ 'requests_response_text': 'ok', }), + ################################## + # NotifySMTP2Go + ################################## + ('smtp2go://', { + 'instance': TypeError, + }), + ('smtp2go://:@/', { + 'instance': TypeError, + }), + # No Token specified + ('smtp2go://user@localhost.localdomain', { + 'instance': TypeError, + }), + # Token is valid, but no user name specified + ('smtp2go://localhost.localdomain/{}-{}-{}'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': TypeError, + }), + # Invalid from email address + ('smtp2go://!@localhost.localdomain/{}-{}-{}'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': TypeError, + }), + # No To email address, but everything else is valid + ('smtp2go://user@localhost.localdomain/{}-{}-{}'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + }), + ('smtp2go://user@localhost.localdomain/{}-{}-{}?format=markdown'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + }), + ('smtp2go://user@localhost.localdomain/{}-{}-{}?format=html'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + }), + ('smtp2go://user@localhost.localdomain/{}-{}-{}?format=text'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + }), + # headers + ('smtp2go://user@localhost.localdomain/{}-{}-{}' + '?+X-Customer-Campaign-ID=Apprise'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + }), + # bcc and cc + ('smtp2go://user@localhost.localdomain/{}-{}-{}' + '?bcc=user@example.com&cc=user2@example.com'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + }), + # One To Email address + ('smtp2go://user@localhost.localdomain/{}-{}-{}/test@example.com'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + }), + ('smtp2go://user@localhost.localdomain/' + '{}-{}-{}?to=test@example.com'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go}), + # One To Email address, a from name specified too + ('smtp2go://user@localhost.localdomain/{}-{}-{}/' + 'test@example.com?name="Frodo"'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go}), + # Invalid 'To' Email address + ('smtp2go://user@localhost.localdomain/{}-{}-{}/invalid'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + # Expected notify() response + 'notify_response': False, + }), + # Multiple 'To', 'Cc', and 'Bcc' addresses (with invalid ones) + ('smtp2go://user@example.com/{}-{}-{}/{}?bcc={}&cc={}'.format( + 'a' * 32, 'b' * 8, 'c' * 8, + '/'.join(('user1@example.com', 'invalid', 'User2:user2@example.com')), + ','.join(('user3@example.com', 'i@v', 'User1:user1@example.com')), + ','.join(('user4@example.com', 'g@r@b', 'Da:user5@example.com'))), { + 'instance': plugins.NotifySMTP2Go, + }), + ('smtp2go://user@localhost.localdomain/{}-{}-{}'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + # force a failure + 'response': False, + 'requests_response_code': requests.codes.internal_server_error, + }), + ('smtp2go://user@localhost.localdomain/{}-{}-{}'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + # throw a bizzare code forcing us to fail to look it up + 'response': False, + 'requests_response_code': 999, + }), + ('smtp2go://user@localhost.localdomain/{}-{}-{}'.format( + 'a' * 32, 'b' * 8, 'c' * 8), { + 'instance': plugins.NotifySMTP2Go, + # Throws a series of connection and transfer exceptions when this flag + # is set and tests that we gracfully handle them + 'test_requests_exceptions': True, + }), + ################################## # NotifySNS (AWS) ################################## diff --git a/test/test_smtp2go.py b/test/test_smtp2go.py new file mode 100644 index 0000000000..064053bccb --- /dev/null +++ b/test/test_smtp2go.py @@ -0,0 +1,134 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2021 Chris Caron +# All rights reserved. +# +# This code is licensed under the MIT License. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files(the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions : +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. + +import os +import sys +import mock +import requests +from apprise import plugins +from apprise import Apprise +from apprise import AppriseAttachment +from apprise import NotifyType + +# Disable logging for a cleaner testing output +import logging +logging.disable(logging.CRITICAL) + +# Attachment Directory +TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var') + + +@mock.patch('requests.post') +def test_notify_smtp2go_plugin_attachments(mock_post): + """ + API: NotifySMTP2Go() Attachments + + """ + # Disable Throttling to speed testing + plugins.NotifyBase.request_rate_per_sec = 0 + + okay_response = requests.Request() + okay_response.status_code = requests.codes.ok + okay_response.content = "" + + # Assign our mock object our return value + mock_post.return_value = okay_response + + # API Key + apikey = 'abc123' + + obj = Apprise.instantiate( + 'smtp2go://user@localhost.localdomain/{}'.format(apikey)) + assert isinstance(obj, plugins.NotifySMTP2Go) + + # Test Valid Attachment + path = os.path.join(TEST_VAR_DIR, 'apprise-test.gif') + attach = AppriseAttachment(path) + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO, + attach=attach) is True + + # Test invalid attachment + path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg') + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO, + attach=path) is False + + mock_post.return_value = None + mock_post.side_effect = OSError() + # We can't send the message if we can't read the attachment + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO, + attach=attach) is False + + # Get a appropriate "builtin" module name for pythons 2/3. + if sys.version_info.major >= 3: + builtin_open_function = 'builtins.open' + + else: + builtin_open_function = '__builtin__.open' + + # Test Valid Attachment (load 3) + path = ( + os.path.join(TEST_VAR_DIR, 'apprise-test.gif'), + os.path.join(TEST_VAR_DIR, 'apprise-test.gif'), + os.path.join(TEST_VAR_DIR, 'apprise-test.gif'), + ) + attach = AppriseAttachment(path) + + # Return our good configuration + mock_post.side_effect = None + mock_post.return_value = okay_response + with mock.patch(builtin_open_function, side_effect=OSError()): + # We can't send the message we can't open the attachment for reading + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO, + attach=attach) is False + + # test the handling of our batch modes + obj = Apprise.instantiate( + 'smtp2go://no-reply@example.com/{}/' + 'user1@example.com/user2@example.com?batch=yes'.format(apikey)) + assert isinstance(obj, plugins.NotifySMTP2Go) + + # Force our batch to break into separate messages + obj.default_batch_size = 1 + # We'll send 2 messages + mock_post.reset_mock() + + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO, + attach=attach) is True + assert mock_post.call_count == 2 + + # single batch + mock_post.reset_mock() + # We'll send 1 message + obj.default_batch_size = 2 + + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO, + attach=attach) is True + assert mock_post.call_count == 1