Using pytest from now on

.drone.jsonnet

@@ -6,9 +6,8 @@ local Pipeline(py_version) = {
       name: "test",
       image: "python:" + py_version,
       commands: [
-        "pip install .",
+        "pip install .[tests]",
-        "pip install -r requirements-dev.txt",
+        "pytest"
-        "nosetests"
       ]
     }
   ]

.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/

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
+pytest
-python -m unittest discover test
 ```
 
 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 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 local 3.7.1 3.6.7 3.5.6
 pip install tox
 tox
 ```

requirements-dev.txt

@@ -1,3 +0,0 @@
-freezegun
-nose
-mock; python_version < '3.3'

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',

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:
+import pytest
-    # Python 3
-    from unittest.mock import MagicMock, Mock, patch
-except ImportError:
-    # Python 2
-    from mock import MagicMock, Mock, patch
 
 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):
+@pytest.fixture(scope='session')
-        self.open_streams = 0
+def tokens():
-        self.max_open_streams = 0
+    return ['%064x' % i for i in range(1000)]
-        self.mock_results = None
-        self.next_stream_id = 0
 
-        with patch('apns2.credentials.HTTP20Connection') as mock_connection_constructor, patch('apns2.credentials.init_context'):
+
-            self.mock_connection = MagicMock()
+@pytest.fixture(scope='session')
-            self.mock_connection.get_response.side_effect = self.mock_get_response
+def notifications(tokens):
-            self.mock_connection.request.side_effect = self.mock_request
+    payload = Payload(alert='Test alert')
-            self.mock_connection._conn.__enter__.return_value = self.mock_connection._conn
+    return [Notification(token=token, payload=payload) for token in tokens]
-            self.mock_connection._conn.remote_settings.max_concurrent_streams = 500
+
-            mock_connection_constructor.return_value = self.mock_connection
+
-            self.client = APNsClient(credentials=None)
+@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):
+    def mock_get_response(stream_id):
-        self.open_streams -= 1
+        mock_connection.__open_streams -= 1
-        if self.mock_results:
+        if mock_connection.__mock_results:
-            reason = self.mock_results[stream_id]
+            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):
+    def mock_request(*_args):
-        self.open_streams += 1
+        mock_connection.__open_streams += 1
-        if self.open_streams > self.max_open_streams:
+        mock_connection.__max_open_streams = max(mock_connection.__open_streams, mock_connection.__max_open_streams)
-            self.max_open_streams = self.open_streams
 
-        stream_id = self.next_stream_id
+        stream_id = mock_connection.__next_stream_id
-        self.next_stream_id += 1
+        mock_connection.__next_stream_id += 1
         return stream_id
 
-    def test_send_notification_batch_returns_results_in_order(self):
+    mock_connection.get_response.side_effect = mock_get_response
-        results = self.client.send_notification_batch(self.notifications, self.topic)
+    mock_connection.request.side_effect = mock_request
-        expected_results = {token: 'Success' for token in self.tokens}
+    mock_connection._conn.__enter__.return_value = mock_connection._conn
-        self.assertEqual(results, expected_results)
+    mock_connection._conn.remote_settings.max_concurrent_streams = 500
 
-    def test_send_notification_batch_respects_max_concurrent_streams_from_server(self):
+    return mock_connection
-        self.client.send_notification_batch(self.notifications, self.topic)
-        self.assertEqual(self.max_open_streams, 500)
 
-    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):
+def test_connect_establishes_connection(client, mock_connection):
-        self.mock_connection._conn.remote_settings.max_concurrent_streams = 0
+    client.connect()
-        self.client.send_notification_batch(self.notifications, self.topic)
+    mock_connection.connect.assert_called_once_with()
-        self.assertEqual(self.max_open_streams, 1)
 
-    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)
+    results = client.send_notification_batch(notifications, TOPIC)
-        expected_results = dict(zip(self.tokens, self.mock_results))
+    expected_results = dict(zip(tokens, mock_connection.__mock_results))
-        self.assertEqual(results, expected_results)
+    assert 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)

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
 
-
+TOPIC = 'com.example.first_app'
-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)
 
 
-if __name__ == '__main__':
+@pytest.fixture
-    main()
+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

test/test_payload.py

@@ -1,76 +1,82 @@
-# pylint: disable=protected-access
+import pytest
-
-from unittest import TestCase
 
 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):
+def test_payload_alert(payload_alert):
-        payload = Payload(
+    assert payload_alert.dict() == {
-            alert='my_alert', badge=2, sound='chime',
+        'title': 'title',
-            content_available=1, mutable_content=3,
+        'title-loc-key': 'loc_k',
-            category='my_category', url_args='args', custom={'extra': 'something'}, thread_id=42)
+        'title-loc-args': 'loc_a',
-        self.assertEqual(payload.dict(), {
+        'body': 'body',
-            'aps': {
+        'loc-key': 'body_loc_k',
-                'alert': 'my_alert',
+        'loc-args': 'body_loc_a',
-                'badge': 2,
+        'action-loc-key': 'ac_loc_k',
-                'sound': 'chime',
+        'action': 'send',
-                'content-available': 1,
+        'launch-image': 'img'
-                'mutable-content': 1,
+    }
-                'thread-id': 42,
+
-                'category': 'my_category',
+
-                'url-args': 'args'
+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'
+            'badge': 2,
-        })
+            'sound': 'chime',
-
+            'content-available': 1,
-    def test_payload_with_payload_alert(self):
+            'mutable-content': 1,
-        payload = Payload(
+            'thread-id': 42,
-            alert=self.payload_alert, badge=2, sound='chime',
+            'category': 'my_category',
-            content_available=1, mutable_content=1,
+            'url-args': 'args',
-            category='my_category', url_args='args', custom={'extra': 'something'}, thread_id=42)
+        },
-        self.assertEqual(payload.dict(), {
+        'extra': 'something'
-            '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'
-        })

tox.ini

@@ -1,11 +1,9 @@
 [tox]
-envlist = py27, py35, py36, py37
+envlist = py35, py36, py37
 
 [testenv]
-commands = nosetests []
+commands = pytest {posargs}
 
-deps =
+extras = tests
-    nose
-    -rrequirements-dev.txt
 
 usedevelop=True