diff --git a/.travis.yml b/.travis.yml index 512db9c..f0db249 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,11 +1,14 @@ language: python python: - "2.7" - - "3.3" - "3.4" - "3.5" - - "3.5-dev" # 3.5 development branch - - "nightly" # currently points to 3.6-dev + - "3.5" + - "3.5-dev" + - "3.6" + - "3.6-dev" + - "3.7-dev" + - "nightly" # command to install dependencies install: - pip install -r requirements.txt diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..dd85ab7 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,9 @@ +{ + "python.unitTest.pyTestArgs": [ + "tests" + ], + "python.unitTest.unittestEnabled": false, + "python.unitTest.nosetestsEnabled": false, + "python.unitTest.pyTestEnabled": true, + "python.pythonPath": "/anaconda3/envs/pypo_dev/bin/python" +} \ No newline at end of file diff --git a/pypushover/_base.py b/pypushover/_base.py index 2eabc9e..26c1a6e 100644 --- a/pypushover/_base.py +++ b/pypushover/_base.py @@ -51,16 +51,9 @@ def send(url, data_out=None, get_method=False): if ret_dict['status'] == 0: raise PushoverError(ret_dict['errors']) - if 'X-Limit-App-Limit' in res.headers: - ret_dict['app_limit'] = res.headers['X-Limit-App-Limit'] - if 'X-Limit-App-Remaining' in res.headers: - ret_dict['app_remaining'] = res.headers['X-Limit-App-Remaining'] - if 'X-Limit-App-Reset' in res.headers: - ret_dict['app_reset'] = res.headers['X-Limit-App-Reset'] - return ret_dict - except decode_error as e: + except decode_error: res.raise_for_status() diff --git a/pypushover/client.py b/pypushover/client.py index 044099b..ae001ff 100644 --- a/pypushover/client.py +++ b/pypushover/client.py @@ -37,7 +37,7 @@ stored on the Pushover servers are then stored into the `messages` property. These messages are a list of dictionaries with items as [defined in the Pushover API](https://pushover.net/api/client#download). - >>> cm.retrieve_message() + >>> cm.retrieve_messages() >>> for msg in cm.messages: ... print(msg['message']) @@ -55,7 +55,7 @@ API guidelines](https://pushover.net/api/client#p2). Once the user has acknowledged the message, using the `acknowledge_message` method passing in the emergency messages `receipt`. - >>> cm.retrieve_message() + >>> cm.retrieve_messages() >>> for msg in cm.messages: ... print(msg['message']) ... if msg['priority'] == py_po.PRIORITIES.EMERGENCY: @@ -166,7 +166,7 @@ def register_device(self, name): self.__device_id__ = self.latest_response_dict['id'] return self.__device_id__ - def retrieve_message(self): + def retrieve_messages(self): """ Retrieves messages stored on the Pushover servers and saves them into the `messages` property. """ @@ -189,6 +189,7 @@ def clear_server_messages(self): } self.latest_response_dict = send(self._del_message_url.format(device_id=self.__device_id__), params) + self.messages = [] def acknowledge_message(self, receipt): """ @@ -261,7 +262,7 @@ def _on_ws_message(self, ws, message): pass elif message == "!": - self.retrieve_message() + self.retrieve_messages() if self.__on_msg_receipt__: self.__on_msg_receipt__(self.messages) diff --git a/pypushover/groups.py b/pypushover/groups.py index d251d71..83a6f5a 100644 --- a/pypushover/groups.py +++ b/pypushover/groups.py @@ -112,6 +112,7 @@ class _User(object): This class is generated dynamically based on the response from the Pushover servers. """ + def __init__(self, **kwargs): for k, v in kwargs.items(): setattr(self, k + '_key' if k == 'user' else k, v) @@ -123,6 +124,7 @@ class _Group(object): This class is generated dynamically based on the response from the Pushover servers. """ + def __init__(self, **kwargs): self.users = None for k, v in kwargs.items(): @@ -173,7 +175,8 @@ def add_user(self, user, device=None, memo=None): :return: A dictionary representing the json response. """ - self.latest_response_dict = add_user(self._app_token, self._group_key, user, device=device, memo=memo) + self.latest_response_dict = add_user( + self._app_token, self._group_key, user, device=device, memo=memo) self.__update_group() return self.latest_response_dict @@ -184,7 +187,8 @@ def remove_user(self, user): :return: A dictionary representing the json response. """ - self.latest_response_dict = remove_user(self._app_token, self._group_key, user) + self.latest_response_dict = remove_user( + self._app_token, self._group_key, user) self.__update_group() return self.latest_response_dict @@ -195,7 +199,8 @@ def disable_user(self, user): :return: A dictionary representing the json response. """ - self.latest_response_dict = disable_user(self._app_token, self._group_key, user) + self.latest_response_dict = disable_user( + self._app_token, self._group_key, user) self.__update_group() return self.latest_response_dict @@ -206,7 +211,8 @@ def enable_user(self, user): :return: A dictionary representing the json response. """ - self.latest_response_dict = enable_user(self._app_token, self._group_key, user) + self.latest_response_dict = enable_user( + self._app_token, self._group_key, user) self.__update_group() return self.latest_response_dict @@ -217,7 +223,8 @@ def rename(self, name): :return: A dictionary representing the json response. """ - self.latest_response_dict = rename(self._app_token, self._group_key, name) + self.latest_response_dict = rename( + self._app_token, self._group_key, name) self.__update_group() return self.latest_response_dict diff --git a/pypushover/message.py b/pypushover/message.py index 69d2c84..b02018b 100644 --- a/pypushover/message.py +++ b/pypushover/message.py @@ -50,7 +50,7 @@ Below is an example: - >>> res = pm.send_message('Emergency Message!', priority=pypo.PRIORITIES.EMERGENCY, retry=30, expire=3600) + >>> res = pm.push_message('Emergency Message!', priority=pypo.PRIORITIES.EMERGENCY, retry=30, expire=3600) >>> res = pypo.message.push_message( ... '', ... '', @@ -88,19 +88,19 @@ """ import time -from pypushover import PRIORITIES, BaseManager, base_url, send +from pypushover import PRIORITIES, BaseManager as _BaseManager, base_url as _base_url, send as _send _MAX_EXPIRE = 86400 _MIN_RETRY = 30 -_push_url = base_url + "messages.json" -_base_receipt_url = base_url + "receipts/{receipt}" +_push_url = _base_url + "messages.json" +_base_receipt_url = _base_url + "receipts/{receipt}" _receipt_url = _base_receipt_url + ".json" _cancel_receipt_url = _base_receipt_url + "/cancel.json" -class MessageManager(BaseManager): +class MessageManager(_BaseManager): """ Manager class used to send messages and check receipts. Stores the given app_token for future use. Also stores the latest response from the API. @@ -277,7 +277,7 @@ def push_message(token, user, message, **kwargs): if 'html' in kwargs: data_out['html'] = int(kwargs['html']) - return send(_push_url, data_out=data_out) + return _send(_push_url, data_out=data_out) def check_receipt(token, receipt): @@ -289,7 +289,7 @@ def check_receipt(token, receipt): :return: """ url_to_send = _receipt_url.format(receipt=receipt) - return send(url_to_send, data_out={'token': token}, get_method=True) + return _send(url_to_send, data_out={'token': token}, get_method=True) def cancel_retries(token, receipt): @@ -301,5 +301,5 @@ def cancel_retries(token, receipt): :param str receipt: receipt of the message """ url_to_send = _cancel_receipt_url.format(receipt=receipt) - return send(url_to_send, data_out={'token': token}) + return _send(url_to_send, data_out={'token': token}) diff --git a/requirements.txt b/requirements.txt index feb31e7..345d5eb 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,5 @@ requests coveralls -websocket-client \ No newline at end of file +websocket-client +pytest +mock \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py index c004c08..ebb7764 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,18 +1,30 @@ import unittest +# The token and user key are taken from the online Pushover docs to help +# make this more real and won't work in real life. +APP_TOKEN = "KzGDORePKggMaC0QOYAMyEEuzJnyUi;" +USER_KEY = "e9e1495ec75826de5983cd1abc8031" def get_tests(): return full_suite() def full_suite(): - from .runtests import TestBasic, TestClient, TestGroup, TestLicense, TestMessage, TestVerifcation, TestIssues + from .test_Basic import TestBasic + from .test_Client import TestClient + from .test_Groups import TestGroup + from .test_Issues import TestIssues + from .test_Messages import TestMessages + from .test_Verification import TestVerification return unittest.TestSuite([ unittest.TestLoader().loadTestsFromTestCase(TestBasic), - unittest.TestLoader().loadTestsFromTestCase(TestClient), + # unittest.TestLoader().loadTestsFromTestCase(TestClient), unittest.TestLoader().loadTestsFromTestCase(TestGroup), - unittest.TestLoader().loadTestsFromTestCase(TestMessage), - unittest.TestLoader().loadTestsFromTestCase(TestVerifcation), - unittest.TestLoader().loadTestsFromTestCase(TestIssues) + unittest.TestLoader().loadTestsFromTestCase(TestMessages), + unittest.TestLoader().loadTestsFromTestCase(TestVerification), + # unittest.TestLoader().loadTestsFromTestCase(TestIssues) ]) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/helpers/__init__.py b/tests/helpers/__init__.py deleted file mode 100644 index 369c482..0000000 --- a/tests/helpers/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from tests.helpers.keys import user_key, group_key, app_key, device_id - -__all__ = ['user_key', 'group_key', 'app_key', 'secret', 'device_id'] \ No newline at end of file diff --git a/tests/responses/test_bad_verify_user_response.json b/tests/responses/test_bad_verify_user_response.json new file mode 100644 index 0000000..9b1e757 --- /dev/null +++ b/tests/responses/test_bad_verify_user_response.json @@ -0,0 +1,31 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "774ad663-d919-4539-b237-e0c589c2a837", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "user": "invalid", + "errors": ["user key is invalid"], + "status": 0, + "request": "774ad663-d919-4539-b237-e0c589c2a837" + } +} \ No newline at end of file diff --git a/tests/responses/test_good_cancel_response.json b/tests/responses/test_good_cancel_response.json new file mode 100644 index 0000000..146c7b8 --- /dev/null +++ b/tests/responses/test_good_cancel_response.json @@ -0,0 +1,29 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "855cef21-654d-4b4c-8235-824658b464dc", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "status": 1, + "request": "855cef21-654d-4b4c-8235-824658b464dc" + } +} \ No newline at end of file diff --git a/tests/responses/test_good_check_receipt_response.json b/tests/responses/test_good_check_receipt_response.json new file mode 100644 index 0000000..e0d6e03 --- /dev/null +++ b/tests/responses/test_good_check_receipt_response.json @@ -0,0 +1,37 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "73d54061-072a-4004-8190-a1991f2df844", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "status": 1, + "acknowledged": 1, + "acknowledged_at": 1541103706, + "acknowledged_by": "e9e1495ec75826de5983cd1abc8031", + "acknowledged_by_device": "krono-jpl", + "last_delivered_at": 1541103690, + "expired": 0, + "expires_at": 1541106308, + "called_back": 0, + "called_back_at": 0, + "request": "73d54061-072a-4004-8190-a1991f2df844"} +} \ No newline at end of file diff --git a/tests/responses/test_good_emergency_response.json b/tests/responses/test_good_emergency_response.json new file mode 100644 index 0000000..ea0b44c --- /dev/null +++ b/tests/responses/test_good_emergency_response.json @@ -0,0 +1,30 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "cc1cbb0a-f59c-4c20-9191-1578849e7150", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "receipt": "rbp115t2ohpuwh27gaa1jqips9y8by", + "status": 1, + "request": "cc1cbb0a-f59c-4c20-9191-1578849e7150" + } +} \ No newline at end of file diff --git a/tests/responses/test_good_group_info_add_user_response.json b/tests/responses/test_good_group_info_add_user_response.json new file mode 100644 index 0000000..c76206c --- /dev/null +++ b/tests/responses/test_good_group_info_add_user_response.json @@ -0,0 +1,37 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "f634e63c-253d-45bf-8d98-fa96a517c398", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "name": "KronoTestGroup", + "users": [ + { + "user": "e9e1495ec75826de5983cd1abc8031", + "device": "test_device", + "memo": "Added using UnitTests", + "disabled": false + } + ], + "status": 1, + "request": "f634e63c-253d-45bf-8d98-fa96a517c398"} +} \ No newline at end of file diff --git a/tests/responses/test_good_group_info_disabled_user_response.json b/tests/responses/test_good_group_info_disabled_user_response.json new file mode 100644 index 0000000..382dd08 --- /dev/null +++ b/tests/responses/test_good_group_info_disabled_user_response.json @@ -0,0 +1,37 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "f634e63c-253d-45bf-8d98-fa96a517c398", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "name": "KronoTestGroup", + "users": [ + { + "user": "e9e1495ec75826de5983cd1abc8031", + "device": "test_device", + "memo": "Added using UnitTests", + "disabled": true + } + ], + "status": 1, + "request": "f634e63c-253d-45bf-8d98-fa96a517c398"} +} \ No newline at end of file diff --git a/tests/responses/test_good_group_info_enabled_user_response.json b/tests/responses/test_good_group_info_enabled_user_response.json new file mode 100644 index 0000000..c76206c --- /dev/null +++ b/tests/responses/test_good_group_info_enabled_user_response.json @@ -0,0 +1,37 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "f634e63c-253d-45bf-8d98-fa96a517c398", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "name": "KronoTestGroup", + "users": [ + { + "user": "e9e1495ec75826de5983cd1abc8031", + "device": "test_device", + "memo": "Added using UnitTests", + "disabled": false + } + ], + "status": 1, + "request": "f634e63c-253d-45bf-8d98-fa96a517c398"} +} \ No newline at end of file diff --git a/tests/responses/test_good_group_info_rename_response.json b/tests/responses/test_good_group_info_rename_response.json new file mode 100644 index 0000000..675394e --- /dev/null +++ b/tests/responses/test_good_group_info_rename_response.json @@ -0,0 +1,30 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "f634e63c-253d-45bf-8d98-fa96a517c398", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "name": "KronoGroup", + "users": [ ], + "status": 1, + "request": "f634e63c-253d-45bf-8d98-fa96a517c398"} +} \ No newline at end of file diff --git a/tests/responses/test_good_group_info_response.json b/tests/responses/test_good_group_info_response.json new file mode 100644 index 0000000..95d65e1 --- /dev/null +++ b/tests/responses/test_good_group_info_response.json @@ -0,0 +1,30 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "f634e63c-253d-45bf-8d98-fa96a517c398", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "name": "KronoTestGroup", + "users": [ ], + "status": 1, + "request": "f634e63c-253d-45bf-8d98-fa96a517c398"} +} \ No newline at end of file diff --git a/tests/responses/test_good_response.json b/tests/responses/test_good_response.json new file mode 100644 index 0000000..969f8c5 --- /dev/null +++ b/tests/responses/test_good_response.json @@ -0,0 +1,29 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "f5f07044-ffe2-42ff-9d37-492650801155", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "status": 1, + "request": "f5f07044-ffe2-42ff-9d37-492650801155" + } +} \ No newline at end of file diff --git a/tests/responses/test_good_verify_group_response.json b/tests/responses/test_good_verify_group_response.json new file mode 100644 index 0000000..8d5deea --- /dev/null +++ b/tests/responses/test_good_verify_group_response.json @@ -0,0 +1,31 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "49112cae-da34-4ac6-9d0c-977c66e41c2c", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "status": 1, + "group": 1, + "devices": [], + "request": "49112cae-da34-4ac6-9d0c-977c66e41c2c" + } +} \ No newline at end of file diff --git a/tests/responses/test_good_verify_user_response.json b/tests/responses/test_good_verify_user_response.json new file mode 100644 index 0000000..662b035 --- /dev/null +++ b/tests/responses/test_good_verify_user_response.json @@ -0,0 +1,31 @@ +{ + "status_code": 200, + "headers": { + "Server": "nginx", + "Date": "Thu, 01 Nov 2018 16:43:21 GMT", + "Content-Type": "application/json; charset=utf-8", + "Transfer-Encoding": "chunked", + "Connection": "keep-alive", + "X-Frame-Options": "SAMEORIGIN, DENY", + "X-XSS-Protection": "1; mode=block", + "X-Content-Type-Options": "nosniff", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Methods": "POST, OPTIONS", + "Access-Control-Allow-Headers": "X-Requested-With, X-Prototype-Version, Origin, Accept, Content-Type, X-CSRF-Token, X-Pushover-App, Authorization", + "Access-Control-Max-Age": "1728000", + "X-Limit-App-Limit": "7500", + "X-Limit-App-Remaining": "7498", + "X-Limit-App-Reset": "1543644000", + "ETag": "W/\"2b287cc51514704e37448855a4c4e855\"", + "Cache-Control": "max-age=0, private, must-revalidate", + "X-Request-Id": "49112cae-da34-4ac6-9d0c-977c66e41c2c", + "X-Runtime": "0.040146", + "Strict-Transport-Security": "max-age=31536000" + }, + "json": { + "status": 1, + "group": 0, + "devices": ["test_device"], + "request": "49112cae-da34-4ac6-9d0c-977c66e41c2c" + } +} \ No newline at end of file diff --git a/tests/runtests.py b/tests/runtests.py deleted file mode 100644 index b3d1965..0000000 --- a/tests/runtests.py +++ /dev/null @@ -1,362 +0,0 @@ -import unittest -import time -import requests -import datetime -import re - -import pypushover as pypo - -try: - from tests.helpers.keys import user_key, group_key, app_key, device_id, email, pw, secret -except ImportError: # support for Travis CI - try: - import os - app_key = os.environ['app_key'] - group_key = os.environ['group_key'] - user_key = os.environ['user_key'] - device_id = os.environ['device_id'] - email = os.environ['email'] - pw = os.environ['pw'] - secret = os.environ['secret'] - except KeyError as e: - raise ImportError(e) # Environment var missing. Raise an Import Error - - -class TestMessage(unittest.TestCase): - """ - Tests message related API's. - """ - def setUp(self): - self.pm = pypo.message.MessageManager(app_key, user_key) - self.client = pypo.client.ClientManager(app_key, secret=secret, device_id=device_id) - #time.sleep(5) - #self.client.login(email, pw) - self.cleanUpClient() - # self.client.listen_async(self.client_message_receieved) - - def tearDown(self): - # self.client.stop_listening() - self.cleanUpClient() - - def cleanUpClient(self): - self.client.retrieve_message() - for msg in self.client.messages: - if msg['priority'] >= pypo.PRIORITIES.EMERGENCY and msg['acked'] != 1: - self.client.acknowledge_message(msg['receipt']) - self.client.clear_server_messages() - - self.client.retrieve_message() - self.assertEquals(len(self.client.messages), 0) - - def client_message_receieved(self, messages): - self.stored_messages = messages - - def test_val_msg(self): - - # Testing a normal push message - send_message = 'Testing normal push' - self.pm.push_message(send_message, device='test_device') - self.client.retrieve_message() - - self.assertEquals(send_message, self.client.messages[0]['message']) - - pypo.message.push_message(app_key, user_key, send_message, device='test_device') - self.client.retrieve_message() - - self.assertEquals(send_message, self.client.messages[1]['message']) - - self.client.clear_server_messages() - - # Testing normal push message with Title - - val_pm = pypo.message.MessageManager(app_key) - val_pm.push_message('Testing Title and manual user_key', title='Success!', user=user_key, device='test_device') - - - self.pm.push_message("Valid message with 'device' param", device='test_device') - - self.pm.push_message("Valid message with 'url', and 'url_title' params", - url="https://pushover.net/api#urls", - url_title="Pushover Api URLS", - device='test_device' - ) - - self.pm.push_message("Valid message with 'timestamp' param", timestamp=datetime.datetime.now(), device='test_device') - - self.pm.push_message("Valid message with 'sound' param", sound=pypo.SOUNDS.SHORT_BIKE, device='test_device') - - def test_inv_msg(self): - inv_pm = pypo.message.MessageManager(app_key) - with self.assertRaises(ValueError): - inv_pm.push_message('Missing User Key') - - def test_inv_emergency_msg(self): - with self.assertRaises(TypeError): - self.pm.push_message('Emergency: missing retry and expire', priority=pypo.PRIORITIES.EMERGENCY) - with self.assertRaises(TypeError): - pypo.message.push_message(app_key, user_key, 'Emergency: missing retry and expire', priority=pypo.PRIORITIES.EMERGENCY) - - with self.assertRaises(TypeError): - self.pm.push_message('Emergency: missing expire', priority=pypo.PRIORITIES.EMERGENCY, retry=30) - with self.assertRaises(TypeError): - pypo.message.push_message(app_key, user_key, 'Emergency: missing expire', priority=pypo.PRIORITIES.EMERGENCY, retry=30) - - with self.assertRaises(TypeError): - self.pm.push_message('Emergency: missing retry', priority=pypo.PRIORITIES.EMERGENCY, expire=3600) - with self.assertRaises(TypeError): - pypo.message.push_message(app_key, user_key, 'Emergency: missing retry', priority=pypo.PRIORITIES.EMERGENCY, expire=3600) - - with self.assertRaises(ValueError): - self.pm.push_message('Invalid expire', priority=pypo.PRIORITIES.EMERGENCY, expire=86500, retry=30) - with self.assertRaises(ValueError): - pypo.message.push_message(app_key, user_key, 'Invalid retry', priority=pypo.PRIORITIES.EMERGENCY, expire=3600, retry=20) - - def test_emergency_msg(self): - res = self.pm.push_message( - "Emergency: Valid", - priority=pypo.PRIORITIES.EMERGENCY, - retry=30, - expire=3600, - device='test_device' - ) - self.assertEqual(self.pm.check_receipt()['status'], 1) - self.assertEqual(self.pm.check_receipt(res['receipt'])['status'], 1) - self.pm.cancel_retries(res['receipt']) - - self.pm.push_message( - "Valid Emergency: Last response Cancel", - priority=pypo.PRIORITIES.EMERGENCY, - retry=30, - expire=3600, - device='test_device' - ) - self.pm.cancel_retries() - - res = pypo.message.push_message( - app_key, - user_key, - 'Emergency Valid', - priority=pypo.PRIORITIES.EMERGENCY, - retry=30, - expire=3600, - device='test_device' - ) - time.sleep(0.5) - self.assertEqual(pypo.message.check_receipt(app_key, res['receipt'])['status'], 1) - pypo.message.cancel_retries(app_key, res['receipt']) - - def test_inv_check_msg(self): - self.pm.push_message("Valid Message: no receipt", device="test_device") - with self.assertRaises(TypeError): - self.pm.check_receipt() - - def test_inv_cancel_retry(self): - self.pm.push_message("Valid Message: no reciept", device="test_device") - with self.assertRaises(TypeError): - self.pm.cancel_retries() - - -class TestGroup(unittest.TestCase): - def setUp(self): - self.valid_gm = pypo.groups.GroupManager(app_key, group_key) - - # clean up any previously failed test - if len(self.valid_gm.group.users) > 0: - self.valid_gm.remove_user(user_key) - - def test_group_info(self): - info = self.valid_gm.info() - self.assertEqual(info['name'], 'KronoTestGroup') - info = pypo.groups.info(app_key, group_key) - self.assertEqual(info['name'], 'KronoTestGroup') - - self.valid_gm.add_user(user_key, device='test_device', memo='group info test') - - self.assertEquals(self.valid_gm.group.name, 'KronoTestGroup') - self.assertEquals(self.valid_gm.group.users[0].device, 'test_device') - - self.valid_gm.remove_user(user_key) - - def test_group_add_del_user(self): - - self.valid_gm.add_user(user_key, device='test_device', memo='Added using UnitTests') - info = self.valid_gm.info() - self.assertEqual(info['users'][0]['device'], 'test_device') - self.assertEqual(info['users'][0]['memo'], 'Added using UnitTests') - - self.valid_gm.remove_user(user_key) - info = self.valid_gm.info() - self.assertEqual(len(info['users']), 0) - - def test_group_disable_enable_user(self): - self.valid_gm.add_user(user_key, device='test_device', memo='dis/ena test') - - self.valid_gm.disable_user(user_key) - info = self.valid_gm.info() - self.assertEqual(info['users'][0]['disabled'], True) - self.valid_gm.enable_user(user_key) - info = self.valid_gm.info() - self.assertEqual(info['users'][0]['disabled'], False) - - self.valid_gm.remove_user(user_key) - - def test_group_rename(self): - self.valid_gm.rename('KronoGroup') - info = self.valid_gm.info() - self.assertEqual(info['name'], 'KronoGroup') - self.valid_gm.rename('KronoTestGroup') - info = self.valid_gm.info() - self.assertEqual(info['name'], 'KronoTestGroup') - - -class TestVerifcation(unittest.TestCase): - def setUp(self): - self.valid_vm = pypo.verification.VerificationManager(app_key) - self.valid_gm = pypo.groups.GroupManager(app_key, group_key) - time.sleep(10) - - def test_val_user(self): - self.assertTrue(self.valid_vm.verify_user(user_key, device='test_device')) - time.sleep(10) - self.assertTrue(pypo.verification.verify_user(app_key, user_key, device='test_device')) - - def test_inv_user(self): - inv_user_key = "justabunchofjunk" - with self.assertRaises(pypo.PushoverError): - self.valid_vm.verify_user(inv_user_key) - - def test_val_group(self): - self.assertTrue(self.valid_vm.verify_group(group_key)) - time.sleep(10) - self.assertTrue(pypo.verification.verify_group(app_key, group_key)) - time.sleep(10) - self.assertTrue(self.valid_vm.verify_group(group_key)) - time.sleep(10) - self.assertTrue(pypo.verification.verify_group(app_key, group_key)) - - def test_inv_group(self): - inv_group_key = "justabunchofjunk" - with self.assertRaises(pypo.PushoverError): - self.valid_vm.verify_group(inv_group_key) - time.sleep(10) - with self.assertRaises(pypo.PushoverError): - pypo.verification.verify_group(app_key, inv_group_key) - time.sleep(10) - with self.assertRaises(pypo.PushoverError): - self.valid_vm.verify_user(inv_group_key) - time.sleep(10) - with self.assertRaises(pypo.PushoverError): - pypo.verification.verify_user(app_key, inv_group_key) - time.sleep(10) - with self.assertRaises(pypo.PushoverError): - self.valid_vm.verify_user(user_key, device='junk') - time.sleep(10) - with self.assertRaises(pypo.PushoverError): - pypo.verification.verify_user(app_key, user_key, device='junk') - - -class TestLicense(unittest.TestCase): - def setUp(self): - raise NotImplementedError - - -class TestClient(unittest.TestCase): - def setUp(self): - self.pm = pypo.message.MessageManager(app_key, user_key) - self.cm = pypo.client.ClientManager(app_key, device_id=device_id) - self.cm.login(email, pw) - self.cm.retrieve_message() - self.cm.clear_server_messages() - - def tearDown(self): - self.cm.stop_listening() - - def test_rec_msg(self): - msg_to_snd = 'Simple Message Sent' - self.pm.push_message(msg_to_snd, device='test_device') - self.cm.retrieve_message() - self.assertEqual(msg_to_snd, self.cm.messages[0]['message']) - - def test_ack_msg(self): - msg_to_snd = 'Emergency Message to Ack' - self.pm.push_message( - msg_to_snd, - device='test_device', - priority=pypo.PRIORITIES.EMERGENCY, - retry=30, - expire=30 - ) - self.cm.retrieve_message() - msg = self.cm.messages[0] - self.assertEqual(msg['acked'], 0) - self.cm.acknowledge_message(msg['receipt']) - self.cm.retrieve_message() - self.assertEqual(msg['id'], self.cm.messages[0]['id']) - msg = self.cm.messages[0] - self.assertEqual(msg['acked'], 1) - - @staticmethod - def callback(messages): - test_msg = "callback test message" - assert(test_msg == messages[0]['message']) - cm = pypo.client.ClientManager(app_key, secret, device_id) - cm.clear_server_messages() - - def test_listen(self): - test_msg = "test_listen message" - self.cm.listen_async(TestClient.callback) - self.pm.push_message(test_msg, device='test_device') - - -class TestBasic(unittest.TestCase): - def test_inv_app_token(self): - inv_pm = pypo.message.MessageManager(group_key, user_key) - with self.assertRaises(pypo.PushoverError): - inv_pm.push_message('This will never work') - pypo.message.push_message(group_key, app_key, 'This will never work') - - -class TestIssuesManual(unittest.TestCase): - """ - These tests require manual setup or cleanup and therefore cannot be automated using Travis CI. Run these manually - before submitting to the rel branch. - """ - def test_37_dash_support(self): - """ - WARNING! You will need to DELETE any devices created using this test manually or this test will no longer - work. - :return: - """ - cm = pypo.client.ClientManager(app_key) - vm = pypo.verification.VerificationManager(app_key) - - cm.login(email, pw) - device_id = cm.register_device('Example-2') - vm.verify_user(user_key, 'Example-2') - device_id = cm.register_device("name-with-multiple-dashes") - vm.verify_user(user_key, 'name-with-multiple-dashes') - - self._del_devices() - - def _del_devices(self): - s = requests.session() - r = s.post("https://pushover.net/login/login") - matchme = 'meta content="(.*)" name="csrf-token" /' - auth_token = re.search(matchme, str(r.text)).group(1) - payload = { - 'user[email]': email, - 'user[password]': pw, - 'authenticity_token': auth_token - } - s.post("https://pushover.net/login/login", data=payload) - payload = {'authenticity_token': auth_token} - s.post("https://pushover.net/devices/destroy/Example-2", data=payload) - s.post("https://pushover.net/devices/destroy/name-with-multiple-dashes", data=payload) - - -class TestIssues(unittest.TestCase): - pass - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_Basic.py b/tests/test_Basic.py new file mode 100644 index 0000000..0428a23 --- /dev/null +++ b/tests/test_Basic.py @@ -0,0 +1,37 @@ +import unittest + +try: + import unittest.mock as mock +except ImportError: + import mock + +import pypushover as pypo + +from tests import APP_TOKEN, USER_KEY + +class TestBasic(unittest.TestCase): + def test_inv_app_token(self): + inv_pm = pypo.message.MessageManager('invalid app token', 'invalid user key') + with self.assertRaises(pypo.PushoverError): + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + mock_response.status_code = 400 + mock_response.json.return_value = { + 'token': 'invalid', + 'errors': ['application token is invalid'], + 'status': 0, + 'request': '116f0a10-7ca7-42c2-b521-f4ba3ed6b0de' + } + inv_pm.push_message('This will never work') + + with self.assertRaises(pypo.PushoverError): + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + mock_response.status_code = 400 + mock_response.json.return_value = { + 'token': 'invalid', + 'errors': ['application token is invalid'], + 'status': 0, + 'request': '116f0a10-7ca7-42c2-b521-f4ba3ed6b0de' + } + pypo.message.push_message('invalid app token', 'invalid user key', 'This will never work') \ No newline at end of file diff --git a/tests/test_Client.py b/tests/test_Client.py new file mode 100644 index 0000000..6dfc72d --- /dev/null +++ b/tests/test_Client.py @@ -0,0 +1,50 @@ +import unittest + +import pypushover as pypo + +class TestClient(unittest.TestCase): + def setUp(self): + self.pm = pypo.message.MessageManager(app_key, user_key) + self.cm = pypo.client.ClientManager(app_key, device_id=device_id) + self.cm.login(email, pw) + self.cm.retrieve_messages() + self.cm.clear_server_messages() + + def tearDown(self): + self.cm.stop_listening() + + def test_rec_msg(self): + msg_to_snd = 'Simple Message Sent' + self.pm.push_message(msg_to_snd, device='test_device') + self.cm.retrieve_messages() + self.assertEqual(msg_to_snd, self.cm.messages[0]['message']) + + def test_ack_msg(self): + msg_to_snd = 'Emergency Message to Ack' + self.pm.push_message( + msg_to_snd, + device='test_device', + priority=pypo.PRIORITIES.EMERGENCY, + retry=30, + expire=30 + ) + self.cm.retrieve_messages() + msg = self.cm.messages[0] + self.assertEqual(msg['acked'], 0) + self.cm.acknowledge_message(msg['receipt']) + self.cm.retrieve_messages() + self.assertEqual(msg['id'], self.cm.messages[0]['id']) + msg = self.cm.messages[0] + self.assertEqual(msg['acked'], 1) + + @staticmethod + def callback(messages): + test_msg = "callback test message" + assert(test_msg == messages[0]['message']) + cm = pypo.client.ClientManager(app_key, secret, device_id) + cm.clear_server_messages() + + def test_listen(self): + test_msg = "test_listen message" + self.cm.listen_async(TestClient.callback) + self.pm.push_message(test_msg, device='test_device') \ No newline at end of file diff --git a/tests/test_Groups.py b/tests/test_Groups.py new file mode 100644 index 0000000..cf8e641 --- /dev/null +++ b/tests/test_Groups.py @@ -0,0 +1,191 @@ +import unittest +import json + +try: + import unittest.mock as mock +except ImportError as e: + import mock + +import pypushover as pypo + +from tests import APP_TOKEN, USER_KEY + +GROUP_KEY = USER_KEY + +class TestGroup(unittest.TestCase): + def setUp(self): + # Mock the response from the server + with mock.patch('pypushover._base.requests.get') as mock_get: + mock_get.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_group_info_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + self.valid_gm = pypo.groups.GroupManager(APP_TOKEN, GROUP_KEY) + + def test_group_info(self): + with mock.patch('pypushover._base.requests.get') as mock_get: + mock_get.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_group_info_response.json', 'r') as reader: + res = json.load(reader) + + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + info = self.valid_gm.info() + self.assertEqual(info['name'], 'KronoTestGroup') + + info = pypo.groups.info(APP_TOKEN, GROUP_KEY) + self.assertEqual(info['name'], 'KronoTestGroup') + + + def test_group_add_del_user(self): + with mock.patch('pypushover._base.requests.post') as mock_post, mock.patch('pypushover._base.requests.get') as mock_get: + mock_post.return_value = mock_post_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_response.json', 'r') as reader: + res = json.load(reader) + + mock_post_response.status_code = res['status_code'] + mock_post_response.json.return_value = res['json'] + mock_post_response.headers = res['headers'] + + mock_get.return_value = mock_get_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_group_info_add_user_response.json', 'r') as reader: + res = json.load(reader) + + mock_get_response.status_code = res['status_code'] + mock_get_response.json.return_value = res['json'] + mock_get_response.headers = res['headers'] + + self.valid_gm.add_user(USER_KEY, device='test_device', memo='Added using UnitTests') + + info = self.valid_gm.info() + self.assertEqual(info['users'][0]['device'], 'test_device') + self.assertEqual(info['users'][0]['memo'], 'Added using UnitTests') + + with mock.patch('pypushover._base.requests.post') as mock_post, mock.patch('pypushover._base.requests.get') as mock_get: + mock_post.return_value = mock_post_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_response.json', 'r') as reader: + res = json.load(reader) + + mock_post_response.status_code = res['status_code'] + mock_post_response.json.return_value = res['json'] + mock_post_response.headers = res['headers'] + + mock_get.return_value = mock_get_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_group_info_response.json', 'r') as reader: + res = json.load(reader) + + mock_get_response.status_code = res['status_code'] + mock_get_response.json.return_value = res['json'] + mock_get_response.headers = res['headers'] + + self.valid_gm.remove_user(USER_KEY) + + info = self.valid_gm.info() + self.assertEqual(len(info['users']), 0) + + def test_group_disable_enable_user(self): + + with mock.patch('pypushover._base.requests.post') as mock_post, mock.patch('pypushover._base.requests.get') as mock_get: + mock_post.return_value = mock_post_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_response.json', 'r') as reader: + res = json.load(reader) + + mock_post_response.status_code = res['status_code'] + mock_post_response.json.return_value = res['json'] + mock_post_response.headers = res['headers'] + + mock_get.return_value = mock_get_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_group_info_disabled_user_response.json', 'r') as reader: + res = json.load(reader) + + mock_get_response.status_code = res['status_code'] + mock_get_response.json.return_value = res['json'] + mock_get_response.headers = res['headers'] + + self.valid_gm.disable_user(USER_KEY) + info = self.valid_gm.info() + self.assertEqual(info['users'][0]['disabled'], True) + + with mock.patch('pypushover._base.requests.post') as mock_post, mock.patch('pypushover._base.requests.get') as mock_get: + mock_post.return_value = mock_post_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_response.json', 'r') as reader: + res = json.load(reader) + + mock_post_response.status_code = res['status_code'] + mock_post_response.json.return_value = res['json'] + mock_post_response.headers = res['headers'] + + mock_get.return_value = mock_get_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_group_info_enabled_user_response.json', 'r') as reader: + res = json.load(reader) + + mock_get_response.status_code = res['status_code'] + mock_get_response.json.return_value = res['json'] + mock_get_response.headers = res['headers'] + + self.valid_gm.enable_user(USER_KEY) + info = self.valid_gm.info() + self.assertEqual(info['users'][0]['disabled'], False) + + def test_group_rename(self): + with mock.patch('pypushover._base.requests.post') as mock_post, mock.patch('pypushover._base.requests.get') as mock_get: + mock_post.return_value = mock_post_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_response.json', 'r') as reader: + res = json.load(reader) + + mock_post_response.status_code = res['status_code'] + mock_post_response.json.return_value = res['json'] + mock_post_response.headers = res['headers'] + + mock_get.return_value = mock_get_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_group_info_rename_response.json', 'r') as reader: + res = json.load(reader) + + mock_get_response.status_code = res['status_code'] + mock_get_response.json.return_value = res['json'] + mock_get_response.headers = res['headers'] + + self.valid_gm.rename('KronoGroup') + info = self.valid_gm.info() + self.assertEqual(info['name'], 'KronoGroup') diff --git a/tests/test_Issues.py b/tests/test_Issues.py new file mode 100644 index 0000000..d94dc58 --- /dev/null +++ b/tests/test_Issues.py @@ -0,0 +1,54 @@ +import unittest +import requests + +import pypushover as pypo + + +class TestIssues(unittest.TestCase): + def test_37_dash_support(self): + """ + WARNING! You will need to DELETE any devices created using this test manually or this test will no longer + work. + :return: + """ + self._del_devices() + cm = pypo.client.ClientManager(app_key) + vm = pypo.verification.VerificationManager(app_key) + + cm.login(email, pw) + device_id = cm.register_device('Example-2') + vm.verify_user(user_key, 'Example-2') + device_id = cm.register_device("name-with-multiple-dashes") + vm.verify_user(user_key, 'name-with-multiple-dashes') + + self._del_devices() + + def test_39_clear_messages(self): + cm = pypo.client.ClientManager(app_key, secret=secret, device_id=device_id) + pm = pypo.message.MessageManager(app_key, user_key) + + cm.retrieve_messages() + cm.clear_server_messages() + pm.push_message('test1', device='test_device') + pm.push_message('test2', device='test_device') + + cm.retrieve_messages() + self.assertEquals(len(cm.messages), 2) + + cm.clear_server_messages() + self.assertEquals(len(cm.messages), 0) + + def _del_devices(self): + s = requests.session() + r = s.post("https://pushover.net/login/login") + matchme = 'meta content="(.*)" name="csrf-token" /' + auth_token = re.search(matchme, str(r.text)).group(1) + payload = { + 'user[email]': email, + 'user[password]': pw, + 'authenticity_token': auth_token + } + s.post("https://pushover.net/login/login", data=payload) + payload = {'authenticity_token': auth_token} + s.post("https://pushover.net/devices/destroy/Example-2", data=payload) + s.post("https://pushover.net/devices/destroy/name-with-multiple-dashes", data=payload) \ No newline at end of file diff --git a/tests/test_Messages.py b/tests/test_Messages.py new file mode 100644 index 0000000..bc222ee --- /dev/null +++ b/tests/test_Messages.py @@ -0,0 +1,308 @@ +"""Message Tests - A collection of tests to verify the functionality of the Message API +""" +import unittest +import datetime +import time +import json + +try: + import unittest.mock as mock +except ImportError as e: + import mock + +import pypushover as pypo + +from tests import APP_TOKEN, USER_KEY + + +class TestMessages(unittest.TestCase): + """ + Tests message related API's. + """ + + def setUp(self): + self.pm = pypo.message.MessageManager(APP_TOKEN, USER_KEY) + + def client_message_receieved(self, messages): + self.stored_messages = messages + + def test_val_simple_msg(self): + # Mock the response from the server + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + # Testing a normal push message + send_message = 'Testing normal push' + self.pm.push_message(send_message) + self.assertEqual(self.pm.latest_response_dict['status'], 1) + mock_post.assert_called_once_with( + pypo.message._push_url, + params={'token': APP_TOKEN, 'user': USER_KEY, + 'message': send_message} + ) + + def test_val_cmplx_msg(self): + send_message = 'Testing complex normal push' + + cur_time = datetime.datetime.now() + payload = { + 'title': 'Simple Title', + 'device': ['device-1', 'device-2'], + 'url': 'http://pushover.net/', + 'url_title': 'Pushover', + 'timestamp': cur_time, + 'sound': pypo.SOUNDS.SHORT_CLASSICAL, + 'html': True + } + + expected_payload = payload.copy() + expected_payload.update({ + 'token': APP_TOKEN, + 'user': USER_KEY, + 'message': send_message, + 'timestamp': int(time.mktime(cur_time.timetuple())) + }) + + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + self.pm.push_message( + send_message, + **payload + ) + + self.assertEqual(mock_response.json(), self.pm.latest_response_dict) + mock_post.assert_called_once_with( + pypo.message._push_url, + params=expected_payload + ) + + def test_inv_emergency_msg(self): + with self.assertRaises(TypeError): + self.pm.push_message( + 'Emergency: missing retry and expire', priority=pypo.PRIORITIES.EMERGENCY) + with self.assertRaises(TypeError): + pypo.message.push_message( + APP_TOKEN, USER_KEY, 'Emergency: missing retry and expire', priority=pypo.PRIORITIES.EMERGENCY) + + with self.assertRaises(TypeError): + self.pm.push_message('Emergency: missing expire', + priority=pypo.PRIORITIES.EMERGENCY, retry=30) + with self.assertRaises(TypeError): + pypo.message.push_message( + APP_TOKEN, USER_KEY, 'Emergency: missing expire', priority=pypo.PRIORITIES.EMERGENCY, retry=30) + + with self.assertRaises(TypeError): + self.pm.push_message('Emergency: missing retry', + priority=pypo.PRIORITIES.EMERGENCY, expire=3600) + with self.assertRaises(TypeError): + pypo.message.push_message( + APP_TOKEN, USER_KEY, 'Emergency: missing retry', priority=pypo.PRIORITIES.EMERGENCY, expire=3600) + + with self.assertRaises(ValueError): + self.pm.push_message( + 'Invalid expire', priority=pypo.PRIORITIES.EMERGENCY, expire=86500, retry=30) + with self.assertRaises(ValueError): + pypo.message.push_message(APP_TOKEN, USER_KEY, 'Invalid retry', + priority=pypo.PRIORITIES.EMERGENCY, expire=3600, retry=20) + + def test_emergency_msg(self): + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_emergency_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + send_message = "Emergency: Valid" + payload = { + 'priority': pypo.PRIORITIES.EMERGENCY, + 'retry': 30, + 'expire': 3600, + 'device': 'test_device' + } + + expected_payload = payload.copy() + expected_payload.update( + {'token': APP_TOKEN, 'user': USER_KEY, 'message': send_message}) + + self.pm.push_message( + send_message, + **payload + ) + + self.assertEqual(mock_response.json(), self.pm.latest_response_dict) + mock_post.assert_called_with( + pypo.message._push_url, + params=expected_payload + ) + + def test_check_receipt(self): + # Setup the initial 'message' sent + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the json and use it for the + # mocked response + with open('tests/responses/test_good_emergency_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + receipt = res['json']['receipt'] + + send_message = "Emergency: Valid" + payload = { + 'priority': pypo.PRIORITIES.EMERGENCY, + 'retry': 30, + 'expire': 3600, + 'device': 'test_device' + } + + self.pm.push_message( + send_message, + **payload + ) + + with mock.patch('pypushover._base.requests.get') as mock_get: + mock_get.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_check_receipt_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + self.pm.check_receipt(receipt=receipt) + + self.assertEqual(mock_response.json(), self.pm.latest_response_dict) + mock_get.assert_called_once_with( + pypo.message._receipt_url.format(receipt=receipt), + params={'token': APP_TOKEN} + ) + + def test_cancel_retry(self): + # Setup the initial 'message' sent + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the json and use it for the + # mocked response + with open('tests/responses/test_good_emergency_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + receipt = res['json']['receipt'] + + send_message = "Emergency: Valid" + payload = { + 'priority': pypo.PRIORITIES.EMERGENCY, + 'retry': 30, + 'expire': 3600, + 'device': 'test_device' + } + + self.pm.push_message( + send_message, + **payload + ) + + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the json and use it for the + # mocked response + with open('tests/responses/test_good_check_receipt_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + self.pm.cancel_retries(receipt=receipt) + + self.assertEqual(mock_response.json(), self.pm.latest_response_dict) + mock_post.assert_called_once_with( + pypo.message._cancel_receipt_url.format(receipt=receipt), + params={'token': APP_TOKEN} + ) + + def test_inv_check_msg(self): + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + 'status': 1, + 'request': 'cbc6c03f-0667-43dc-b151-96971a5c4605' + } + + self.pm.push_message( + "Valid Message: no receipt", device="test_device") + with self.assertRaises(TypeError): + self.pm.check_receipt() + + def test_inv_cancel_retry(self): + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + 'request': '6af7c78e85c1edcb2f1426214eb6ec48', 'status': 1 + } + + # 'fake' latest response + self.pm.latest_response_dict = { + 'app_reset': '1467349200', + 'app_remaining': '7172', + 'receipt': 'ryd8w31fydoo19ipjysa4cwc9eh87h', + 'app_limit': '7500', + 'request': '0219487621ddc7718a0eb71961d9c6f9', + 'status': 1 + } + + receipt = 'ryd8w31fydoo19ipjysa4cwc9eh87h' + + self.pm.cancel_retries(receipt=receipt) + + self.assertEqual(mock_response.json(), + self.pm.latest_response_dict) + mock_post.assert_called_once_with( + pypo.message._cancel_receipt_url.format(receipt=receipt), + params={'token': APP_TOKEN} + ) + + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + 'status': 1, + 'request': 'cbc6c03f-0667-43dc-b151-96971a5c4605' + } + + self.pm.push_message( + "Valid Message: no reciept", device="test_device") + with self.assertRaises(TypeError): + self.pm.cancel_retries() diff --git a/tests/test_Verification.py b/tests/test_Verification.py new file mode 100644 index 0000000..1fe8a42 --- /dev/null +++ b/tests/test_Verification.py @@ -0,0 +1,108 @@ +import unittest +import time +import json + +try: + import unittest.mock as mock +except ImportError as e: + import mock + +import pypushover as pypo + +from tests import APP_TOKEN, USER_KEY + +GROUP_KEY = USER_KEY + +class TestVerification(unittest.TestCase): + def setUp(self): + self.valid_vm = pypo.verification.VerificationManager(APP_TOKEN) + + with mock.patch('pypushover._base.requests.get') as mock_get: + mock_get.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_group_info_enabled_user_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + self.valid_gm = pypo.groups.GroupManager(APP_TOKEN, GROUP_KEY) + + def test_val_user(self): + # Mock the response from the server + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_verify_user_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + self.assertTrue(self.valid_vm.verify_user(USER_KEY, device='test_device')) + + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_verify_user_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + self.assertTrue(pypo.verification.verify_user(APP_TOKEN, USER_KEY, device='test_device')) + + def test_inv_user(self): + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_bad_verify_user_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + inv_USER_KEY = "justabunchofjunk" + with self.assertRaises(pypo.PushoverError): + self.valid_vm.verify_user(inv_USER_KEY) + + def test_val_group(self): + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_good_verify_group_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + self.assertTrue(self.valid_vm.verify_group(GROUP_KEY)) + self.assertTrue(pypo.verification.verify_group(APP_TOKEN, GROUP_KEY)) + + def test_inv_group(self): + with mock.patch('pypushover._base.requests.post') as mock_post: + mock_post.return_value = mock_response = mock.Mock() + + # load a real response from the pickle_jar and use it for the + # mocked response + with open('tests/responses/test_bad_verify_user_response.json', 'r') as reader: + res = json.load(reader) + mock_response.status_code = res['status_code'] + mock_response.json.return_value = res['json'] + mock_response.headers = res['headers'] + + inv_GROUP_KEY = "justabunchofjunk" + with self.assertRaises(pypo.PushoverError): + pypo.verification.verify_group(APP_TOKEN, inv_GROUP_KEY) + with self.assertRaises(pypo.PushoverError): + pypo.verification.verify_user(APP_TOKEN, inv_GROUP_KEY)