Enable JWT authentication in addition to old certificate authentication (#40)

test/eckey.pem

@@ -0,0 +1,8 @@
+-----BEGIN EC PARAMETERS-----
+BggqhkjOPQMBBw==
+-----END EC PARAMETERS-----
+-----BEGIN EC PRIVATE KEY-----
+MHcCAQEEIM0scFXkVBBc7d8DSL9rfFB1PvET/dWQa9eWfxpgaqaBoAoGCCqGSM49
+AwEHoUQDQgAEnVxQ41VKMN6uSsSYCCdOUhCms+HT2VpUhFGML5SzYGoodKtRD/6J
+YI9Rxq1lPGHMwECSaPtPf9kVDCUM6UHvhA==
+-----END EC PRIVATE KEY-----

test/test_client.py

@@ -32,14 +32,14 @@ class ClientTestCase(TestCase):
         self.mock_results = None
         self.next_stream_id = 0
 
-        with patch('apns2.client.HTTP20Connection') as mock_connection_constructor, patch('apns2.client.init_context'):
+        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(cert_file=None)
+            self.client = APNsClient(credentials=None)
 
     @contextlib.contextmanager
     def mock_get_response(self, stream_id):

test/test_credentials.py

@@ -0,0 +1,56 @@
+# This only tests the TokenCredentials test case, since the
+# CertificateCredentials would be mocked out anyway.
+# Namely:
+# - timing out of the token
+# - creating multiple tokens for different topics
+
+from unittest import TestCase, main
+
+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])
+        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_create_multiple_topics(self):
+        h1 = self.normal_creds.get_authorization_header(self.topics[0])
+        self.assertEqual(len(self.normal_creds.get_tokens()), 1)
+        h2 = self.normal_creds.get_authorization_header(self.topics[1])
+        self.assertNotEqual(h1, h2)
+        self.assertEqual(len(self.normal_creds.get_tokens()), 2)
+
+    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__':
+    main()