mirror of
https://github.com/caronc/apprise.git
synced 2024-11-24 17:14:00 +01:00
Added Twist Support (#137)
This commit is contained in:
Chris Caron
GitHub
parent
3124ef26a1
commit
2609680792
@ -59,6 +59,7 @@ The table below identifies the services this tool supports and some example serv
|
||||
| [Slack](https://github.com/caronc/apprise/wiki/Notify_slack) | slack:// | (TCP) 443 | slack://TokenA/TokenB/TokenC/<br />slack://TokenA/TokenB/TokenC/Channel<br />slack://botname@TokenA/TokenB/TokenC/Channel<br />slack://user@TokenA/TokenB/TokenC/Channel1/Channel2/ChannelN
|
||||
| [Telegram](https://github.com/caronc/apprise/wiki/Notify_telegram) | tgram:// | (TCP) 443 | tgram://bottoken/ChatID<br />tgram://bottoken/ChatID1/ChatID2/ChatIDN
|
||||
| [Twitter](https://github.com/caronc/apprise/wiki/Notify_twitter) | twitter:// | (TCP) 443 | twitter://CKey/CSecret/AKey/ASecret<br/>twitter://user@CKey/CSecret/AKey/ASecret<br/>twitter://CKey/CSecret/AKey/ASecret/User1/User2/User2<br/>twitter://CKey/CSecret/AKey/ASecret?mode=tweet
|
||||
| [Twist](https://github.com/caronc/apprise/wiki/Notify_twist) | twist:// | (TCP) 443 | twist://pasword:login<br/>twist://password:login/#channel<br/>twist://password:login/#team:channel<br/>twist://password:login/#team:channel1/channel2/#team3:channel
|
||||
| [XBMC](https://github.com/caronc/apprise/wiki/Notify_xbmc) | xbmc:// or xbmcs:// | (TCP) 8080 or 443 | xbmc://hostname<br />xbmc://user@hostname<br />xbmc://user:password@hostname:port
|
||||
| [XMPP](https://github.com/caronc/apprise/wiki/Notify_xmpp) | xmpp:// or xmpps:// | (TCP) 5222 or 5223 | xmpp://password@hostname<br />xmpp://user:password@hostname<br />xmpps://user:password@hostname:port?jid=user@hostname/resource<br/>xmpps://password@hostname/target@myhost, target2@myhost/resource
|
||||
| [Windows Notification](https://github.com/caronc/apprise/wiki/Notify_windows) | windows:// | n/a | windows://
|
||||
|
||||
805
apprise/plugins/NotifyTwist.py
Normal file
805
apprise/plugins/NotifyTwist.py
Normal file
@ -0,0 +1,805 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
|
||||
# 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.
|
||||
#
|
||||
# All of the documentation needed to work with the Twist API can be found
|
||||
# here: https://developer.twist.com/v3/
|
||||
|
||||
import re
|
||||
import requests
|
||||
from json import loads
|
||||
from itertools import chain
|
||||
|
||||
from .NotifyBase import NotifyBase
|
||||
from ..common import NotifyFormat
|
||||
from ..common import NotifyType
|
||||
from ..utils import parse_list
|
||||
from ..utils import GET_EMAIL_RE
|
||||
from ..AppriseLocale import gettext_lazy as _
|
||||
|
||||
|
||||
# A workspace can also be interpreted as a team name too!
|
||||
IS_CHANNEL = re.compile(
|
||||
r'^#?(?P<name>((?P<workspace>[A-Za-z0-9_-]+):)?'
|
||||
r'(?P<channel>[^\s]{1,64}))$')
|
||||
|
||||
IS_CHANNEL_ID = re.compile(
|
||||
r'^(?P<name>((?P<workspace>[0-9]+):)?(?P<channel>[0-9]+))$')
|
||||
|
||||
# Used to break apart list of potential tags by their delimiter
|
||||
# into a usable list.
|
||||
LIST_DELIM = re.compile(r'[ \t\r\n,\\/]+')
|
||||
|
||||
|
||||
class NotifyTwist(NotifyBase):
|
||||
"""
|
||||
A wrapper for Notify Twist Notifications
|
||||
"""
|
||||
|
||||
# The default descriptive name associated with the Notification
|
||||
service_name = 'Twist'
|
||||
|
||||
# The services URL
|
||||
service_url = 'https://twist.com'
|
||||
|
||||
# The default secure protocol
|
||||
secure_protocol = 'twist'
|
||||
|
||||
# A URL that takes you to the setup/help of the specific protocol
|
||||
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_twist'
|
||||
|
||||
# The maximum size of the message
|
||||
body_maxlen = 1000
|
||||
|
||||
# Default to markdown
|
||||
notify_format = NotifyFormat.MARKDOWN
|
||||
|
||||
# The default Notification URL to use
|
||||
api_url = 'https://api.twist.com/api/v3/'
|
||||
|
||||
# Allow 300 requests per minute.
|
||||
# 60/300 = 0.2
|
||||
request_rate_per_sec = 0.2
|
||||
|
||||
# The default channel to notify if no targets are specified
|
||||
default_notification_channel = 'general'
|
||||
|
||||
# Define object templates
|
||||
templates = (
|
||||
'{schema}://{password}:{email}',
|
||||
'{schema}://{password}:{email}/{targets}',
|
||||
)
|
||||
|
||||
# Define our template arguments
|
||||
template_tokens = dict(NotifyBase.template_tokens, **{
|
||||
'password': {
|
||||
'name': _('Password'),
|
||||
'type': 'string',
|
||||
'private': True,
|
||||
},
|
||||
'email': {
|
||||
'name': _('Email'),
|
||||
'type': 'string',
|
||||
},
|
||||
'target_channel': {
|
||||
'name': _('Target Channel'),
|
||||
'type': 'string',
|
||||
'prefix': '#',
|
||||
'map_to': 'targets',
|
||||
},
|
||||
'target_channel_id': {
|
||||
'name': _('Target Channel ID'),
|
||||
'type': 'string',
|
||||
'map_to': 'targets',
|
||||
},
|
||||
'targets': {
|
||||
'name': _('Targets'),
|
||||
'type': 'list:string',
|
||||
},
|
||||
})
|
||||
|
||||
# Define our template arguments
|
||||
template_args = dict(NotifyBase.template_args, **{
|
||||
'to': {
|
||||
'alias_of': 'targets',
|
||||
},
|
||||
})
|
||||
|
||||
def __init__(self, email=None, targets=None, **kwargs):
|
||||
"""
|
||||
Initialize Notify Twist Object
|
||||
"""
|
||||
super(NotifyTwist, self).__init__(**kwargs)
|
||||
|
||||
# Initialize channels list
|
||||
self.channels = set()
|
||||
|
||||
# Initialize Channel ID which are stored as:
|
||||
# <workspace_id>:<channel_id>
|
||||
self.channel_ids = set()
|
||||
|
||||
# Initialize our Email Object
|
||||
self.email = email if email else '{}@{}'.format(
|
||||
self.user,
|
||||
self.host,
|
||||
)
|
||||
|
||||
# The token is None if we're not logged in and False if we
|
||||
# failed to log in. Otherwise it is set to the actual token
|
||||
self.token = None
|
||||
|
||||
# Our default workspace (associated with our token)
|
||||
self.default_workspace = None
|
||||
|
||||
# A set of all of the available workspaces
|
||||
self._cached_workspaces = set()
|
||||
|
||||
# A mapping of channel names, the layout is as follows:
|
||||
# {
|
||||
# <workspace_id>: {
|
||||
# <channel_name>: <channel_id>,
|
||||
# <channel_name>: <channel_id>,
|
||||
# ...
|
||||
# },
|
||||
# <workspace2_id>: {
|
||||
# <channel_name>: <channel_id>,
|
||||
# <channel_name>: <channel_id>,
|
||||
# ...
|
||||
# },
|
||||
# }
|
||||
self._cached_channels = dict()
|
||||
|
||||
try:
|
||||
result = GET_EMAIL_RE.match(self.email)
|
||||
if not result:
|
||||
# let outer exception handle this
|
||||
raise TypeError
|
||||
|
||||
if email:
|
||||
# Force user/host to be that of the defined email for
|
||||
# consistency. This is very important for those initializing
|
||||
# this object with the the email object would could potentially
|
||||
# cause inconsistency to contents in the NotifyBase() object
|
||||
self.user = result.group('fulluser')
|
||||
self.host = result.group('domain')
|
||||
|
||||
except (TypeError, AttributeError):
|
||||
msg = 'The Twist Auth email specified ({}) is invalid.'\
|
||||
.format(self.email)
|
||||
self.logger.warning(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
if not self.password:
|
||||
msg = 'No Twist password was specified with account: {}'\
|
||||
.format(self.email)
|
||||
self.logger.warning(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
# Validate recipients and drop bad ones:
|
||||
for recipient in parse_list(targets):
|
||||
result = IS_CHANNEL_ID.match(recipient)
|
||||
if result:
|
||||
# store valid channel id
|
||||
self.channel_ids.add(result.group('name'))
|
||||
continue
|
||||
|
||||
result = IS_CHANNEL.match(recipient)
|
||||
if result:
|
||||
# store valid device
|
||||
self.channels.add(result.group('name').lower())
|
||||
continue
|
||||
|
||||
self.logger.warning(
|
||||
'Dropped invalid channel/id '
|
||||
'({}) specified.'.format(recipient),
|
||||
)
|
||||
|
||||
if len(self.channels) + len(self.channel_ids) == 0:
|
||||
# Notify our default channel
|
||||
self.channels.add(self.default_notification_channel)
|
||||
self.logger.warning(
|
||||
'Added default notification channel {}'.format(
|
||||
self.default_notification_channel))
|
||||
return
|
||||
|
||||
def url(self):
|
||||
"""
|
||||
Returns the URL built dynamically based on specified arguments.
|
||||
"""
|
||||
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
'verify': 'yes' if self.verify_certificate else 'no',
|
||||
}
|
||||
|
||||
return '{schema}://{password}:{user}@{host}/{targets}/?{args}'.format(
|
||||
schema=self.secure_protocol,
|
||||
password=self.quote(self.password, safe=''),
|
||||
user=self.quote(self.user, safe=''),
|
||||
host=self.host,
|
||||
targets='/'.join(
|
||||
[NotifyTwist.quote(x, safe='') for x in chain(
|
||||
# Channels are prefixed with a pound/hashtag symbol
|
||||
['#{}'.format(x) for x in self.channels],
|
||||
# Channel IDs
|
||||
self.channel_ids,
|
||||
)]),
|
||||
args=NotifyTwist.urlencode(args),
|
||||
)
|
||||
|
||||
def login(self):
|
||||
"""
|
||||
A simple wrapper to authenticate with the Twist Server
|
||||
"""
|
||||
|
||||
# Prepare our payload
|
||||
payload = {
|
||||
'email': self.email,
|
||||
'password': self.password,
|
||||
}
|
||||
|
||||
# Reset our default workspace
|
||||
self.default_workspace = None
|
||||
|
||||
# Reset our cached objects
|
||||
self._cached_workspaces = set()
|
||||
self._cached_channels = dict()
|
||||
|
||||
# Send Login Information
|
||||
postokay, response = self._fetch(
|
||||
'users/login',
|
||||
payload=payload,
|
||||
# We set this boolean so internal recursion doesn't take place.
|
||||
login=True,
|
||||
)
|
||||
|
||||
if not postokay or not response:
|
||||
# Setting this variable to False as a way of letting us know
|
||||
# we failed to authenticate on our last attempt
|
||||
self.token = False
|
||||
return False
|
||||
|
||||
# Our response object looks like this (content has been altered for
|
||||
# presentation purposes):
|
||||
# {
|
||||
# "contact_info": null,
|
||||
# "profession": null,
|
||||
# "timezone": "UTC",
|
||||
# "avatar_id": null,
|
||||
# "id": 123456,
|
||||
# "first_name": "Jordan",
|
||||
# "comet_channel":
|
||||
# "124371-34be423219130343030d4ec0a3dabbbbbe565eee",
|
||||
# "restricted": false,
|
||||
# "default_workspace": 92020,
|
||||
# "snooze_dnd_end": null,
|
||||
# "email": "user@example.com",
|
||||
# "comet_server": "https://comet.twist.com",
|
||||
# "snooze_until": null,
|
||||
# "lang": "en",
|
||||
# "feature_flags": [],
|
||||
# "short_name": "Jordan P.",
|
||||
# "away_mode": null,
|
||||
# "time_format": "12",
|
||||
# "client_id": "cb01f37e-a5b2-13e9-ba2a-023a33d10dc0",
|
||||
# "removed": false,
|
||||
# "emails": [
|
||||
# {
|
||||
# "connected": [],
|
||||
# "email": "user@example.com",
|
||||
# "primary": true
|
||||
# }
|
||||
# ],
|
||||
# "scheduled_banners": [
|
||||
# "threads_3",
|
||||
# "threads_1",
|
||||
# "notification_permissions",
|
||||
# "search_1",
|
||||
# "messages_1",
|
||||
# "team_1",
|
||||
# "inbox_2",
|
||||
# "inbox_1"
|
||||
# ],
|
||||
# "snooze_dnd_start": null,
|
||||
# "name": "Jordan Peterson",
|
||||
# "off_days": [],
|
||||
# "bot": false,
|
||||
# "token": "2e82c1e4e8b0091fdaa34ff3972351821406f796",
|
||||
# "snoozed": false,
|
||||
# "setup_pending": false,
|
||||
# "date_format": "MM/DD/YYYY"
|
||||
# }
|
||||
|
||||
# Store our default workspace
|
||||
self.default_workspace = response.get('default_workspace')
|
||||
|
||||
# Acquire our token
|
||||
self.token = response.get('token')
|
||||
|
||||
self.logger.info('Authenticated to Twist as {}'.format(self.email))
|
||||
return True
|
||||
|
||||
def logout(self):
|
||||
"""
|
||||
A simple wrapper to log out of the server
|
||||
"""
|
||||
|
||||
if not self.token:
|
||||
# Nothing more to do
|
||||
return True
|
||||
|
||||
# Send Logout Message
|
||||
postokay, response = self._fetch('users/logout')
|
||||
|
||||
# reset our token
|
||||
self.token = None
|
||||
|
||||
# There is no need to handling failed log out attempts at this time
|
||||
return True
|
||||
|
||||
def get_workspaces(self):
|
||||
"""
|
||||
Returns all workspaces associated with this user account as a set
|
||||
|
||||
This returned object is either an empty dictionary or one that
|
||||
looks like this:
|
||||
{
|
||||
'workspace': <workspace_id>,
|
||||
'workspace': <workspace_id>,
|
||||
'workspace': <workspace_id>,
|
||||
}
|
||||
|
||||
All workspaces are made lowercase for comparison purposes
|
||||
"""
|
||||
if not self.token and not self.login():
|
||||
# Nothing more to do
|
||||
return dict()
|
||||
|
||||
postokay, response = self._fetch('workspaces/get')
|
||||
if not postokay or not response:
|
||||
# We failed to retrieve
|
||||
return dict()
|
||||
|
||||
# The response object looks like so:
|
||||
# [
|
||||
# {
|
||||
# "created_ts": 1563044447,
|
||||
# "name": "apprise",
|
||||
# "creator": 123571,
|
||||
# "color": 1,
|
||||
# "default_channel": 13245,
|
||||
# "plan": "free",
|
||||
# "default_conversation": 63022,
|
||||
# "id": 12345
|
||||
# }
|
||||
# ]
|
||||
|
||||
# Knowing our response, we can iterate over each object and cache our
|
||||
# object
|
||||
result = {}
|
||||
for entry in response:
|
||||
result[entry.get('name', '').lower()] = entry.get('id', '')
|
||||
|
||||
return result
|
||||
|
||||
def get_channels(self, wid):
|
||||
"""
|
||||
Simply returns the channel objects associated with the specified
|
||||
workspace id.
|
||||
|
||||
This returned object is either an empty dictionary or one that
|
||||
looks like this:
|
||||
{
|
||||
'channel1': <channel_id>,
|
||||
'channel2': <channel_id>,
|
||||
'channel3': <channel_id>,
|
||||
}
|
||||
|
||||
All channels are made lowercase for comparison purposes
|
||||
"""
|
||||
if not self.token and not self.login():
|
||||
# Nothing more to do
|
||||
return {}
|
||||
|
||||
payload = {'workspace_id': wid}
|
||||
postokay, response = self._fetch(
|
||||
'channels/get', payload=payload)
|
||||
|
||||
if not postokay or not isinstance(response, list):
|
||||
# We failed to retrieve
|
||||
return {}
|
||||
|
||||
# Response looks like this:
|
||||
# [
|
||||
# {
|
||||
# "id": 123,
|
||||
# "name": "General"
|
||||
# "workspace_id": 12345,
|
||||
# "color": 1,
|
||||
# "description": "",
|
||||
# "archived": false,
|
||||
# "public": true,
|
||||
# "user_ids": [
|
||||
# 8754
|
||||
# ],
|
||||
# "created_ts": 1563044447,
|
||||
# "creator": 123571,
|
||||
# }
|
||||
# ]
|
||||
#
|
||||
# Knowing our response, we can iterate over each object and cache our
|
||||
# object
|
||||
result = {}
|
||||
for entry in response:
|
||||
result[entry.get('name', '').lower()] = entry.get('id', '')
|
||||
|
||||
return result
|
||||
|
||||
def _channel_migration(self):
|
||||
"""
|
||||
A simple wrapper to get all of the current workspaces including
|
||||
the default one. This plays a role in what channel(s) get notified
|
||||
and where.
|
||||
|
||||
A cache lookup has overhead, and is only required to be preformed
|
||||
if the user specified channels by their string value
|
||||
"""
|
||||
|
||||
if not self.token and not self.login():
|
||||
# Nothing more to do
|
||||
return False
|
||||
|
||||
if not len(self.channels):
|
||||
# Nothing to do; take an early exit
|
||||
return True
|
||||
|
||||
if self.default_workspace \
|
||||
and self.default_workspace not in self._cached_channels:
|
||||
# Get our default workspace entries
|
||||
self._cached_channels[self.default_workspace] = \
|
||||
self.get_channels(self.default_workspace)
|
||||
|
||||
# initialize our error tracking
|
||||
has_error = False
|
||||
|
||||
while len(self.channels):
|
||||
# Pop our channel off of the stack
|
||||
result = IS_CHANNEL.match(self.channels.pop())
|
||||
|
||||
# Populate our key variables
|
||||
workspace = result.group('workspace')
|
||||
channel = result.group('channel').lower()
|
||||
|
||||
# Acquire our workspace_id if we can
|
||||
if workspace:
|
||||
# We always work with the workspace in it's lowercase form
|
||||
workspace = workspace.lower()
|
||||
|
||||
# A workspace was defined
|
||||
if not len(self._cached_workspaces):
|
||||
# cache our workspaces; this only needs to be done once
|
||||
self._cached_workspaces = self.get_workspaces()
|
||||
|
||||
if workspace not in self._cached_workspaces:
|
||||
# not found
|
||||
self.logger.warning(
|
||||
'The Twist User {} is not associated with the '
|
||||
'Team {}'.format(self.email, workspace))
|
||||
|
||||
# Toggle our return flag
|
||||
has_error = True
|
||||
continue
|
||||
|
||||
# Store the workspace id
|
||||
workspace_id = self._cached_workspaces[workspace]
|
||||
|
||||
else:
|
||||
# use default workspace
|
||||
workspace_id = self.default_workspace
|
||||
|
||||
# Check to see if our channel exists in our default workspace
|
||||
if workspace_id in self._cached_channels \
|
||||
and channel in self._cached_channels[workspace_id]:
|
||||
# Store our channel ID
|
||||
self.channel_ids.add('{}:{}'.format(
|
||||
workspace_id,
|
||||
self._cached_channels[workspace_id][channel],
|
||||
))
|
||||
continue
|
||||
|
||||
# if we reach here, we failed to add our channel
|
||||
self.logger.warning(
|
||||
'The Channel #{} was not found{}.'.format(
|
||||
channel,
|
||||
'' if not workspace
|
||||
else ' with Team {}'.format(workspace),
|
||||
))
|
||||
|
||||
# Toggle our return flag
|
||||
has_error = True
|
||||
continue
|
||||
|
||||
# There is no need to handling failed log out attempts at this time
|
||||
return not has_error
|
||||
|
||||
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
|
||||
"""
|
||||
Perform Twist Notification
|
||||
"""
|
||||
|
||||
# error tracking (used for function return)
|
||||
has_error = False
|
||||
|
||||
if not self.token and not self.login():
|
||||
# We failed to authenticate - we're done
|
||||
return False
|
||||
|
||||
if len(self.channels) > 0:
|
||||
# Converts channels to their maped IDs if found; this is the only
|
||||
# way to send notifications to Twist
|
||||
self._channel_migration()
|
||||
|
||||
if not len(self.channel_ids):
|
||||
# We have nothing to notify
|
||||
return False
|
||||
|
||||
# Notify all of our identified channels
|
||||
ids = list(self.channel_ids)
|
||||
while len(ids) > 0:
|
||||
# Retrieve our Channel Object
|
||||
result = IS_CHANNEL_ID.match(ids.pop())
|
||||
|
||||
# We need both the workspace/team id and channel id
|
||||
channel_id = int(result.group('channel'))
|
||||
|
||||
# Prepare our payload
|
||||
payload = {
|
||||
'channel_id': channel_id,
|
||||
'title': title,
|
||||
'content': body,
|
||||
}
|
||||
|
||||
postokay, response = self._fetch(
|
||||
'threads/add',
|
||||
payload=payload,
|
||||
)
|
||||
|
||||
# only toggle has_error flag if we had an error
|
||||
if not postokay:
|
||||
# Mark our failure
|
||||
has_error = True
|
||||
continue
|
||||
|
||||
# If we reach here, we were successful
|
||||
self.logger.info(
|
||||
'Sent Twist notification to {}.'.format(
|
||||
result.group('name')))
|
||||
|
||||
return not has_error
|
||||
|
||||
def _fetch(self, url, payload=None, method='POST', login=False):
|
||||
"""
|
||||
Wrapper to Twist API requests object
|
||||
"""
|
||||
|
||||
# use what was specified, otherwise build headers dynamically
|
||||
headers = {
|
||||
'User-Agent': self.app_id,
|
||||
}
|
||||
|
||||
headers['Content-Type'] = \
|
||||
'application/x-www-form-urlencoded; charset=utf-8'
|
||||
|
||||
if self.token:
|
||||
# Set our token
|
||||
headers['Authorization'] = 'Bearer {}'.format(self.token)
|
||||
|
||||
# Prepare our api url
|
||||
api_url = '{}{}'.format(self.api_url, url)
|
||||
|
||||
# Some Debug Logging
|
||||
self.logger.debug('Twist {} URL: {} (cert_verify={})'.format(
|
||||
method, api_url, self.verify_certificate))
|
||||
self.logger.debug('Twist Payload: %s' % str(payload))
|
||||
|
||||
# Always call throttle before any remote server i/o is made;
|
||||
self.throttle()
|
||||
|
||||
# Initialize a default value for our content value
|
||||
content = {}
|
||||
|
||||
# acquire our request mode
|
||||
fn = requests.post if method == 'POST' else requests.get
|
||||
try:
|
||||
r = fn(
|
||||
api_url,
|
||||
data=payload,
|
||||
headers=headers,
|
||||
verify=self.verify_certificate)
|
||||
|
||||
# Get our JSON content if it's possible
|
||||
try:
|
||||
content = loads(r.content)
|
||||
|
||||
except (TypeError, ValueError, AttributeError):
|
||||
# TypeError = r.content is not a String
|
||||
# ValueError = r.content is Unparsable
|
||||
# AttributeError = r.content is None
|
||||
content = {}
|
||||
|
||||
# handle authentication errors where our token has just simply
|
||||
# expired. The error response content looks like this:
|
||||
# {
|
||||
# "error_code": 200,
|
||||
# "error_uuid": "af80bd0715434231a649f2258d7fb946",
|
||||
# "error_extra": {},
|
||||
# "error_string": "Invalid token"
|
||||
# }
|
||||
#
|
||||
# Authentication related codes:
|
||||
# 120 = You are not logged in
|
||||
# 200 = Invalid Token
|
||||
#
|
||||
# Source: https://developer.twist.com/v3/#errors
|
||||
#
|
||||
# We attempt to login again and retry the original request
|
||||
# if we aren't in the process of handling a login already
|
||||
if r.status_code != requests.codes.ok and login is False \
|
||||
and isinstance(content, dict) and \
|
||||
content.get('error_code') in (120, 200):
|
||||
# We failed to authenticate with our token; login one more
|
||||
# time and retry this original request
|
||||
if self.login():
|
||||
r = fn(
|
||||
api_url,
|
||||
data=payload,
|
||||
headers=headers,
|
||||
verify=self.verify_certificate)
|
||||
|
||||
# Get our JSON content if it's possible
|
||||
try:
|
||||
content = loads(r.content)
|
||||
|
||||
except (TypeError, ValueError, AttributeError):
|
||||
# TypeError = r.content is not a String
|
||||
# ValueError = r.content is Unparsable
|
||||
# AttributeError = r.content is None
|
||||
content = {}
|
||||
|
||||
if r.status_code != requests.codes.ok:
|
||||
# We had a problem
|
||||
status_str = \
|
||||
NotifyTwist.http_response_code_lookup(r.status_code)
|
||||
|
||||
self.logger.warning(
|
||||
'Failed to send Twist {} to {}: '
|
||||
'{}error={}.'.format(
|
||||
method,
|
||||
api_url,
|
||||
', ' if status_str else '',
|
||||
r.status_code))
|
||||
|
||||
self.logger.debug(
|
||||
'Response Details:\r\n{}'.format(r.content))
|
||||
|
||||
# Mark our failure
|
||||
return (False, content)
|
||||
|
||||
except requests.RequestException as e:
|
||||
self.logger.warning(
|
||||
'Exception received when sending Twist {} to {}: '.
|
||||
format(method, api_url))
|
||||
self.logger.debug('Socket Exception: %s' % str(e))
|
||||
|
||||
# Mark our failure
|
||||
return (False, content)
|
||||
|
||||
return (True, content)
|
||||
|
||||
@staticmethod
|
||||
def parse_url(url):
|
||||
"""
|
||||
Parses the URL and returns enough arguments that can allow
|
||||
us to substantiate this object.
|
||||
|
||||
"""
|
||||
results = NotifyBase.parse_url(url)
|
||||
|
||||
if not results:
|
||||
# We're done early as we couldn't load the results
|
||||
return results
|
||||
|
||||
if not results.get('user'):
|
||||
# A username is required
|
||||
return None
|
||||
|
||||
# Acquire our targets
|
||||
results['targets'] = NotifyTwist.split_path(results['fullpath'])
|
||||
|
||||
if not results.get('password'):
|
||||
# Password is required; we will accept the very first entry on the
|
||||
# path as a password instead
|
||||
if len(results['targets']) == 0:
|
||||
# No targets to get our password from
|
||||
return None
|
||||
|
||||
# We need to requote contents since this variable will get
|
||||
# unquoted later on in the process. This step appears a bit
|
||||
# hacky, but it allows us to support the password in this location
|
||||
# - twist://user@example.com/password
|
||||
results['password'] = NotifyTwist.quote(
|
||||
results['targets'].pop(0), safe='')
|
||||
|
||||
else:
|
||||
# Now we handle our format:
|
||||
# twist://password:email
|
||||
#
|
||||
# since URL logic expects
|
||||
# schema://user:password@host
|
||||
#
|
||||
# you can see how this breaks. The colon at the front delmits
|
||||
# passwords and you can see the twist:// url inverts what we
|
||||
# expect:
|
||||
# twist://password:user@example.com
|
||||
#
|
||||
# twist://abc123:bob@example.com using normal conventions would
|
||||
# have interpreted 'bob' as the password and 'abc123' as the user.
|
||||
# For the purpose of apprise simplifying this for us, we need to
|
||||
# swap these arguments when we prepare the email.
|
||||
|
||||
_password = results['user']
|
||||
results['user'] = results['password']
|
||||
results['password'] = _password
|
||||
|
||||
# The 'to' makes it easier to use yaml configuration
|
||||
if 'to' in results['qsd'] and len(results['qsd']['to']):
|
||||
results['targets'] += \
|
||||
NotifyTwist.parse_list(results['qsd']['to'])
|
||||
|
||||
return results
|
||||
|
||||
def __del__(self):
|
||||
"""
|
||||
Deconstructor
|
||||
"""
|
||||
try:
|
||||
self.logout()
|
||||
|
||||
except LookupError:
|
||||
# Python v3.5 call to requests can sometimes throw the exception
|
||||
# "/usr/lib64/python3.7/socket.py", line 748, in getaddrinfo
|
||||
# LookupError: unknown encoding: idna
|
||||
#
|
||||
# This occurs every time when running unit-tests against Apprise:
|
||||
# LANG=C.UTF-8 PYTHONPATH=$(pwd) py.test-3.7
|
||||
#
|
||||
# There has been an open issue on this since Jan 2017.
|
||||
# - https://bugs.python.org/issue29288
|
||||
#
|
||||
# A ~similar~ issue can be identified here in the requests
|
||||
# ticket system as unresolved and has provided work-arounds
|
||||
# - https://github.com/kennethreitz/requests/issues/3578
|
||||
pass
|
||||
@ -104,10 +104,10 @@ GET_SCHEMA_RE = re.compile(r'\s*(?P<schema>[a-z0-9]{2,9})://.*$', re.I)
|
||||
# Regular expression based and expanded from:
|
||||
# http://www.regular-expressions.info/email.html
|
||||
GET_EMAIL_RE = re.compile(
|
||||
r"((?P<label>[^+]+)\+)?"
|
||||
r"(?P<fulluser>((?P<label>[^+]+)\+)?"
|
||||
r"(?P<userid>[a-z0-9$%=_~-]+"
|
||||
r"(?:\.[a-z0-9$%+=_~-]+)"
|
||||
r"*)@(?P<domain>(?:[a-z0-9](?:[a-z0-9-]*[a-z0-9])?\.)+"
|
||||
r"*))@(?P<domain>(?:[a-z0-9](?:[a-z0-9-]*[a-z0-9])?\.)+"
|
||||
r"[a-z0-9](?:[a-z0-9-]*"
|
||||
r"[a-z0-9]))?",
|
||||
re.IGNORECASE,
|
||||
|
||||
@ -51,7 +51,7 @@ Boxcar, Discord, E-Mail, Emby, Faast, Flock, Gitter, Gotify, Growl, IFTTT,
|
||||
Join, KODI, Mailgun, MatterMost, Matrix, Microsoft Windows Notifications,
|
||||
Microsoft Teams, Nexmo, Notify MyAndroid, Prowl, Pushalot, PushBullet, Pushjet,
|
||||
Pushover, Rocket.Chat, Slack, Super Toasty, Stride, Telegram, Twilio, Twitter,
|
||||
XBMC, XMPP, Webex Teams}
|
||||
Twist, XBMC, XMPP, Webex Teams}
|
||||
|
||||
Name: python-%{pypi_name}
|
||||
Version: 0.7.8
|
||||
|
||||
2
setup.py
2
setup.py
@ -72,7 +72,7 @@ setup(
|
||||
keywords='Push Notifications Alerts Email AWS SNS Boxcar Discord Dbus '
|
||||
'Emby Faast Flock Gitter Gnome Gotify Growl IFTTT Join KODI Mailgun '
|
||||
'Matrix Mattermost Nexmo Prowl PushBullet Pushjet Pushed Pushover '
|
||||
'Rocket.Chat Ryver Slack Stride Telegram Twilio Twitter XBMC '
|
||||
'Rocket.Chat Ryver Slack Stride Telegram Twilio Twist Twitter XBMC '
|
||||
'Microsoft MSTeams Windows Webex CLI API',
|
||||
author='Chris Caron',
|
||||
author_email='lead2gold@gmail.com',
|
||||
|
||||
@ -2171,6 +2171,59 @@ TEST_URLS = (
|
||||
'test_requests_exceptions': True,
|
||||
}),
|
||||
|
||||
##################################
|
||||
# NotifyTwist
|
||||
##################################
|
||||
('twist://', {
|
||||
# Missing Email and Login
|
||||
'instance': None,
|
||||
}),
|
||||
('twist://:@/', {
|
||||
'instance': None,
|
||||
}),
|
||||
('twist://user@example.com/', {
|
||||
# No password
|
||||
'instance': None,
|
||||
}),
|
||||
('twist://user@example.com/password', {
|
||||
# Password acceptable as first entry in path
|
||||
'instance': plugins.NotifyTwist,
|
||||
# Expected notify() response is False because internally we would
|
||||
# have failed to login
|
||||
'notify_response': False,
|
||||
}),
|
||||
('twist://password:user1@example.com', {
|
||||
# password:login acceptable
|
||||
'instance': plugins.NotifyTwist,
|
||||
# Expected notify() response is False because internally we would
|
||||
# have failed to login
|
||||
'notify_response': False,
|
||||
}),
|
||||
('twist://password:user2@example.com', {
|
||||
# password:login acceptable
|
||||
'instance': plugins.NotifyTwist,
|
||||
# Expected notify() response is False because internally we would
|
||||
# have logged in, but we would have failed to look up the #General
|
||||
# channel and workspace.
|
||||
'requests_response_text': {
|
||||
# Login expected response
|
||||
'id': 1234,
|
||||
'default_workspace': 9876,
|
||||
},
|
||||
'notify_response': False,
|
||||
}),
|
||||
('twist://password:user2@example.com', {
|
||||
'instance': plugins.NotifyTwist,
|
||||
# throw a bizzare code forcing us to fail to look it up
|
||||
'response': False,
|
||||
'requests_response_code': 999,
|
||||
}),
|
||||
('twist://password:user2@example.com', {
|
||||
'instance': plugins.NotifyTwist,
|
||||
# Throws a series of connection and transfer exceptions when this flag
|
||||
# is set and tests that we gracfully handle them
|
||||
'test_requests_exceptions': True,
|
||||
}),
|
||||
##################################
|
||||
# NotifyTwitter
|
||||
##################################
|
||||
|
||||
485
test/test_twist_plugin.py
Normal file
485
test/test_twist_plugin.py
Normal file
@ -0,0 +1,485 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
|
||||
# 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 mock
|
||||
import requests
|
||||
from json import dumps
|
||||
from apprise import plugins
|
||||
from apprise import Apprise
|
||||
|
||||
# Disable logging for a cleaner testing output
|
||||
import logging
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
|
||||
def test_twist_plugin_init():
|
||||
"""
|
||||
API: NotifyTwist init()
|
||||
|
||||
"""
|
||||
try:
|
||||
plugins.NotifyTwist(email='invalid', targets=None)
|
||||
assert False
|
||||
except TypeError:
|
||||
# Invalid email address
|
||||
assert True
|
||||
|
||||
try:
|
||||
plugins.NotifyTwist(email='user@domain', targets=None)
|
||||
assert False
|
||||
except TypeError:
|
||||
# No password was specified
|
||||
assert True
|
||||
|
||||
# Simple object initialization
|
||||
result = plugins.NotifyTwist(
|
||||
password='abc123', email='user@domain.com', targets=None)
|
||||
assert result.user == 'user'
|
||||
assert result.host == 'domain.com'
|
||||
assert result.password == 'abc123'
|
||||
|
||||
# Channel Instantiation by name
|
||||
obj = Apprise.instantiate('twist://password:user@example.com/#Channel')
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
|
||||
# Channel Instantiation by id (faster if you know the translation)
|
||||
obj = Apprise.instantiate('twist://password:user@example.com/12345')
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
|
||||
# Invalid Channel - (max characters is 64), the below drops it
|
||||
obj = Apprise.instantiate(
|
||||
'twist://password:user@example.com/{}'.format('a' * 65))
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
|
||||
# No User detect
|
||||
result = plugins.NotifyTwist.parse_url('twist://example.com')
|
||||
assert result is None
|
||||
|
||||
# test usage of to=
|
||||
result = plugins.NotifyTwist.parse_url(
|
||||
'twist://password:user@example.com?to=#channel')
|
||||
assert isinstance(result, dict)
|
||||
assert 'user' in result
|
||||
assert result['user'] == 'user'
|
||||
assert 'host' in result
|
||||
assert result['host'] == 'example.com'
|
||||
assert 'password' in result
|
||||
assert result['password'] == 'password'
|
||||
assert 'targets' in result
|
||||
assert isinstance(result['targets'], list) is True
|
||||
assert len(result['targets']) == 1
|
||||
assert '#channel' in result['targets']
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@mock.patch('requests.post')
|
||||
def test_twist_plugin_auth(mock_post, mock_get):
|
||||
"""
|
||||
API: NotifyTwist login/logout()
|
||||
|
||||
"""
|
||||
# Disable Throttling to speed testing
|
||||
plugins.NotifyBase.request_rate_per_sec = 0
|
||||
|
||||
# Prepare Mock
|
||||
mock_get.return_value = requests.Request()
|
||||
mock_post.return_value = requests.Request()
|
||||
mock_post.return_value.status_code = requests.codes.ok
|
||||
mock_get.return_value.status_code = requests.codes.ok
|
||||
mock_post.return_value.content = dumps({
|
||||
'token': '2e82c1e4e8b0091fdaa34ff3972351821406f796',
|
||||
'default_workspace': 12345,
|
||||
})
|
||||
mock_get.return_value.content = mock_post.return_value.content
|
||||
|
||||
# Instantiate an object
|
||||
obj = Apprise.instantiate('twist://password:user@example.com/#Channel')
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
# not logged in yet
|
||||
obj.logout()
|
||||
assert obj.login() is True
|
||||
|
||||
# Clear our channel listing
|
||||
obj.channels.clear()
|
||||
# No channels mean there is no internal migration/lookups required
|
||||
assert obj._channel_migration() is True
|
||||
|
||||
# Workspace Success
|
||||
mock_post.return_value.content = dumps([
|
||||
{
|
||||
'name': 'TesT',
|
||||
'id': 1,
|
||||
}, {
|
||||
'name': 'tESt2',
|
||||
'id': 2,
|
||||
},
|
||||
])
|
||||
mock_get.return_value.content = mock_post.return_value.content
|
||||
|
||||
results = obj.get_workspaces()
|
||||
assert len(results) == 2
|
||||
assert 'test' in results
|
||||
assert results['test'] == 1
|
||||
assert 'test2' in results
|
||||
assert results['test2'] == 2
|
||||
|
||||
mock_post.return_value.content = dumps([
|
||||
{
|
||||
'name': 'ChaNNEL1',
|
||||
'id': 1,
|
||||
}, {
|
||||
'name': 'chaNNel2',
|
||||
'id': 2,
|
||||
},
|
||||
])
|
||||
mock_get.return_value.content = mock_post.return_value.content
|
||||
results = obj.get_channels(wid=1)
|
||||
assert len(results) == 2
|
||||
assert 'channel1' in results
|
||||
assert results['channel1'] == 1
|
||||
assert 'channel2' in results
|
||||
assert results['channel2'] == 2
|
||||
|
||||
# Test result failure response
|
||||
mock_post.return_value.status_code = 403
|
||||
mock_get.return_value.status_code = 403
|
||||
assert obj.get_workspaces() == dict()
|
||||
|
||||
# Return things how they were
|
||||
mock_post.return_value.status_code = requests.codes.ok
|
||||
mock_get.return_value.status_code = requests.codes.ok
|
||||
|
||||
# Forces call to logout:
|
||||
del obj
|
||||
|
||||
#
|
||||
# Authentication failures
|
||||
#
|
||||
mock_post.return_value.status_code = 403
|
||||
mock_get.return_value.status_code = 403
|
||||
|
||||
# Instantiate an object
|
||||
obj = Apprise.instantiate('twist://password:user@example.com/#Channel')
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
|
||||
# Authentication failed
|
||||
assert obj.get_workspaces() == dict()
|
||||
assert obj.get_channels(wid=1) == dict()
|
||||
assert obj._channel_migration() is False
|
||||
assert obj.send('body', 'title') is False
|
||||
|
||||
obj = Apprise.instantiate('twist://password:user@example.com/#Channel')
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
|
||||
# Calling logout on an object already logged out
|
||||
obj.logout()
|
||||
|
||||
# Force a token (to imply we've logged in)
|
||||
obj.token = 'abc'
|
||||
|
||||
mock_post.return_value.status_code = requests.codes.ok
|
||||
mock_get.return_value.status_code = requests.codes.ok
|
||||
|
||||
# Test Python v3.5 LookupError Bug: https://bugs.python.org/issue29288
|
||||
mock_post.side_effect = LookupError()
|
||||
mock_get.side_effect = LookupError()
|
||||
obj.access_token = 'abc'
|
||||
obj.user_id = '123'
|
||||
|
||||
# Tidy object
|
||||
del obj
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@mock.patch('requests.post')
|
||||
def test_twist_plugin_cache(mock_post, mock_get):
|
||||
"""
|
||||
API: NotifyTwist cache()
|
||||
|
||||
Test cache handling
|
||||
|
||||
"""
|
||||
# Disable Throttling to speed testing
|
||||
plugins.NotifyBase.request_rate_per_sec = 0
|
||||
|
||||
def _response(url, *args, **kwargs):
|
||||
|
||||
# Default configuration
|
||||
request = mock.Mock()
|
||||
request.status_code = requests.codes.ok
|
||||
request.content = '{}'
|
||||
|
||||
if url.endswith('/login'):
|
||||
# Simulate a successful login
|
||||
request.content = dumps({
|
||||
'token': '2e82c1e4e8b0091fdaa34ff3972351821406f796',
|
||||
'default_workspace': 1,
|
||||
})
|
||||
|
||||
elif url.endswith('workspaces/get'):
|
||||
request.content = dumps([
|
||||
{
|
||||
'name': 'TeamA',
|
||||
'id': 1,
|
||||
}, {
|
||||
'name': 'TeamB',
|
||||
'id': 2,
|
||||
},
|
||||
])
|
||||
|
||||
elif url.endswith('channels/get'):
|
||||
request.content = dumps([
|
||||
{
|
||||
'name': 'ChanA',
|
||||
'id': 1,
|
||||
}, {
|
||||
'name': 'ChanB',
|
||||
'id': 2,
|
||||
},
|
||||
])
|
||||
|
||||
return request
|
||||
|
||||
mock_get.side_effect = _response
|
||||
mock_post.side_effect = _response
|
||||
|
||||
# Instantiate an object
|
||||
obj = Apprise.instantiate(
|
||||
'twist://password:user@example.com/'
|
||||
'#ChanB/1:1/TeamA:ChanA/Ignore:Chan/3:1')
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
|
||||
# Will detect channels except Ignore:Chan
|
||||
assert obj._channel_migration() is False
|
||||
|
||||
# Add another channel
|
||||
obj.channels.add('ChanB')
|
||||
assert obj._channel_migration() is True
|
||||
|
||||
# Nothing more to detect the second time around
|
||||
assert obj._channel_migration() is True
|
||||
|
||||
# Send a notification
|
||||
assert obj.send('body', 'title') is True
|
||||
|
||||
def _can_not_send_response(url, *args, **kwargs):
|
||||
"""
|
||||
Simulate a case where we can't send a notification
|
||||
"""
|
||||
# Force a failure
|
||||
request = mock.Mock()
|
||||
request.status_code = 403
|
||||
request.content = '{}'
|
||||
return request
|
||||
|
||||
mock_get.side_effect = _can_not_send_response
|
||||
mock_post.side_effect = _can_not_send_response
|
||||
|
||||
# Send a notification and fail at it
|
||||
assert obj.send('body', 'title') is False
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@mock.patch('requests.post')
|
||||
def test_twist_plugin_fetch(mock_post, mock_get):
|
||||
"""
|
||||
API: NotifyTwist fetch()
|
||||
|
||||
fetch() is a wrapper that handles all kinds of edge cases and even
|
||||
attempts to re-authenticate to the Twist server if our token
|
||||
happens to expire. This tests these edge cases
|
||||
|
||||
"""
|
||||
# Disable Throttling to speed testing
|
||||
plugins.NotifyBase.request_rate_per_sec = 0
|
||||
|
||||
# Track our iteration; by tracing within an object, we can re-reference
|
||||
# it within a function scope.
|
||||
_cache = {
|
||||
'first_time': True,
|
||||
}
|
||||
|
||||
def _reauth_response(url, *args, **kwargs):
|
||||
"""
|
||||
Tests re-authentication process and then a successful
|
||||
retry
|
||||
"""
|
||||
|
||||
# Default configuration
|
||||
request = mock.Mock()
|
||||
request.status_code = requests.codes.ok
|
||||
|
||||
# Simulate a successful login
|
||||
request.content = dumps({
|
||||
'token': '2e82c1e4e8b0091fdaa34ff3972351821406f796',
|
||||
'default_workspace': 12345,
|
||||
})
|
||||
|
||||
if url.endswith('threads/add') and _cache['first_time'] is True:
|
||||
# First time iteration; act as if we failed; our second iteration
|
||||
# will not enter this and be successful. This is done by simply
|
||||
# toggling the first_time flag:
|
||||
_cache['first_time'] = False
|
||||
|
||||
# otherwise, we set our first-time failure settings
|
||||
request.status_code = 403
|
||||
request.content = dumps({
|
||||
'error_code': 200,
|
||||
'error_string': 'Invalid token',
|
||||
})
|
||||
|
||||
return request
|
||||
|
||||
mock_get.side_effect = _reauth_response
|
||||
mock_post.side_effect = _reauth_response
|
||||
|
||||
# Instantiate an object
|
||||
obj = Apprise.instantiate('twist://password:user@example.com/#Channel/34')
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
|
||||
# Simulate a re-authentication
|
||||
postokay, response = obj._fetch('threads/add')
|
||||
|
||||
##########################################################################
|
||||
_cache = {
|
||||
'first_time': True,
|
||||
}
|
||||
|
||||
def _reauth_exception_response(url, *args, **kwargs):
|
||||
"""
|
||||
Tests exception thrown after re-authentication process
|
||||
"""
|
||||
|
||||
# Default configuration
|
||||
request = mock.Mock()
|
||||
request.status_code = requests.codes.ok
|
||||
|
||||
# Simulate a successful login
|
||||
request.content = dumps({
|
||||
'token': '2e82c1e4e8b0091fdaa34ff3972351821406f796',
|
||||
'default_workspace': 12345,
|
||||
})
|
||||
|
||||
if url.endswith('threads/add') and _cache['first_time'] is True:
|
||||
# First time iteration; act as if we failed; our second iteration
|
||||
# will not enter this and be successful. This is done by simply
|
||||
# toggling the first_time flag:
|
||||
_cache['first_time'] = False
|
||||
|
||||
# otherwise, we set our first-time failure settings
|
||||
request.status_code = 403
|
||||
request.content = dumps({
|
||||
'error_code': 200,
|
||||
'error_string': 'Invalid token',
|
||||
})
|
||||
|
||||
elif url.endswith('threads/add') and _cache['first_time'] is False:
|
||||
# unparseable response throws the exception
|
||||
request.status_code = 200
|
||||
request.content = '{'
|
||||
|
||||
return request
|
||||
|
||||
mock_get.side_effect = _reauth_exception_response
|
||||
mock_post.side_effect = _reauth_exception_response
|
||||
|
||||
# Instantiate an object
|
||||
obj = Apprise.instantiate('twist://password:user@example.com/#Channel/34')
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
|
||||
# Simulate a re-authentication
|
||||
postokay, response = obj._fetch('threads/add')
|
||||
|
||||
##########################################################################
|
||||
_cache = {
|
||||
'first_time': True,
|
||||
}
|
||||
|
||||
def _reauth_failed_response(url, *args, **kwargs):
|
||||
"""
|
||||
Tests re-authentication process and have it not succeed
|
||||
"""
|
||||
|
||||
# Default configuration
|
||||
request = mock.Mock()
|
||||
request.status_code = requests.codes.ok
|
||||
|
||||
# Simulate a successful login
|
||||
request.content = dumps({
|
||||
'token': '2e82c1e4e8b0091fdaa34ff3972351821406f796',
|
||||
'default_workspace': 12345,
|
||||
})
|
||||
|
||||
if url.endswith('threads/add') and _cache['first_time'] is True:
|
||||
# First time iteration; act as if we failed; our second iteration
|
||||
# will not enter this and be successful. This is done by simply
|
||||
# toggling the first_time flag:
|
||||
_cache['first_time'] = False
|
||||
|
||||
# otherwise, we set our first-time failure settings
|
||||
request.status_code = 403
|
||||
request.content = dumps({
|
||||
'error_code': 200,
|
||||
'error_string': 'Invalid token',
|
||||
})
|
||||
|
||||
elif url.endswith('/login') and _cache['first_time'] is False:
|
||||
# Fail to login
|
||||
request.status_code = 403
|
||||
request.content = '{}'
|
||||
|
||||
return request
|
||||
|
||||
mock_get.side_effect = _reauth_failed_response
|
||||
mock_post.side_effect = _reauth_failed_response
|
||||
|
||||
# Instantiate an object
|
||||
obj = Apprise.instantiate('twist://password:user@example.com/#Channel/34')
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
|
||||
# Simulate a re-authentication
|
||||
postokay, response = obj._fetch('threads/add')
|
||||
|
||||
def _unparseable_json_response(url, *args, **kwargs):
|
||||
|
||||
# Default configuration
|
||||
request = mock.Mock()
|
||||
request.status_code = requests.codes.ok
|
||||
request.content = '{'
|
||||
return request
|
||||
|
||||
mock_get.side_effect = _unparseable_json_response
|
||||
mock_post.side_effect = _unparseable_json_response
|
||||
|
||||
# Instantiate our object
|
||||
obj = Apprise.instantiate('twist://password:user@example.com/#Channel/34')
|
||||
assert isinstance(obj, plugins.NotifyTwist)
|
||||
|
||||
# Simulate a re-authentication
|
||||
postokay, response = obj._fetch('threads/add')
|
||||
assert postokay is True
|
||||
# When we can't parse the content, we still default to an empty
|
||||
# dictionary
|
||||
assert response == {}
|
||||
Loading…
Reference in New Issue
Block a user