Commits

Jeffrey Goettsch  committed de68447

Added pushnotify/pushover.py. Can now verify user tokens and device strings for the Pushover notification system.

  • Participants
  • Parent commits 8948cbf

Comments (0)

Files changed (4)

 syntax: regexp
 (^|/)\..*
 ^tests/nmakeys.py
+^tests/pushoverkeys.py
 
 syntax: glob
 *.egg-info

File pushnotify/nma.py

 
         return root
 
+    def _post(self, url, data):
+
+        request = urllib2.Request(url, data)
+        response_stream = self._browser.open(request)
+        response = response_stream.read()
+
+        return response
+
     def _raise_exception(self):
 
         if self._last_code == '400':
             raise exceptions.UnknownError(self._last_message,
                                           int(self._last_code))
 
-    def _post(self, url, data):
-
-        request = urllib2.Request(url, data)
-        response_stream = self._browser.open(request)
-        response = response_stream.read()
-
-        return response
-
     def notify(self, app, event, desc, kwargs=None):
         """Send a notification to each apikey in self.apikeys.
 

File pushnotify/pushover.py

+#!/usr/bin/env python
+# vim: set fileencoding=utf-8
+
+"""Module for sending push notificiations to Android and iOS devices
+that have Pushover installed. See https://pushover.net/ for more
+information.
+
+copyright: Copyright (c) Jeffrey Goettsch and other contributors.
+license: BSD, see LICENSE for details.
+
+"""
+
+
+import json
+import urllib
+import urllib2
+
+from pushnotify import exceptions
+
+
+PUBLIC_API_URL = u'https://api.pushover.net/1'
+VERIFY_URL = u'/'.join([PUBLIC_API_URL, u'users/validate.json'])
+NOTIFY_URL = u'/'.join([PUBLIC_API_URL, u'messages.json'])
+
+
+class Client(object):
+    """Client for sending push notifications to Android and iOS devices
+    with the Pushover application installed.
+
+    Member Vars:
+        token: A string containing a valid API token.
+
+    """
+
+    def __init__(self, token):
+        """Initialize the Pushover client.
+
+        Args:
+            token: A string containing a valid API token.
+
+        """
+
+        self._browser = urllib2.build_opener(urllib2.HTTPSHandler())
+
+        self.token = token
+
+    def _parse_response(self, stream, verify=False):
+
+        response = json.loads(stream.read())
+
+        self._last_code = stream.code
+        if 'user' in response.keys():
+            self._last_user = response['user']
+        else:
+            self._last_user = None
+        if 'status' in response.keys():
+            self._last_status = response['status']
+        else:
+            self._last_status = None
+        if 'device' in response.keys():
+            self._last_device = response['device']
+        else:
+            self._last_device = None
+
+    def _post(self, url, data):
+
+        request = urllib2.Request(url, data)
+        try:
+            response_stream = self._browser.open(request)
+        except urllib2.HTTPError, exc:
+            return exc
+        else:
+            return response_stream
+
+    def verify_user(self, user):
+        """Verify a user token.
+
+        Args:
+            user: A string containing a valid user token.
+
+        Returns:
+            A boolean containing True if the user token is valid, and
+            False if it is not.
+
+        """
+
+        data = {'token': self.token, 'user': user}
+
+        data = urllib.urlencode(data)
+        response_stream = self._post(VERIFY_URL, data)
+
+        self._parse_response(response_stream, True)
+
+        return self._last_status
+
+    def verify_device(self, user, device):
+        """Verify a device for a user.
+
+        Args:
+            user: A string containing a valid user token.
+            device: A string containing a device name.
+
+        Raises:
+            pushnotify.exceptions.ApiKeyError
+
+        Returns:
+            A boolean containing True if the device is valid, and
+            False if it is not.
+
+        """
+
+        data = {'token': self.token, 'user': user, 'device': device}
+
+        data = urllib.urlencode(data)
+        response_stream = self._post(VERIFY_URL, data)
+
+        self._parse_response(response_stream, True)
+
+        if self._last_user and 'invalid' in self._last_user.lower():
+            raise exceptions.ApiKeyError(self._last_user, self._last_code)
+
+        return self._last_status
+
+
+if __name__ == '__main__':
+    pass

File tests/tests.py

 
 from pushnotify import exceptions
 from pushnotify import nma
+from pushnotify import pushover
+
 try:
     imp.find_module('nmakeys', [os.path.dirname(__file__)])
 except ImportError:
-    API_KEYS = {}
-    DEVELOEPER_KEY = ''
+    NMA_API_KEYS = {}
+    NMA_DEVELOPER_KEY = ''
 else:
-    from nmakeys import API_KEYS
-    from nmakeys import DEVELOPER_KEY
+    from nmakeys import API_KEYS as NMA_API_KEYS
+    from nmakeys import DEVELOPER_KEY as NMA_DEVELOPER_KEY
+
+try:
+    imp.find_module('pushoverkeys', [os.path.dirname(__file__)])
+except ImportError:
+    PUSHOVER_TOKEN = ''
+    PUSHOVER_USER = ''
+    PUSHOVER_DEVICE = ''
+else:
+    from pushoverkeys import TOKEN as PUSHOVER_TOKEN
+    from pushoverkeys import USER as PUSHOVER_USER
+    from pushoverkeys import DEVICE as PUSHOVER_DEVICE
 
 
 class NMATest(unittest.TestCase):
 
     def setUp(self):
 
-        self.client = nma.Client(API_KEYS, DEVELOPER_KEY)
+        self.client = nma.Client(NMA_API_KEYS, NMA_DEVELOPER_KEY)
 
-    def test_notify(self):
+        self.app = 'pushnotify unit tests'
+        self.event = 'unit test: test_notify'
+        self.desc = 'valid notification test for pushnotify'
+
+    def test_notify_valid(self):
+        """Test notify with valid notifications.
+
+        """
 
         """valid notification"""
 
-        app = 'pushnotify unit tests'
-        event = 'unit test: test_notify'
-        desc = 'valid notification test for pushnotify'
-
-        self.client.notify(app, event, desc)
+        self.client.notify(self.app, self.event, self.desc)
 
         """valid notification, extra arguments, html"""
 
-        html_desc = '<h1>{0}</h1><p>{1}<br>{2}</p>'.format(app, event, desc)
+        html_desc = '<h1>{0}</h1><p>{1}<br>{2}</p>'.format(
+            self.app, self.event, self.desc)
         priority = 0
         url = nma.NOTIFY_URL
 
-        self.client.notify(app, event, html_desc,
+        self.client.notify(self.app, self.event, html_desc,
                            kwargs={'priority': priority, 'url': url,
                                    'content-type': 'text/html'})
 
+    def test_notify_invalid(self):
+        """Test notify with invalid notifications.
+
+        """
+
         """invalid API key"""
 
         char = self.client.apikeys[0][0]
         self.client.developerkey = ''
 
         self.assertRaises(exceptions.ApiKeyError,
-                          self.client.notify, app, event, desc)
+                          self.client.notify, self.app, self.event, self.desc)
 
-        self.client.apikeys = API_KEYS
-        self.client.developerkey = DEVELOPER_KEY
+        self.client.apikeys = NMA_API_KEYS
+        self.client.developerkey = NMA_DEVELOPER_KEY
 
-        """invalid argument length"""
+        """invalid argument lengths"""
 
         bad_app = 'a' * 257
+        self.assertRaises(exceptions.FormatError,
+                          self.client.notify, bad_app, self.event, self.desc)
 
-        self.assertRaises(exceptions.FormatError,
-                          self.client.notify, bad_app, event, desc)
+    def test_verify_valid(self):
+        """Test verify with a valid API key.
 
