From be7438a653f743f168c5bda85021c212992f9bc6 Mon Sep 17 00:00:00 2001 From: Sergey Petrov Date: Sun, 2 Jun 2019 19:39:40 +0300 Subject: [PATCH] Using pytest from now on --- .drone.jsonnet | 5 +- .gitignore | 22 +---- README.md | 9 +- requirements-dev.txt | 3 - setup.py | 6 ++ test/test_client.py | 181 +++++++++++++++++++++------------------ test/test_credentials.py | 59 ++++--------- test/test_payload.py | 144 ++++++++++++++++--------------- tox.ini | 8 +- 9 files changed, 207 insertions(+), 230 deletions(-) delete mode 100644 requirements-dev.txt diff --git a/.drone.jsonnet b/.drone.jsonnet index 7c62ebb..00e17af 100644 --- a/.drone.jsonnet +++ b/.drone.jsonnet @@ -6,9 +6,8 @@ local Pipeline(py_version) = { name: "test", image: "python:" + py_version, commands: [ - "pip install .", - "pip install -r requirements-dev.txt", - "nosetests" + "pip install .[tests]", + "pytest" ] } ] diff --git a/.gitignore b/.gitignore index fd9f31c..bb4279c 100644 --- a/.gitignore +++ b/.gitignore @@ -25,12 +25,6 @@ MANIFEST .installed.cfg *.egg -# PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. -*.manifest -*.spec - # Installer logs pip-log.txt pip-delete-this-directory.txt @@ -45,28 +39,14 @@ nosetests.xml coverage.xml *,cover .hypothesis/ +.pytest_cache/ # Translations *.mo *.pot -# Django stuff: -*.log - # Sphinx documentation docs/_build/ -# PyBuilder -target/ - -#Ipython Notebook -.ipynb_checkpoints - -# IntelliJ IDEA -.idea/ - # Virtualenv venv/ - -# Tox -.tox/ diff --git a/README.md b/README.md index 7a59f11..6d5d517 100644 --- a/README.md +++ b/README.md @@ -54,19 +54,18 @@ cd PyAPNs2 # Create a virtualenv and install dependencies. virtualenv venv . venv/bin/activate -pip install -e . +pip install -e .[tests] ``` To run the tests: ```shell -pip install -r requirements-dev.txt -python -m unittest discover test +pytest ``` You can use `tox` for running tests with all supported Python versions: ```shell -pyenv install 2.7.15; pyenv install 3.4.9; pyenv install 3.5.6; pyenv install 3.6.7; pyenv install 3.7.1 -pyenv local 3.7.1 3.6.7 3.5.6 3.4.9 2.7.15 +pyenv install 3.5.6; pyenv install 3.6.7; pyenv install 3.7.1 +pyenv local 3.7.1 3.6.7 3.5.6 pip install tox tox ``` diff --git a/requirements-dev.txt b/requirements-dev.txt deleted file mode 100644 index cca3309..0000000 --- a/requirements-dev.txt +++ /dev/null @@ -1,3 +0,0 @@ -freezegun -nose -mock; python_version < '3.3' diff --git a/setup.py b/setup.py index 774aea4..b3cb0ee 100755 --- a/setup.py +++ b/setup.py @@ -11,6 +11,12 @@ setup( 'PyJWT>=1.4.0', 'cryptography>=1.7.2', ], + extras_require={ + "tests": [ + 'freezegun', + 'pytest', + ], + }, url='https://github.com/Pr0Ger/PyAPNs2', license='MIT', author='Sergey Petrov', diff --git a/test/test_client.py b/test/test_client.py index 7681d5a..ef8c017 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -1,110 +1,125 @@ -# pylint: disable=protected-access - -from unittest import TestCase import contextlib -import logging +from unittest.mock import MagicMock, Mock, patch -try: - # Python 3 - from unittest.mock import MagicMock, Mock, patch -except ImportError: - # Python 2 - from mock import MagicMock, Mock, patch +import pytest from apns2.client import APNsClient, CONCURRENT_STREAMS_SAFETY_MAXIMUM, Notification from apns2.errors import ConnectionFailed from apns2.payload import Payload +TOPIC = 'com.example.App' -class ClientTestCase(TestCase): - @classmethod - def setUpClass(cls): - # Ignore all log messages so that test output is not cluttered. - logging.basicConfig(level=logging.CRITICAL) - cls.tokens = ['%064x' % i for i in range(10000)] - cls.payload = Payload(alert='Test alert') - cls.notifications = [Notification(token=token, payload=cls.payload) for token in cls.tokens] - cls.topic = 'com.example.App' - def setUp(self): - self.open_streams = 0 - self.max_open_streams = 0 - self.mock_results = None - self.next_stream_id = 0 +@pytest.fixture(scope='session') +def tokens(): + return ['%064x' % i for i in range(1000)] - with patch('apns2.credentials.HTTP20Connection') as mock_connection_constructor, patch('apns2.credentials.init_context'): - self.mock_connection = MagicMock() - self.mock_connection.get_response.side_effect = self.mock_get_response - self.mock_connection.request.side_effect = self.mock_request - self.mock_connection._conn.__enter__.return_value = self.mock_connection._conn - self.mock_connection._conn.remote_settings.max_concurrent_streams = 500 - mock_connection_constructor.return_value = self.mock_connection - self.client = APNsClient(credentials=None) + +@pytest.fixture(scope='session') +def notifications(tokens): + payload = Payload(alert='Test alert') + return [Notification(token=token, payload=payload) for token in tokens] + + +@patch('apns2.credentials.init_context') +@pytest.fixture +def client(mock_connection): + with patch('apns2.credentials.HTTP20Connection') as mock_connection_constructor: + mock_connection_constructor.return_value = mock_connection + return APNsClient(credentials=None) + + +@pytest.fixture +def mock_connection(): + mock_connection = MagicMock() + mock_connection.__max_open_streams = 0 + mock_connection.__open_streams = 0 + mock_connection.__mock_results = None + mock_connection.__next_stream_id = 0 @contextlib.contextmanager - def mock_get_response(self, stream_id): - self.open_streams -= 1 - if self.mock_results: - reason = self.mock_results[stream_id] + def mock_get_response(stream_id): + mock_connection.__open_streams -= 1 + if mock_connection.__mock_results: + reason = mock_connection.__mock_results[stream_id] response = Mock(status=200 if reason == 'Success' else 400) response.read.return_value = ('{"reason": "%s"}' % reason).encode('utf-8') yield response else: yield Mock(status=200) - def mock_request(self, *dummy_args): - self.open_streams += 1 - if self.open_streams > self.max_open_streams: - self.max_open_streams = self.open_streams + def mock_request(*_args): + mock_connection.__open_streams += 1 + mock_connection.__max_open_streams = max(mock_connection.__open_streams, mock_connection.__max_open_streams) - stream_id = self.next_stream_id - self.next_stream_id += 1 + stream_id = mock_connection.__next_stream_id + mock_connection.__next_stream_id += 1 return stream_id - def test_send_notification_batch_returns_results_in_order(self): - results = self.client.send_notification_batch(self.notifications, self.topic) - expected_results = {token: 'Success' for token in self.tokens} - self.assertEqual(results, expected_results) + mock_connection.get_response.side_effect = mock_get_response + mock_connection.request.side_effect = mock_request + mock_connection._conn.__enter__.return_value = mock_connection._conn + mock_connection._conn.remote_settings.max_concurrent_streams = 500 - def test_send_notification_batch_respects_max_concurrent_streams_from_server(self): - self.client.send_notification_batch(self.notifications, self.topic) - self.assertEqual(self.max_open_streams, 500) + return mock_connection - def test_send_notification_batch_overrides_server_max_concurrent_streams_if_too_large(self): - self.mock_connection._conn.remote_settings.max_concurrent_streams = 5000 - self.client.send_notification_batch(self.notifications, self.topic) - self.assertEqual(self.max_open_streams, CONCURRENT_STREAMS_SAFETY_MAXIMUM) - def test_send_notification_batch_overrides_server_max_concurrent_streams_if_too_small(self): - self.mock_connection._conn.remote_settings.max_concurrent_streams = 0 - self.client.send_notification_batch(self.notifications, self.topic) - self.assertEqual(self.max_open_streams, 1) +def test_connect_establishes_connection(client, mock_connection): + client.connect() + mock_connection.connect.assert_called_once_with() - def test_send_notification_batch_reports_different_results(self): - self.mock_results = ( + +def test_connect_retries_failed_connection(client, mock_connection): + mock_connection.connect.side_effect = [RuntimeError, RuntimeError, None] + client.connect() + assert mock_connection.connect.call_count == 3 + + +def test_connect_stops_on_reaching_max_retries(client, mock_connection): + mock_connection.connect.side_effect = [RuntimeError] * 4 + with pytest.raises(ConnectionFailed): + client.connect() + + assert mock_connection.connect.call_count == 3 + + +def test_send_empty_batch_does_nothing(client, mock_connection): + client.send_notification_batch([], TOPIC) + assert mock_connection.request.call_count == 0 + + +def test_send_notification_batch_returns_results_in_order(client, mock_connection, tokens, notifications): + results = client.send_notification_batch(notifications, TOPIC) + expected_results = {token: 'Success' for token in tokens} + assert results == expected_results + + +def test_send_notification_batch_respects_max_concurrent_streams_from_server(client, mock_connection, tokens, + notifications): + client.send_notification_batch(notifications, TOPIC) + assert mock_connection.__max_open_streams == 500 + + +def test_send_notification_batch_overrides_server_max_concurrent_streams_if_too_large(client, mock_connection, tokens, + notifications): + mock_connection._conn.remote_settings.max_concurrent_streams = 5000 + client.send_notification_batch(notifications, TOPIC) + assert mock_connection.__max_open_streams == CONCURRENT_STREAMS_SAFETY_MAXIMUM + + +def test_send_notification_batch_overrides_server_max_concurrent_streams_if_too_small(client, mock_connection, tokens, + notifications): + mock_connection._conn.remote_settings.max_concurrent_streams = 0 + client.send_notification_batch(notifications, TOPIC) + assert mock_connection.__max_open_streams == 1 + + +def test_send_notification_batch_reports_different_results(client, mock_connection, tokens, + notifications): + mock_connection.__mock_results = ( ['BadDeviceToken'] * 1000 + ['Success'] * 1000 + ['DeviceTokenNotForTopic'] * 2000 + ['Success'] * 1000 + ['BadDeviceToken'] * 500 + ['PayloadTooLarge'] * 4500 - ) - results = self.client.send_notification_batch(self.notifications, self.topic) - expected_results = dict(zip(self.tokens, self.mock_results)) - self.assertEqual(results, expected_results) - - def test_send_empty_batch_does_nothing(self): - self.client.send_notification_batch([], self.topic) - self.assertEqual(self.mock_connection.request.call_count, 0) - - def test_connect_establishes_connection(self): - self.client.connect() - self.mock_connection.connect.assert_called_once_with() - - def test_connect_retries_failed_connection(self): - self.mock_connection.connect.side_effect = [RuntimeError, RuntimeError, None] - self.client.connect() - self.assertEqual(self.mock_connection.connect.call_count, 3) - - def test_connect_stops_on_reaching_max_retries(self): - self.mock_connection.connect.side_effect = [RuntimeError] * 4 - with self.assertRaises(ConnectionFailed): - self.client.connect() - - self.assertEqual(self.mock_connection.connect.call_count, 3) + ) + results = client.send_notification_batch(notifications, TOPIC) + expected_results = dict(zip(tokens, mock_connection.__mock_results)) + assert results == expected_results diff --git a/test/test_credentials.py b/test/test_credentials.py index 6091859..9c4525b 100644 --- a/test/test_credentials.py +++ b/test/test_credentials.py @@ -4,50 +4,27 @@ # - timing out of the token # - creating multiple tokens for different topics -from unittest import TestCase, main - +import pytest from freezegun import freeze_time -import time - from apns2.credentials import TokenCredentials - -class TokenCredentialsTestCase(TestCase): - @classmethod - def setUpClass(cls): - cls.key_path = 'test/eckey.pem' - cls.team_id = '3Z24IP123A' - cls.key_id = '1QBCDJ9RST' - cls.topics = ('com.example.first_app', 'com.example.second_app',) - cls.token_lifetime = 0.5 - - def setUp(self): - # Create an 'ephemeral' token so we can test token timeouts. We - # want a timeout long enough to last the test, but we don't want to - # slow down the tests too much either. - self.normal_creds = TokenCredentials(self.key_path, self.key_id, - self.team_id) - self.lasting_header = self.normal_creds.get_authorization_header( - self.topics[0]) - - with freeze_time('2012-01-14'): - self.expiring_creds = \ - TokenCredentials(self.key_path, self.key_id, - self.team_id, - token_lifetime=self.token_lifetime) - self.expiring_header = self.expiring_creds.get_authorization_header( - self.topics[0]) - - def test_token_expiration(self): - # As long as the token lifetime hasn't elapsed, this should work. To - # be really careful, we should check how much time has elapsed to - # know if it fail. But, either way, we'd have to come up with a good - # lifetime for future tests... - time.sleep(self.token_lifetime) - h3 = self.expiring_creds.get_authorization_header(self.topics[0]) - self.assertNotEqual(self.expiring_header, h3) +TOPIC = 'com.example.first_app' -if __name__ == '__main__': - main() +@pytest.fixture +def token_credentials(): + return TokenCredentials('test/eckey.pem', '1QBCDJ9RST', '3Z24IP123A') + + +def test_token_expiration(token_credentials): + # As long as the token lifetime hasn't elapsed, this should work. To + # be really careful, we should check how much time has elapsed to + # know if it fail. But, either way, we'd have to come up with a good + # lifetime for future tests... + + with freeze_time('2012-01-14'): + expiring_header = token_credentials.get_authorization_header(TOPIC) + + new_header = token_credentials.get_authorization_header(TOPIC) + assert expiring_header != new_header diff --git a/test/test_payload.py b/test/test_payload.py index 006a888..3da555f 100644 --- a/test/test_payload.py +++ b/test/test_payload.py @@ -1,76 +1,82 @@ -# pylint: disable=protected-access - -from unittest import TestCase +import pytest from apns2.payload import Payload, PayloadAlert -class PayloadTestCase(TestCase): +@pytest.fixture +def payload_alert(): + return PayloadAlert( + title='title', + title_localized_key='loc_k', + title_localized_args='loc_a', + body='body', + body_localized_key='body_loc_k', + body_localized_args='body_loc_a', + action_localized_key='ac_loc_k', + action='send', + launch_image='img' + ) - def setUp(self): - self.payload_alert = payload_alert = PayloadAlert( - title='title', title_localized_key='loc_k', title_localized_args='loc_a', - body='body', body_localized_key='body_loc_k', body_localized_args='body_loc_a', - action_localized_key='ac_loc_k', action='send', - launch_image='img') - def test_payload(self): - payload = Payload( - alert='my_alert', badge=2, sound='chime', - content_available=1, mutable_content=3, - category='my_category', url_args='args', custom={'extra': 'something'}, thread_id=42) - self.assertEqual(payload.dict(), { - 'aps': { - 'alert': 'my_alert', - 'badge': 2, - 'sound': 'chime', - 'content-available': 1, - 'mutable-content': 1, - 'thread-id': 42, - 'category': 'my_category', - 'url-args': 'args' +def test_payload_alert(payload_alert): + assert payload_alert.dict() == { + 'title': 'title', + 'title-loc-key': 'loc_k', + 'title-loc-args': 'loc_a', + 'body': 'body', + 'loc-key': 'body_loc_k', + 'loc-args': 'body_loc_a', + 'action-loc-key': 'ac_loc_k', + 'action': 'send', + 'launch-image': 'img' + } + + +def test_payload(): + payload = Payload( + alert='my_alert', badge=2, sound='chime', + content_available=1, mutable_content=3, + category='my_category', url_args='args', custom={'extra': 'something'}, thread_id=42) + assert payload.dict() == { + 'aps': { + 'alert': 'my_alert', + 'badge': 2, + 'sound': 'chime', + 'content-available': 1, + 'mutable-content': 1, + 'thread-id': 42, + 'category': 'my_category', + 'url-args': 'args' + }, + 'extra': 'something' + } + + +def test_payload_with_payload_alert(payload_alert): + payload = Payload( + alert=payload_alert, badge=2, sound='chime', + content_available=1, mutable_content=1, + category='my_category', url_args='args', custom={'extra': 'something'}, thread_id=42) + assert payload.dict() == { + 'aps': { + 'alert': { + 'title': 'title', + 'title-loc-key': 'loc_k', + 'title-loc-args': 'loc_a', + 'body': 'body', + 'loc-key': 'body_loc_k', + 'loc-args': 'body_loc_a', + 'action-loc-key': 'ac_loc_k', + 'action': 'send', + 'launch-image': 'img' }, - 'extra': 'something' - }) - - def test_payload_with_payload_alert(self): - payload = Payload( - alert=self.payload_alert, badge=2, sound='chime', - content_available=1, mutable_content=1, - category='my_category', url_args='args', custom={'extra': 'something'}, thread_id=42) - self.assertEqual(payload.dict(), { - 'aps': { - 'alert': { - 'title': 'title', - 'title-loc-key': 'loc_k', - 'title-loc-args': 'loc_a', - 'body': 'body', - 'loc-key': 'body_loc_k', - 'loc-args': 'body_loc_a', - 'action-loc-key': 'ac_loc_k', - 'action': 'send', - 'launch-image': 'img' - }, - 'badge': 2, - 'sound': 'chime', - 'content-available': 1, - 'mutable-content': 1, - 'thread-id': 42, - 'category': 'my_category', - 'url-args': 'args', - }, - 'extra': 'something' - }) - - def test_payload_alert(self): - self.assertEqual(self.payload_alert.dict(), { - 'title': 'title', - 'title-loc-key': 'loc_k', - 'title-loc-args': 'loc_a', - 'body': 'body', - 'loc-key': 'body_loc_k', - 'loc-args': 'body_loc_a', - 'action-loc-key': 'ac_loc_k', - 'action': 'send', - 'launch-image': 'img' - }) + 'badge': 2, + 'sound': 'chime', + 'content-available': 1, + 'mutable-content': 1, + 'thread-id': 42, + 'category': 'my_category', + 'url-args': 'args', + }, + 'extra': 'something' + } diff --git a/tox.ini b/tox.ini index 0ccb67c..f2b7bfc 100644 --- a/tox.ini +++ b/tox.ini @@ -1,11 +1,9 @@ [tox] -envlist = py27, py35, py36, py37 +envlist = py35, py36, py37 [testenv] -commands = nosetests [] +commands = pytest {posargs} -deps = - nose - -rrequirements-dev.txt +extras = tests usedevelop=True