Commits

Jeffrey Goettsch committed 87bbfc9

pushover client now works with the Abstract interface.

Comments (0)

Files changed (2)

pushnotify/pushover.py

 
 
 import json
-import logging
 import time
-import urllib
-import urllib2
 
+from pushnotify import abstract
 from pushnotify import exceptions
 
 
 VERIFY_URL = u'/'.join([PUBLIC_API_URL, u'users/validate.json'])
 NOTIFY_URL = u'/'.join([PUBLIC_API_URL, u'messages.json'])
 
+DESC_LIMIT = 512
 
-class Client(object):
+
+class Client(abstract.AbstractClient):
     """Client for sending push notifications to Android and iOS devices
     with the Pushover application installed.
 
 
     """
 
-    def __init__(self, token, users=None):
+    def __init__(self, developerkey='', application=''):
         """Initialize the Pushover client.
 
         Args:
-            token: A string of 30 characters containing a valid API
-                token.
-            users: A list containing 1 or 2-item tuples, where the first
-                item is a string of 30 characters containing a user
-                token, and the second is an optional string of up to 25
-                characters containing a device name for the given user.
-                (default: None)
+            developerkey: A string containing a valid provider key for
+                the Prowl client.
+            application: A string containing the name of the application
+                on behalf of whom the Prowl client will be sending
+                messages.
 
         """
 
-        self.logger = logging.getLogger('{0}.{1}'.format(
-            self.__module__, self.__class__.__name__))
+        super(self.__class__, self).__init__(developerkey, application)
 
-        self._browser = urllib2.build_opener(urllib2.HTTPSHandler())
-        self._last_code = None
-        self._last_device = None
-        self._last_errors = None
-        self._last_status = None
-        self._last_token = None
-        self._last_user = None
+        self._type = 'nma'
+        self._urls = {'notify': NOTIFY_URL, 'verify': VERIFY_URL}
 
-        self.token = token
-        self.users = [] if users is None else users
-
-    def _parse_response(self, stream, verify=False):
+    def _parse_response_stream(self, stream, verify=False):
 
         response = stream.read()
         self.logger.info('received response: {0}'.format(response))
 
         response = json.loads(response)
 
-        self._last_code = stream.code
+        self._last['code'] = stream.code
         if 'device' in response.keys():
-            self._last_device = response['device']
+            self._last['device'] = response['device']
         else:
-            self._last_device = None
+            self._last['device'] = None
         if 'errors' in response.keys():
-            self._last_errors = response['errors']
+            self._last['errors'] = response['errors']
         else:
-            self._last_errors = None
+            self._last['errors'] = None
         if 'status' in response.keys():
-            self._last_status = response['status']
+            self._last['status'] = response['status']
         else:
-            self._last_status = None
+            self._last['status'] = None
         if 'token' in response.keys():
-            self._last_token = response['token']
+            self._last['token'] = response['token']
+        else:
+            self._last['token'] = None
         if 'user' in response.keys():
-            self._last_user = response['user']
+            self._last['user'] = response['user']
         else:
-            self._last_user = None
+            self._last['user'] = None
 
-        return self._last_status
-
-    def _post(self, url, data):
-
-        self.logger.debug('_post sending data: {0}'.format(data))
-        self.logger.debug('_post sending to url: {0}'.format(url))
-
-        request = urllib2.Request(url, data)
-        try:
-            response_stream = self._browser.open(request)
-        except urllib2.HTTPError, exc:
-            return exc
-        else:
-            return response_stream
+        return self._last['status']
 
     def _raise_exception(self):
 
         msg = ''
-        if self._last_errors:
+        if self._last['errors']:
             messages = []
-            for key, value in self._last_errors.items():
+            for key, value in self._last['errors'].items():
                 messages.append('{0} {1}'.format(key, value[0]))
             msg = '; '.join(messages)
 