-        bad_event = 'e' * 1001
-
-        self.assertRaises(exceptions.FormatError,
-                          self.client.notify, app, bad_event, desc)
-
-        bad_desc = 'd' * 10001
-
-        self.assertRaises(exceptions.FormatError,
-                          self.client.notify, app, event, bad_desc)
-
-    def test_verify(self):
-
-        """valid API key"""
+        """
 
         self.assertTrue(self.client.verify(self.client.apikeys[0]))
 
+    def test_verify_invalid(self):
+        """Test verify with invalid API keys.
+
+        """
+
         """invalid API key of incorrect length"""
 
         apikey = u'{0}{1}'.format(self.client.apikeys[0], '1')
         self.assertFalse(self.client.verify(apikey))
 
 
+class PushoverTest(unittest.TestCase):
+
+    def setUp(self):
+
+        self.client = pushover.Client(PUSHOVER_TOKEN)
+
+    def test_verify_user_valid(self):
+        """Test veriy_user with a valid user token.
+
+        """
+
+        self.assertTrue(self.client.verify_user(PUSHOVER_USER))
+
+    def test_verify_user_invalid(self):
+        """Test verify_user with an invalid user token.
+
+        """
+
+        self.assertFalse(self.client.verify_user('foo'))
+
+    def test_verify_device_valid(self):
+        """Test verify_device with a valid device string.
+
+        """
+
+        self.assertTrue(self.client.verify_device(PUSHOVER_USER,
+                                                  PUSHOVER_DEVICE))
+
+    def test_verify_device_invalid(self):
+        """Test verify_device with an invalid device string.
+
+        """
+
+        self.assertFalse(self.client.verify_device(PUSHOVER_USER, 'foo'))
+
+    def test_verify_device_invalid_user(self):
+        """Test verify_device with an invalid user token.
+
+        """
+
+        self.assertRaises(exceptions.ApiKeyError, self.client.verify_device,
+                          'foo', PUSHOVER_DEVICE)
+
+
 if __name__ == '__main__':
     pass