Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_twitter_plugin_init():
"""
API: NotifyTwitter Plugin() (pt2)
"""
with pytest.raises(TypeError):
plugins.NotifyTwitter(
ckey=None, csecret=None, akey=None, asecret=None)
with pytest.raises(TypeError):
plugins.NotifyTwitter(
ckey='value', csecret=None, akey=None, asecret=None)
with pytest.raises(TypeError):
plugins.NotifyTwitter(
ckey='value', csecret='value', akey=None, asecret=None)
with pytest.raises(TypeError):
plugins.NotifyTwitter(
ckey='value', csecret='value', akey='value', asecret=None)
assert isinstance(
plugins.NotifyTwitter(
logging.disable(logging.CRITICAL)
TEST_URLS = (
##################################
# NotifyGrowl
##################################
('growl://', {
'instance': None,
}),
('growl://:@/', {
'instance': None
}),
('growl://pass@growl.server', {
'instance': plugins.NotifyGrowl,
}),
('growl://ignored:pass@growl.server', {
'instance': plugins.NotifyGrowl,
}),
('growl://growl.server', {
'instance': plugins.NotifyGrowl,
# don't include an image by default
'include_image': False,
}),
('growl://growl.server?version=1', {
'instance': plugins.NotifyGrowl,
}),
# Force a failure
('growl://growl.server?version=1', {
'instance': plugins.NotifyGrowl,
'growl_response': None,
request = mock.Mock()
request.content = dumps(response_obj)
request.status_code = requests.codes.ok
request.headers = {
'X-RateLimit-Reset': (datetime.utcnow() - epoch).total_seconds(),
'X-RateLimit-Remaining': 1,
}
# Prepare Mock
mock_get.return_value = request
mock_post.return_value = request
# Variation Initializations
obj = plugins.NotifyGitter(token=token, targets='apprise')
assert isinstance(obj, plugins.NotifyGitter) is True
assert isinstance(obj.url(), six.string_types) is True
# apprise room was found
assert obj.send(body="test") is True
# Change our status code and try again
request.status_code = 403
assert obj.send(body="test") is False
assert obj.ratelimit_remaining == 1
# Return the status
request.status_code = requests.codes.ok
# Force a reset
request.headers['X-RateLimit-Remaining'] = 0
# behind the scenes, it should cause us to update our rate limit
assert obj.send(body="test") is True
assert isinstance(obj.url(), six.string_types) is True
assert obj.notify(
title='title', body='body',
notify_type=apprise.NotifyType.INFO) is True
obj = apprise.Apprise.instantiate(
'dbus://_/?urgency=high', suppress_exceptions=False)
assert isinstance(obj, apprise.plugins.NotifyDBus) is True
assert isinstance(obj.url(), six.string_types) is True
assert obj.notify(
title='title', body='body',
notify_type=apprise.NotifyType.INFO) is True
obj = apprise.Apprise.instantiate(
'dbus://_/?urgency=2', suppress_exceptions=False)
assert isinstance(obj, apprise.plugins.NotifyDBus) is True
assert isinstance(obj.url(), six.string_types) is True
assert obj.notify(
title='title', body='body',
notify_type=apprise.NotifyType.INFO) is True
obj = apprise.Apprise.instantiate(
'dbus://_/?urgency=', suppress_exceptions=False)
assert isinstance(obj, apprise.plugins.NotifyDBus) is True
assert isinstance(obj.url(), six.string_types) is True
assert obj.notify(
title='title', body='body',
notify_type=apprise.NotifyType.INFO) is True
# Test x/y
obj = apprise.Apprise.instantiate(
'dbus://_/?x=5&y=5', suppress_exceptions=False)
response_obj = {
'screen_name': screen_name,
'id': 9876,
}
request.content = dumps(response_obj)
obj = plugins.NotifyTwitter(
ckey=ckey,
csecret=csecret,
akey=akey,
asecret=asecret)
assert obj.send(body="test") is True
# Alter the key forcing us to look up a new value of ourselves again
del plugins.NotifyTwitter._user_cache
del plugins.NotifyTwitter._whoami_cache
obj.ckey = 'different.then.it.was'
assert obj.send(body="test") is True
del plugins.NotifyTwitter._whoami_cache
obj.ckey = 'different.again'
assert obj.send(body="test") is True
def test_notify_matrix_plugin_general(mock_post, mock_get):
"""
API: NotifyMatrix() General Tests
"""
# Disable Throttling to speed testing
plugins.NotifyBase.request_rate_per_sec = 0
response_obj = {
'room_id': '!abc123:localhost',
'room_alias': '#abc123:localhost',
'joined_rooms': ['!abc123:localhost', '!def456:localhost'],
'access_token': 'abcd1234',
'home_server': 'localhost',
}
request = mock.Mock()
request.content = dumps(response_obj)
request.status_code = requests.codes.ok
# Prepare Mock
mock_get.return_value = request
mock_post.return_value = request
('mailto://user:pass@yahoo.ca', {
'instance': plugins.NotifyEmail,
}),
('mailto://user:pass@fastmail.com', {
'instance': plugins.NotifyEmail,
}),
('mailto://user:pass@sendgrid.com', {
'instance': plugins.NotifyEmail,
}),
# Custom Emails
('mailtos://user:pass@nuxref.com:567', {
'instance': plugins.NotifyEmail,
}),
('mailto://user:pass@nuxref.com:567?format=html', {
'instance': plugins.NotifyEmail,
}),
('mailtos://user:pass@nuxref.com:567?to=l2g@nuxref.com', {
'instance': plugins.NotifyEmail,
}),
('mailtos://user:pass@nuxref.com:567/l2g@nuxref.com', {
'instance': plugins.NotifyEmail,
}),
(
'mailtos://user:pass@example.com?smtp=smtp.example.com&timeout=5'
'&name=l2g&from=noreply@example.com', {
'instance': plugins.NotifyEmail,
},
),
('mailto://user:pass@example.com?timeout=invalid.entry', {
'instance': plugins.NotifyEmail,
}),
def test_notify_gitter_plugin_general(mock_post, mock_get):
"""
API: NotifyGitter() General Tests
"""
# Disable Throttling to speed testing
plugins.NotifyBase.request_rate_per_sec = 0
# Generate a valid token (40 characters)
token = 'a' * 40
response_obj = [
{
'noindex': False,
'oneToOne': False,
'avatarUrl': 'https://path/to/avatar/url',
'url': '/apprise-notifications/community',
'public': True,
'tags': [],
'lurk': False,
'uri': 'apprise-notifications/community',
'lastAccessTime': '2019-03-25T00:12:28.144Z',
'topic': '',
# A a bad url
('boxcar://:@/', {
'instance': TypeError,
}),
# No secret specified
('boxcar://%s' % ('a' * 64), {
'instance': TypeError,
}),
# An invalid access and secret key specified
('boxcar://access.key/secret.key/', {
# Thrown because there were no recipients specified
'instance': TypeError,
}),
# Provide both an access and a secret
('boxcar://%s/%s' % ('a' * 64, 'b' * 64), {
'instance': plugins.NotifyBoxcar,
'requests_response_code': requests.codes.created,
}),
# Test without image set
('boxcar://%s/%s?image=True' % ('a' * 64, 'b' * 64), {
'instance': plugins.NotifyBoxcar,
'requests_response_code': requests.codes.created,
# don't include an image in Asset by default
'include_image': False,
}),
('boxcar://%s/%s?image=False' % ('a' * 64, 'b' * 64), {
'instance': plugins.NotifyBoxcar,
'requests_response_code': requests.codes.created,
}),
# our access, secret and device are all 64 characters
# which is what we're doing here
('boxcar://%s/%s/@tag1/tag2///%s/?to=tag3' % (
# Attempt to create an instance of our plugin using the parsed
# URL information
plugin = plugins.SCHEMA_MAP[results['schema']](**results)
# Create log entry of loaded URL
logger.debug('Loaded URL: {}'.format(plugin.url()))
except Exception:
# the arguments are invalid or can not be used.
logger.error('Could not load URL: %s' % url)
return None
else:
# Attempt to create an instance of our plugin using the parsed
# URL information but don't wrap it in a try catch
plugin = plugins.SCHEMA_MAP[results['schema']](**results)
return plugin