-        if self._last_device and 'invalid' in self._last_device:
-            raise exceptions.ApiKeyError('device invalid', self._last_code)
+        if self._last['device'] and 'invalid' in self._last['device']:
+            raise exceptions.ApiKeyError('device invalid', self._last['code'])
 
-        elif self._last_token and 'invalid' in self._last_token:
-            raise exceptions.ApiKeyError('token invalid', self._last_code)
+        elif self._last['token'] and 'invalid' in self._last['token']:
+            raise exceptions.ApiKeyError('token invalid', self._last['code'])
 
-        elif self._last_user and 'invalid' in self._last_user:
-            raise exceptions.ApiKeyError('user invalid', self._last_code)
+        elif self._last['user'] and 'invalid' in self._last['user']:
+            raise exceptions.ApiKeyError('user invalid', self._last['code'])
 
-        elif self._last_code == 429:
+        elif self._last['code'] == 429:
             # TODO: what is actually returned when the rate limit is hit?
 
             msg = 'too many messages sent this month' if not msg else msg
-            raise exceptions.RateLimitExceeded(msg, self._last_code)
+            raise exceptions.RateLimitExceeded(msg, self._last['code'])
 
-        elif self._last_code >= 500 and self._last_code <= 599:
-            raise exceptions.ServerError(msg, self._last_code)
+        elif self._last['code'] >= 500 and self._last['code'] <= 599:
+            raise exceptions.ServerError(msg, self._last['code'])
 
-        elif self._last_errors:
-            raise exceptions.FormatError(msg, self._last_code)
+        elif self._last['errors']:
+            raise exceptions.FormatError(msg, self._last['code'])
 
         else:
-            raise exceptions.UnrecognizedResponseError(msg, self._last_code)
+            raise exceptions.UnrecognizedResponseError(msg, self._last['code'])
 
-    def notify(self, title, message, kwargs=None):
+    def notify(self, description, event, split=True, kwargs=None):
         """Send a notification to each user/device in self.users.
 
         As of 2012-09-18, this is not returning a 4xx status code as
 
         """
 
