extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-11-21 23:53:23 +01:00
Code Issues Packages Projects Releases Wiki Activity

Pushjet refactored; dropped pushjet library and deps (#147)

Browse Source
This commit is contained in:
Chris Caron 2019-09-07 18:39:18 -04:00 committed by GitHub
parent f29af0c55b
commit a420375cc7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 356 additions and 824 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.coveragerc
View File

@ -1,5 +1,5 @@
[run]
omit=*/gntp/*,*/pushjet/*
omit=*/gntp/*
disable_warnings = no-data-collected
branch = True
source =

2
README.md
View File

@ -51,7 +51,7 @@ The table below identifies the services this tool supports and some example serv
| [Microsoft Teams](https://github.com/caronc/apprise/wiki/Notify_msteams) | msteams:// | (TCP) 443 | msteams://TokenA/TokenB/TokenC/
| [Prowl](https://github.com/caronc/apprise/wiki/Notify_prowl) | prowl:// | (TCP) 443 | prowl://apikey<br />prowl://apikey/providerkey
| [PushBullet](https://github.com/caronc/apprise/wiki/Notify_pushbullet) | pbul:// | (TCP) 443 | pbul://accesstoken<br />pbul://accesstoken/#channel<br/>pbul://accesstoken/A_DEVICE_ID<br />pbul://accesstoken/email@address.com<br />pbul://accesstoken/#channel/#channel2/email@address.net/DEVICE
| [Pushjet](https://github.com/caronc/apprise/wiki/Notify_pushjet) | pjet:// or pjets:// | (TCP) 80 or 443 | pjet://secret@hostname<br />pjet://secret@hostname:port<br />pjets://secret@hostname<br />pjets://secret@hostname:port
| [Pushjet](https://github.com/caronc/apprise/wiki/Notify_pushjet) | pjet:// or pjets:// | (TCP) 80 or 443 | pjet://hostname/secret<br />pjet://hostname:port/secret<br />pjets://secret@hostname/secret<br />pjets://hostname:port/secret
| [Push (Techulus)](https://github.com/caronc/apprise/wiki/Notify_techulus) | push:// | (TCP) 443 | push://apikey/
| [Pushed](https://github.com/caronc/apprise/wiki/Notify_pushed) | pushed:// | (TCP) 443 | pushed://appkey/appsecret/<br/>pushed://appkey/appsecret/#ChannelAlias<br/>pushed://appkey/appsecret/#ChannelAlias1/#ChannelAlias2/#ChannelAliasN<br/>pushed://appkey/appsecret/@UserPushedID<br/>pushed://appkey/appsecret/@UserPushedID1/@UserPushedID2/@UserPushedIDN
| [Pushover](https://github.com/caronc/apprise/wiki/Notify_pushover) | pover:// | (TCP) 443 | pover://user@token<br />pover://user@token/DEVICE<br />pover://user@token/DEVICE1/DEVICE2/DEVICEN<br />**Note**: you must specify both your user_id and token

292
apprise/plugins/NotifyPushjet.py Normal file
View File

@ -0,0 +1,292 @@
# -*- 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 requests
from json import dumps
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..AppriseLocale import gettext_lazy as _
class NotifyPushjet(NotifyBase):
"""
A wrapper for Pushjet Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Pushjet'
# The default protocol
protocol = 'pjet'
# The default secure protocol
secure_protocol = 'pjets'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_pushjet'
# Disable throttle rate for Pushjet requests since they are normally
# local anyway (the remote/online service is no more)
request_rate_per_sec = 0
# Define object templates
templates = (
'{schema}://{host}:{port}/{secret_key}',
'{schema}://{host}/{secret_key}',
'{schema}://{user}:{password}@{host}:{port}/{secret_key}',
'{schema}://{user}:{password}@{host}/{secret_key}',
# Kept for backwards compatibility; will be depricated eventually
'{schema}://{secret_key}@{host}',
'{schema}://{secret_key}@{host}:{port}',
)
# Define our tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'secret_key': {
'name': _('Secret Key'),
'type': 'string',
'required': True,
'private': True,
},
'user': {
'name': _('Username'),
'type': 'string',
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
},
})
template_args = dict(NotifyBase.template_args, **{
'secret': {
'alias_of': 'secret_key',
},
})
def __init__(self, secret_key, **kwargs):
"""
Initialize Pushjet Object
"""
super(NotifyPushjet, self).__init__(**kwargs)
if not secret_key:
# You must provide a Pushjet key to work with
msg = 'You must specify a Pushjet Secret Key.'
self.logger.warning(msg)
raise TypeError(msg)
# store our key
self.secret_key = secret_key
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,
'secret': self.secret_key,
'verify': 'yes' if self.verify_certificate else 'no',
}
default_port = 443 if self.secure else 80
# Determine Authentication
auth = ''
if self.user and self.password:
auth = '{user}:{password}@'.format(
user=NotifyPushjet.quote(self.user, safe=''),
password=NotifyPushjet.quote(self.password, safe=''),
)
return '{schema}://{auth}{hostname}{port}/?{args}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyPushjet.quote(self.host, safe=''),
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
args=NotifyPushjet.urlencode(args),
)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Pushjet Notification
"""
params = {
'secret': self.secret_key,
}
# prepare Pushjet Object
payload = {
'message': body,
'title': title,
'link': None,
'level': None,
}
headers = {
'User-Agent': self.app_id,
'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8',
}
auth = None
if self.user:
auth = (self.user, self.password)
notify_url = '{schema}://{host}{port}/message/'.format(
schema="https" if self.secure else "http",
host=self.host,
port=':{}'.format(self.port) if self.port else '')
self.logger.debug('Pushjet POST URL: %s (cert_verify=%r)' % (
notify_url, self.verify_certificate,
))
self.logger.debug('Pushjet Payload: %s' % str(payload))
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.post(
notify_url,
params=params,
data=dumps(payload),
headers=headers,
auth=auth,
verify=self.verify_certificate,
)
if r.status_code != requests.codes.ok:
# We had a problem
status_str = \
NotifyPushjet.http_response_code_lookup(r.status_code)
self.logger.warning(
'Failed to send Pushjet notification: '
'{}{}error={}.'.format(
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug('Response Details:\r\n{}'.format(r.content))
# Return; we're done
return False
else:
self.logger.info('Sent Pushjet notification.')
except requests.RequestException as e:
self.logger.warning(
'A Connection error occured sending Pushjet '
'notification to %s.' % self.host)
self.logger.debug('Socket Exception: %s' % str(e))
# Return; we're done
return False
return True
@staticmethod
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to substantiate this object.
Syntax:
pjet://hostname/secret_key
pjet://hostname:port/secret_key
pjet://user:pass@hostname/secret_key
pjet://user:pass@hostname:port/secret_key
pjets://hostname/secret_key
pjets://hostname:port/secret_key
pjets://user:pass@hostname/secret_key
pjets://user:pass@hostname:port/secret_key
Legacy (Depricated) Syntax:
pjet://secret_key@hostname
pjet://secret_key@hostname:port
pjets://secret_key@hostname
pjets://secret_key@hostname:port
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
try:
# Retrieve our secret_key from the first entry in the url path
results['secret_key'] = \
NotifyPushjet.split_path(results['fullpath'])[0]
except IndexError:
# no secret key specified
results['secret_key'] = None
# Allow over-riding the secret by specifying it as an argument
# this allows people who have http-auth infront to login
# through it in addition to supporting the secret key
if 'secret' in results['qsd'] and len(results['qsd']['secret']):
results['secret_key'] = \
NotifyPushjet.parse_list(results['qsd']['secret'])
if results.get('secret_key') is None:
# Deprication Notice issued for v0.7.9
NotifyPushjet.logger.deprecate(
'The Pushjet URL contains secret_key in the user field'
' which will be deprecated in an upcoming '
'release. Please place this in the path of the URL instead.'
)
# Store it as it's value based on the user field
results['secret_key'] = \
NotifyPushjet.unquote(results.get('user'))
# there is no way http-auth is enabled, be sure to unset the
# current defined user (if present). This is done due to some
# logic that takes place in the send() since we support http-auth.
results['user'] = None
results['password'] = None
return results

175
apprise/plugins/NotifyPushjet/__init__.py
View File

@ -1,175 +0,0 @@
# -*- 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 re
from . import pushjet
from ..NotifyBase import NotifyBase
from ...common import NotifyType
from ...AppriseLocale import gettext_lazy as _
PUBLIC_KEY_RE = re.compile(
r'^[a-z0-9]{4}-[a-z0-9]{6}-[a-z0-9]{12}-[a-z0-9]{5}-[a-z0-9]{9}$', re.I)
SECRET_KEY_RE = re.compile(r'^[a-z0-9]{32}$', re.I)
class NotifyPushjet(NotifyBase):
"""
A wrapper for Pushjet Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Pushjet'
# The default protocol
protocol = 'pjet'
# The default secure protocol
secure_protocol = 'pjets'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_pushjet'
# Disable throttle rate for Pushjet requests since they are normally
# local anyway (the remote/online service is no more)
request_rate_per_sec = 0
# Define object templates
templates = (
'{schema}://{secret_key}@{host}',
'{schema}://{secret_key}@{host}:{port}',
)
# Define our tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'secret_key': {
'name': _('Secret Key'),
'type': 'string',
'required': True,
'private': True,
},
})
def __init__(self, secret_key, **kwargs):
"""
Initialize Pushjet Object
"""
super(NotifyPushjet, self).__init__(**kwargs)
if not secret_key:
# You must provide a Pushjet key to work with
msg = 'You must specify a Pushjet Secret Key.'
self.logger.warning(msg)
raise TypeError(msg)
# store our key
self.secret_key = secret_key
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Pushjet Notification
"""
# Always call throttle before any remote server i/o is made
self.throttle()
server = "https://" if self.secure else "http://"
server += self.host
if self.port:
server += ":" + str(self.port)
try:
api = pushjet.pushjet.Api(server)
service = api.Service(secret_key=self.secret_key)
service.send(body, title)
self.logger.info('Sent Pushjet notification.')
except (pushjet.errors.PushjetError, ValueError) as e:
self.logger.warning('Failed to send Pushjet notification.')
self.logger.debug('Pushjet Exception: %s' % str(e))
return False
return True
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',
}
default_port = 443 if self.secure else 80
return '{schema}://{secret_key}@{hostname}{port}/?{args}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
secret_key=NotifyPushjet.quote(self.secret_key, safe=''),
hostname=NotifyPushjet.quote(self.host, safe=''),
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
args=NotifyPushjet.urlencode(args),
)
@staticmethod
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to substantiate this object.
Syntax:
pjet://secret_key@hostname
pjet://secret_key@hostname:port
pjets://secret_key@hostname
pjets://secret_key@hostname:port
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
# Store it as it's value
results['secret_key'] = \
NotifyPushjet.unquote(results.get('user'))
return results

6
apprise/plugins/NotifyPushjet/pushjet/__init__.py
View File

@ -1,6 +0,0 @@
# -*- coding: utf-8 -*-
"""A Python API for Pushjet. Send notifications to your phone from Python scripts!"""
from .pushjet import Service, Device, Subscription, Message, Api
from .errors import PushjetError, AccessError, NonexistentError, SubscriptionError, RequestError, ServerError

48
apprise/plugins/NotifyPushjet/pushjet/errors.py
View File

@ -1,48 +0,0 @@
# -*- coding: utf-8 -*-
from requests import RequestException
import sys
if sys.version_info[0] < 3:
# This is built into Python 3.
class ConnectionError(Exception):
pass
class PushjetError(Exception):
"""All the errors inherit from this. Therefore, ``except PushjetError`` catches all errors."""
class AccessError(PushjetError):
"""Raised when a secret key is missing for a service method that needs one."""
class NonexistentError(PushjetError):
"""Raised when an attempt to access a nonexistent service is made."""
class SubscriptionError(PushjetError):
"""Raised when an attempt to subscribe to a service that's already subscribed to,
or to unsubscribe from a service that isn't subscribed to, is made."""
class RequestError(PushjetError, ConnectionError):
"""Raised if something goes wrong in the connection to the API server.
Inherits from ``ConnectionError`` on Python 3, and can therefore be caught
with ``except ConnectionError`` there.
:ivar requests_exception: The underlying `requests <http://docs.python-requests.org>`__
exception. Access this if you want to handle different HTTP request errors in different ways.
"""
def __str__(self):
return "requests.{error}: {description}".format(
error=self.requests_exception.__class__.__name__,
description=str(self.requests_exception)
)
def __init__(self, requests_exception):
self.requests_exception = requests_exception
class ServerError(PushjetError):
"""Raised if the API server has an error while processing your request.
This getting raised means there's a bug in the server! If you manage to
track down what caused it, you can `open an issue on Pushjet's GitHub page
<https://github.com/Pushjet/Pushjet-Server-Api/issues>`__.
"""

313
apprise/plugins/NotifyPushjet/pushjet/pushjet.py
View File

@ -1,313 +0,0 @@
# -*- coding: utf-8 -*-
import sys
import requests
from functools import partial
from six import text_type
from six.moves.urllib.parse import urljoin
from .utilities import (
NoNoneDict,
requires_secret_key, with_api_bound,
is_valid_uuid, is_valid_public_key, is_valid_secret_key, repr_format
)
from .errors import NonexistentError, SubscriptionError, RequestError, ServerError
DEFAULT_API_URL = 'https://api.pushjet.io/'
class PushjetModel(object):
_api = None # This is filled in later.
class Service(PushjetModel):
"""A Pushjet service to send messages through. To receive messages, devices
subscribe to these.
:param secret_key: The service's API key for write access. If provided,
:func:`~pushjet.Service.send`, :func:`~pushjet.Service.edit`, and
:func:`~pushjet.Service.delete` become available.
Either this or the public key parameter must be present.
:param public_key: The service's public API key for read access only.
Either this or the secret key parameter must be present.
:ivar name: The name of the service.
:ivar icon_url: The URL to the service's icon. May be ``None``.
:ivar created: When the service was created, as seconds from epoch.
:ivar secret_key: The service's secret API key, or ``None`` if the service is read-only.
:ivar public_key: The service's public API key, to be used when subscribing to the service.
"""
def __repr__(self):
return "<Pushjet Service: \"{}\">".format(repr_format(self.name))
def __init__(self, secret_key=None, public_key=None):
if secret_key is None and public_key is None:
raise ValueError("Either a secret key or public key "
"must be provided.")
elif secret_key and not is_valid_secret_key(secret_key):
raise ValueError("Invalid secret key provided.")
elif public_key and not is_valid_public_key(public_key):
raise ValueError("Invalid public key provided.")
self.secret_key = text_type(secret_key) if secret_key else None
self.public_key = text_type(public_key) if public_key else None
self.refresh()
def _request(self, endpoint, method, is_secret, params=None, data=None):
params = params or {}
if is_secret:
params['secret'] = self.secret_key
else:
params['service'] = self.public_key
return self._api._request(endpoint, method, params, data)
@requires_secret_key
def send(self, message, title=None, link=None, importance=None):
"""Send a message to the service's subscribers.
:param message: The message body to be sent.
:param title: (optional) The message's title. Messages can be without title.
:param link: (optional) An URL to be sent with the message.
:param importance: (optional) The priority level of the message. May be
a number between 1 and 5, where 1 is least important and 5 is most.
"""
data = NoNoneDict({
'message': message,
'title': title,
'link': link,
'level': importance
})
self._request('message', 'POST', is_secret=True, data=data)
@requires_secret_key
def edit(self, name=None, icon_url=None):
"""Edit the service's attributes.
:param name: (optional) A new name to give the service.
:param icon_url: (optional) A new URL to use as the service's icon URL.
Set to an empty string to remove the service's icon entirely.
"""
data = NoNoneDict({
'name': name,
'icon': icon_url
})
if not data:
return
self._request('service', 'PATCH', is_secret=True, data=data)
self.name = text_type(name)
self.icon_url = text_type(icon_url)
@requires_secret_key
def delete(self):
"""Delete the service. Irreversible."""
self._request('service', 'DELETE', is_secret=True)
def _update_from_data(self, data):
self.name = data['name']
self.icon_url = data['icon'] or None
self.created = data['created']
self.public_key = data['public']
self.secret_key = data.get('secret', getattr(self, 'secret_key', None))
def refresh(self):
"""Refresh the server's information, in case it could be edited from elsewhere.
:raises: :exc:`~pushjet.NonexistentError` if the service was deleted before refreshing.
"""
key_name = 'public'
secret = False
if self.secret_key is not None:
key_name = 'secret'
secret = True
status, response = self._request('service', 'GET', is_secret=secret)
if status == requests.codes.NOT_FOUND:
raise NonexistentError("A service with the provided {} key "
"does not exist (anymore, at least).".format(key_name))
self._update_from_data(response['service'])
@classmethod
def _from_data(cls, data):
# This might be a no-no, but I see little alternative if
# different constructors with different parameters are needed,
# *and* a default __init__ constructor should be present.
# This, along with the subclassing for custom API URLs, may
# very well be one of those pieces of code you look back at
# years down the line - or maybe just a couple of weeks - and say
# "what the heck was I thinking"? I assure you, though, future me.
# This was the most reasonable thing to get the API + argspecs I wanted.
obj = cls.__new__(cls)
obj._update_from_data(data)
return obj
@classmethod
def create(cls, name, icon_url=None):
"""Create a new service.
:param name: The name of the new service.
:param icon_url: (optional) An URL to an image to be used as the service's icon.
:return: The newly-created :class:`~pushjet.Service`.
"""
data = NoNoneDict({
'name': name,
'icon': icon_url
})
_, response = cls._api._request('service', 'POST', data=data)
return cls._from_data(response['service'])
class Device(PushjetModel):
"""The "receiver" for messages. Subscribes to services and receives any
messages they send.
:param uuid: The device's unique ID as a UUID. Does not need to be registered
before using it. A UUID can be generated with ``uuid.uuid4()``, for example.
:ivar uuid: The UUID the device was initialized with.
"""
def __repr__(self):
return "<Pushjet Device: {}>".format(self.uuid)
def __init__(self, uuid):
uuid = text_type(uuid)
if not is_valid_uuid(uuid):
raise ValueError("Invalid UUID provided. Try uuid.uuid4().")
self.uuid = text_type(uuid)
def _request(self, endpoint, method, params=None, data=None):
params = (params or {})
params['uuid'] = self.uuid
return self._api._request(endpoint, method, params, data)
def subscribe(self, service):
"""Subscribe the device to a service.
:param service: The service to subscribe to. May be a public key or a :class:`~pushjet.Service`.
:return: The :class:`~pushjet.Service` subscribed to.
:raises: :exc:`~pushjet.NonexistentError` if the provided service does not exist.
:raises: :exc:`~pushjet.SubscriptionError` if the provided service is already subscribed to.
"""
data = {}
data['service'] = service.public_key if isinstance(service, Service) else service
status, response = self._request('subscription', 'POST', data=data)
if status == requests.codes.CONFLICT:
raise SubscriptionError("The device is already subscribed to that service.")
elif status == requests.codes.NOT_FOUND:
raise NonexistentError("A service with the provided public key "
"does not exist (anymore, at least).")
return self._api.Service._from_data(response['service'])
def unsubscribe(self, service):
"""Unsubscribe the device from a service.
:param service: The service to unsubscribe from. May be a public key or a :class:`~pushjet.Service`.
:raises: :exc:`~pushjet.NonexistentError` if the provided service does not exist.
:raises: :exc:`~pushjet.SubscriptionError` if the provided service isn't subscribed to.
"""
data = {}
data['service'] = service.public_key if isinstance(service, Service) else service
status, _ = self._request('subscription', 'DELETE', data=data)
if status == requests.codes.CONFLICT:
raise SubscriptionError("The device is not subscribed to that service.")
elif status == requests.codes.NOT_FOUND:
raise NonexistentError("A service with the provided public key "
"does not exist (anymore, at least).")
def get_subscriptions(self):
"""Get all the subscriptions the device has.
:return: A list of :class:`~pushjet.Subscription`.
"""
_, response = self._request('subscription', 'GET')
subscriptions = []
for subscription_dict in response['subscriptions']:
subscriptions.append(Subscription(subscription_dict))
return subscriptions
def get_messages(self):
"""Get all new (that is, as of yet unretrieved) messages.
:return: A list of :class:`~pushjet.Message`.
"""
_, response = self._request('message', 'GET')
messages = []
for message_dict in response['messages']:
messages.append(Message(message_dict))
return messages
class Subscription(object):
"""A subscription to a service, with the metadata that entails.
:ivar service: The service the subscription is to, as a :class:`~pushjet.Service`.
:ivar time_subscribed: When the subscription was made, as seconds from epoch.
:ivar last_checked: When the device last retrieved messages from the subscription,
as seconds from epoch.
:ivar device_uuid: The UUID of the device that owns the subscription.
"""
def __repr__(self):
return "<Pushjet Subscription to service \"{}\">".format(repr_format(self.service.name))
def __init__(self, subscription_dict):
self.service = Service._from_data(subscription_dict['service'])
self.time_subscribed = subscription_dict['timestamp']
self.last_checked = subscription_dict['timestamp_checked']
self.device_uuid = subscription_dict['uuid'] # Not sure this is needed, but...
class Message(object):
"""A message received from a service.
:ivar message: The message body.
:ivar title: The message title. May be ``None``.
:ivar link: The URL the message links to. May be ``None``.
:ivar time_sent: When the message was sent, as seconds from epoch.
:ivar importance: The message's priority level between 1 and 5, where 1 is
least important and 5 is most.
:ivar service: The :class:`~pushjet.Service` that sent the message.
"""
def __repr__(self):
return "<Pushjet Message: \"{}\">".format(repr_format(self.title or self.message))
def __init__(self, message_dict):
self.message = message_dict['message']
self.title = message_dict['title'] or None
self.link = message_dict['link'] or None
self.time_sent = message_dict['timestamp']
self.importance = message_dict['level']
self.service = Service._from_data(message_dict['service'])
class Api(object):
"""An API with a custom URL. Use this if you're connecting to a self-hosted
Pushjet API instance, or a non-standard one in general.
:param url: The URL to the API instance.
:ivar url: The URL to the API instance, as supplied.
"""
def __repr__(self):
return "<Pushjet Api: {}>".format(self.url).encode(sys.stdout.encoding, errors='replace')
def __init__(self, url):
self.url = text_type(url)
self.Service = with_api_bound(Service, self)
self.Device = with_api_bound(Device, self)
def _request(self, endpoint, method, params=None, data=None):
url = urljoin(self.url, endpoint)
try:
r = requests.request(method, url, params=params, data=data)
except requests.RequestException as e:
raise RequestError(e)
status = r.status_code
if status == requests.codes.INTERNAL_SERVER_ERROR:
raise ServerError(
"An error occurred in the server while processing your request. "
"This should probably be reported to: "
"https://github.com/Pushjet/Pushjet-Server-Api/issues"
)
try:
response = r.json()
except ValueError:
response = {}
return status, response

64
apprise/plugins/NotifyPushjet/pushjet/utilities.py
View File

@ -1,64 +0,0 @@
# -*- coding: utf-8 -*-
import re
import sys
from decorator import decorator
from .errors import AccessError
# Help class(...es? Nah. Just singular for now.)
class NoNoneDict(dict):
"""A dict that ignores values that are None. Not completely API-compatible
with dict, but contains all that's needed.
"""
def __repr__(self):
return "NoNoneDict({dict})".format(dict=dict.__repr__(self))
def __init__(self, initial={}):
self.update(initial)
def __setitem__(self, key, value):
if value is not None:
dict.__setitem__(self, key, value)
def update(self, data):
for key, value in data.items():
self[key] = value
# Decorators / factories
@decorator
def requires_secret_key(func, self, *args, **kwargs):
"""Raise an error if the method is called without a secret key."""
if self.secret_key is None:
raise AccessError("The Service doesn't have a secret "
"key provided, and therefore lacks write permission.")
return func(self, *args, **kwargs)
def with_api_bound(cls, api):
new_cls = type(cls.__name__, (cls,), {
'_api': api,
'__doc__': (
"Create a :class:`~pushjet.{name}` bound to the API. "
"See :class:`pushjet.{name}` for documentation."
).format(name=cls.__name__)
})
return new_cls
# Helper functions
UUID_RE = re.compile(r'^[A-Fa-f0-9]{8}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{12}$')
PUBLIC_KEY_RE = re.compile(r'^[A-Za-z0-9]{4}-[A-Za-z0-9]{6}-[A-Za-z0-9]{12}-[A-Za-z0-9]{5}-[A-Za-z0-9]{9}$')
SECRET_KEY_RE = re.compile(r'^[A-Za-z0-9]{32}$')
is_valid_uuid = lambda s: UUID_RE.match(s) is not None
is_valid_public_key = lambda s: PUBLIC_KEY_RE.match(s) is not None
is_valid_secret_key = lambda s: SECRET_KEY_RE.match(s) is not None
def repr_format(s):
s = s.replace('\n', ' ').replace('\r', '')
original_length = len(s)
s = s[:30]
s += '...' if len(s) != original_length else ''
s = s.encode(sys.stdout.encoding, errors='replace')
return s

6
apprise/plugins/__init__.py
View File

@ -33,9 +33,6 @@ from os.path import abspath
# Used for testing
from . import NotifyEmail as NotifyEmailBase
# Required until re-factored into base code
from .NotifyPushjet import pushjet
from .NotifyGrowl import gntp
# NotifyBase object is passed in as a module not class
@ -62,9 +59,6 @@ __all__ = [
# gntp (used for NotifyGrowl Testing)
'gntp',
# pushjet (used for NotifyPushjet Testing)
'pushjet',
]
# we mirror our base purely for the ability to reset everything; this

4
packaging/redhat/python-apprise.spec
View File

@ -75,7 +75,6 @@ Summary: A simple wrapper to many popular notification services used today
%{?python_provide:%python_provide python2-%{pypi_name}}
BuildRequires: python2-devel
BuildRequires: python-decorator
BuildRequires: python-requests
BuildRequires: python2-requests-oauthlib
BuildRequires: python-six
@ -89,7 +88,6 @@ BuildRequires: python2-babel
BuildRequires: python2-yaml
%endif # using rhel7
Requires: python-decorator
Requires: python-requests
Requires: python2-requests-oauthlib
Requires: python-six
@ -134,7 +132,6 @@ Summary: A simple wrapper to many popular notification services used today
%{?python_provide:%python_provide python%{python3_pkgversion}-%{pypi_name}}
BuildRequires: python%{python3_pkgversion}-devel
BuildRequires: python%{python3_pkgversion}-decorator
BuildRequires: python%{python3_pkgversion}-requests
BuildRequires: python%{python3_pkgversion}-requests-oauthlib
BuildRequires: python%{python3_pkgversion}-six
@ -142,7 +139,6 @@ BuildRequires: python%{python3_pkgversion}-click >= 5.0
BuildRequires: python%{python3_pkgversion}-markdown
BuildRequires: python%{python3_pkgversion}-yaml
BuildRequires: python%{python3_pkgversion}-babel
Requires: python%{python3_pkgversion}-decorator
Requires: python%{python3_pkgversion}-requests
Requires: python%{python3_pkgversion}-requests-oauthlib
Requires: python%{python3_pkgversion}-six

1
requirements.txt
View File

@ -1,4 +1,3 @@
decorator
requests
requests-oauthlib
six

2
setup.cfg
View File

@ -7,7 +7,7 @@ license_file = LICENSE
[flake8]
# We exclude packages we don't maintain
exclude = .eggs,.tox,gntp,pushjet
exclude = .eggs,.tox,gntp
ignore = E722,W503,W504
statistics = true
builtins = _

204
test/test_pushjet_plugin.py
View File

@ -1,204 +0,0 @@
# -*- 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 six
from apprise import plugins
from apprise import NotifyType
from apprise import Apprise
import mock
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
TEST_URLS = (
##################################
# NotifyPushjet
##################################
('pjet://', {
'instance': None,
}),
('pjets://', {
'instance': None,
}),
('pjet://:@/', {
'instance': None,
}),
# You must specify a username
('pjet://%s' % ('a' * 32), {
'instance': TypeError,
}),
# Specify your own server
('pjet://%s@localhost' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
}),
# Specify your own server with port
('pjets://%s@localhost:8080' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
}),
('pjet://%s@localhost:8081' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_notify_exceptions': True,
}),
)
@mock.patch('apprise.plugins.pushjet.pushjet.Service.send')
@mock.patch('apprise.plugins.pushjet.pushjet.Service.refresh')
def test_plugin(mock_refresh, mock_send):
"""
API: NotifyPushjet Plugin() (pt1)
"""
# iterate over our dictionary and test it out
for (url, meta) in TEST_URLS:
# Our expected instance
instance = meta.get('instance', None)
# Our expected server objects
self = meta.get('self', None)
# Our expected Query response (True, False, or exception type)
response = meta.get('response', True)
# Allow us to force the server response code to be something other then
# the defaults
response = meta.get(
'response', True if response else False)
test_notify_exceptions = meta.get(
'test_notify_exceptions', False)
test_exceptions = (
plugins.pushjet.errors.AccessError(
0, 'pushjet.AccessError() not handled'),
plugins.pushjet.errors.NonexistentError(
0, 'pushjet.NonexistentError() not handled'),
plugins.pushjet.errors.SubscriptionError(
0, 'gntp.SubscriptionError() not handled'),
plugins.pushjet.errors.RequestError(
'pushjet.RequestError() not handled'),
)
try:
obj = Apprise.instantiate(url, suppress_exceptions=False)
if obj is None:
if instance is not None:
# We're done (assuming this is what we were expecting)
print("{} didn't instantiate itself "
"(we expected it to)".format(url))
assert False
continue
if instance is None:
# Expected None but didn't get it
print('%s instantiated %s (but expected None)' % (
url, str(obj)))
assert(False)
assert(isinstance(obj, instance))
if isinstance(obj, plugins.NotifyBase):
# We loaded okay; now lets make sure we can reverse this url
assert(isinstance(obj.url(), six.string_types) is True)
# Instantiate the exact same object again using the URL from
# the one that was already created properly
obj_cmp = Apprise.instantiate(obj.url())
# Our object should be the same instance as what we had
# originally expected above.
if not isinstance(obj_cmp, plugins.NotifyBase):
# Assert messages are hard to trace back with the way
# these tests work. Just printing before throwing our
# assertion failure makes things easier to debug later on
print('TEST FAIL: {} regenerated as {}'.format(
url, obj.url()))
assert(False)
if self:
# Iterate over our expected entries inside of our object
for key, val in self.items():
# Test that our object has the desired key
assert(hasattr(key, obj))
assert(getattr(key, obj) == val)
try:
if test_notify_exceptions is False:
# Store our response
mock_send.return_value = response
mock_send.side_effect = None
# check that we're as expected
assert obj.notify(
title='test', body='body',
notify_type=NotifyType.INFO) == response
else:
for exception in test_exceptions:
mock_send.side_effect = exception
mock_send.return_value = None
try:
assert obj.notify(
title='test', body='body',
notify_type=NotifyType.INFO) is False
except AssertionError:
# Don't mess with these entries
raise
except Exception:
# We can't handle this exception type
raise
except AssertionError:
# Don't mess with these entries
print('%s AssertionError' % url)
raise
except Exception as e:
# Check that we were expecting this exception to happen
if not isinstance(e, response):
raise
except AssertionError:
# Don't mess with these entries
print('%s AssertionError' % url)
raise
except Exception as e:
# Handle our exception
if(instance is None):
raise
if not isinstance(e, instance):
raise

61
test/test_rest_plugins.py
View File

@ -1509,6 +1509,67 @@ TEST_URLS = (
'test_requests_exceptions': True,
}),
##################################
# NotifyPushjet
##################################
('pjet://', {
'instance': None,
}),
('pjets://', {
'instance': None,
}),
('pjet://:@/', {
'instance': None,
}),
# You must specify a secret key
('pjet://%s' % ('a' * 32), {
'instance': TypeError,
}),
# Legacy method of logging in (soon to be depricated)
('pjet://%s@localhost' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
}),
# The proper way to log in
('pjet://user:pass@localhost/%s' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
}),
# The proper way to log in
('pjets://localhost/%s' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
}),
# Specify your own server with login (secret= MUST be provided)
('pjet://user:pass@localhost?secret=%s' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
}),
# Specify your own server with login (no secret = fail normally)
# however this will work since we're providing depricated support
# at this time so the 'user' get's picked up as being the secret_key
('pjet://user:pass@localhost', {
'instance': plugins.NotifyPushjet,
}),
# Specify your own server with port
('pjets://localhost:8080/%s' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
}),
('pjets://localhost:8080/%s' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
}),
('pjets://localhost:4343/%s' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('pjet://localhost:8081/%s' % ('a' * 32), {
'instance': plugins.NotifyPushjet,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
##################################
# NotifyPushover
##################################