-        """Here we match the behavior of Notify My Android and Prowl:
-        raise a single exception if and only if every notification
-        fails"""
-
-        raise_exception = False
-
-        if not self.users:
-            self.logger.warn('notify called with no users set')
-
-        for user in self.users:
-            data = {'token': self.token,
-                    'user': user[0],
-                    'title': title,
-                    'message': message,
+        def send_notify(desc_list, event, kwargs, apikey, device_key=''):
+            data = {'token': self.developerkey,
+                    'user': apikey,
+                    'title': event,
+                    'message': description,
                     'timestamp': int(time.time())}
 
-            if user[1]:
-                data['device'] = user[1]
+            if device_key:
+                data['device'] = device_key
 
             if kwargs:
                 data.update(kwargs)
 
-            data = urllib.urlencode(data)
+            response = self._post(self._urls['notify'], data)
+            status = self._parse_response_stream(response)
 
-            response = self._post(NOTIFY_URL, data)
-            status = self._parse_response(response)
-            if not status:
-                raise_exception = not status
+            return not status
+
+        if not self.apikeys:
+            self.logger.warn('notify called with no users set')
+
+        desc_list = []
+        if split:
+            while description:
+                desc_list.append(description[0:DESC_LIMIT])
+                description = description[DESC_LIMIT:]
+        else:
+            desc_list = [description]
+
+        # Here we match the behavior of Notify My Android and Prowl:
+        # raise a single exception if and only if every notification
+        # fails
+
+        raise_exception = True
+
+        for apikey, device_keys in self.apikeys.items():
+            if not device_keys:
+                this_raise_exception = send_notify(desc_list, event, kwargs,
+                                                   apikey)
+            else:
+                for device_key in device_keys:
+                    this_raise_exception = send_notify(
+                        desc_list, event, kwargs, apikey, device_key)
+
+            raise_exception = raise_exception and this_raise_exception
 
         if raise_exception:
             self._raise_exception()
 
-    def verify_user(self, user):
+    def verify_user(self, apikey):
         """Verify a user token.
 
         Args:
-            user: A string containing a valid user token.
+            apikey: A string containing a user's token.
 
         Returns:
-            A boolean containing True if the user token is valid, and
+            A boolean containing True if apikey is valid, and
             False if it is not.
 
         """
 
-        data = {'token': self.token, 'user': user}
+        data = {'token': self.developerkey, 'user': apikey}
 
-        data = urllib.urlencode(data)
-        response_stream = self._post(VERIFY_URL, data)
+        response_stream = self._post(self._urls['verify'], data)
 
-        self._parse_response(response_stream, True)
+        self._parse_response_stream(response_stream, True)
 
-        return self._last_status
+        return self._last['status']
 
-    def verify_device(self, user, device):
+    def verify_device(self, apikey, device_key):
         """Verify a device for a user.
 
         Args:
-            user: A string containing a valid user token.
-            device: A string containing a device name.
+            apikey: A string containing a user's token.
+            device_key: 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.
+            A boolean containing True if device_key is valid for
+            apikey, and False if it is not.
 
         """
 
-        data = {'token': self.token, 'user': user, 'device': device}
+        data = {'token': self.developerkey, 'user': apikey,
+                'device': device_key}
 
-        data = urllib.urlencode(data)
-        response_stream = self._post(VERIFY_URL, data)
+        response_stream = self._post(self._urls['verify'], data)
 
-        self._parse_response(response_stream, True)
+        self._parse_response_stream(response_stream, True)
 
-        if self._last_user and 'invalid' in self._last_user.lower():
+        if self._last['user'] and 'invalid' in self._last['user'].lower():
             self._raise_exception()
 
-        return self._last_status
+        return self._last['status']
 
 
 if __name__ == '__main__':

pushnotify/tests/tests.py

 from pushnotify import get_client
 from pushnotify import exceptions
 from pushnotify import nma
-from pushnotify import pushover
 
 try:
     imp.find_module('nmakeys', [os.path.dirname(__file__)])
 else:
     from pushnotify.tests.pushoverkeys import TOKEN as PUSHOVER_TOKEN
     from pushnotify.tests.pushoverkeys import USER as PUSHOVER_USER
-    from pushnotify.tests.pushoverkeys import DEVICE as PUSHOVER_DEVICE
 
 
 class NMATest(unittest.TestCase):
 
     def setUp(self):
 
-        self.client = pushover.Client(PUSHOVER_TOKEN,
-                                      [(PUSHOVER_USER, PUSHOVER_DEVICE)])
+        self.client = get_client('pushover', PUSHOVER_TOKEN, '')
 
-        self.title = 'pushnotify unit tests'
-        self.message = 'valid notification test for pushnotify'
+        for key in PUSHOVER_USER.keys():
+            self.client.add_key(key, PUSHOVER_USER[key][0])
+
+        self.event = 'pushnotify unit tests'
+        self.desc = 'valid notification test for pushnotify'
 
     def test_notify_valid(self):
         """Test notify with a valid notification.
 
         """
 
-        self.client.notify(self.title, self.message,
+        self.client.notify(self.desc, self.event, split=False,
                            kwargs={'priority': 1, 'url': 'http://google.com/',
                                    'url_title': 'Google'})
 
-    def test_notify_invalid_token(self):
-        """Test notify with an invalid token.
+    def test_notify_invalid_developerkey(self):
+        """Test notify with an invalid developer key.
 
         """
 
-        char = self.client.token[0]
-        bad_token = self.client.token.replace(char, '_')
-        self.client.token = bad_token
+        self.client.developerkey = '_' + self.client.developerkey[1:]
 
         self.assertRaises(exceptions.ApiKeyError, self.client.notify,
-                          self.title, self.message)
+                          self.desc, self.event)
 
-    def test_notify_invalid_user(self):
-        """Test notify with an invalid user.
+    def test_notify_invalid_apikey(self):
+        """Test notify with an invalid API key.
 
         """
 
-        char = self.client.users[0][0][0]
-        bad_users = (self.client.users[0][0].replace(char, '_'),
-                     PUSHOVER_DEVICE)
-        self.client.users = bad_users
+        apikey = self.client.apikeys.keys()[0]
+        device_key = self.client.apikeys[apikey][0]
+
+        apikey = '_' + apikey[1:]
+
+        self.client.apikeys = {}
+        self.client.add_key(apikey, device_key)
 
         self.assertRaises(exceptions.ApiKeyError, self.client.notify,
-                          self.title, self.message)
+                          self.desc, self.event)
 
-    def test_notify_invalid_device(self):
-        """Test notify with an invalid device.
+    def test_notify_invalid_device_key(self):
+        """Test notify with an invalid device key.
 
         """
 
-        char = self.client.users[0][1][0]
-        bad_users = (PUSHOVER_USER, self.client.users[0][1].replace(char, '_'))
-        self.client.users = bad_users
+        apikey = self.client.apikeys.keys()[0]
+
+        self.client.apikeys = {}
+        self.client.add_key(apikey, 'foo')
 
         self.assertRaises(exceptions.ApiKeyError, self.client.notify,
-                          self.title, self.message)
+                          self.desc, self.event)
 
     def test_notify_invalid_args(self):
         """Test notify with invalid argument lengths.
         # per the Pushover API docs, but instead chopping the delivered
         # messages off at 512 characters
 
-        msg = 'a' * 513
+        desc = 'a' * 513
 
         try:
-            self.client.notify(self.title, msg)
+            self.client.notify(desc, self.event, False)
         except exceptions.FormatError:
             pass
 
     def test_verify_user_valid(self):
-        """Test veriy_user with a valid user token.
+        """Test verify_user with a valid user token.
 
         """
 
-        self.assertTrue(self.client.verify_user(PUSHOVER_USER))
+        self.assertTrue(self.client.verify_user(self.client.apikeys.keys()[0]))
 
-    def test_verify_user_invalid(self):
-        """Test verify_user with an invalid user token.
+    def test_verify_apikey_invalid(self):
+        """Test verify_user with an invalid API key.
 
         """
 
         self.assertFalse(self.client.verify_user('foo'))
 
     def test_verify_device_valid(self):
-        """Test verify_device with a valid device string.
+        """Test verify_device with a valid device key.
 
         """
 
-        self.assertTrue(self.client.verify_device(PUSHOVER_USER,
-                                                  PUSHOVER_DEVICE))
+        apikey = self.client.apikeys.keys()[0]
+        device_key = self.client.apikeys[apikey][0]
+
+        self.assertTrue(self.client.verify_device(apikey, device_key))
 
     def test_verify_device_invalid(self):
-        """Test verify_device with an invalid device string.
+        """Test verify_device with an invalid device key.
 
         """
 
-        self.assertFalse(self.client.verify_device(PUSHOVER_USER, 'foo'))
+        apikey = self.client.apikeys.keys()[0]
 
-    def test_verify_device_invalid_user(self):
-        """Test verify_device with an invalid user token.
+        self.assertFalse(self.client.verify_device(apikey, 'foo'))
+
+    def test_verify_device_invalid_apikey(self):
+        """Test verify_device with an invalid API key.
 
         """
 
+        apikey = self.client.apikeys.keys()[0]
+        device_key = self.client.apikeys[apikey][0]
+
         self.assertRaises(exceptions.ApiKeyError, self.client.verify_device,
-                          'foo', PUSHOVER_DEVICE)
+                          'foo', device_key)
 
 
 if __name__ == '__main__':
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.