Merge pull request #15 from patroqueeet/feature/replace-pyapns2

migrate to aioapns #10

.gitignore

@@ -1,7 +1,9 @@
 *.py[cod]
+.DS_Store
 
 # C extensions
 *.so
+*.swp
 
 # Packages
 *.egg

CHANGELOG.md

@@ -1,8 +1,16 @@
 # Changelog
 
-## [Unreleased]
+## [3.0]
+
+## Breaking Changes
+
+- deprecate cert based authentication for APNS
+
+## Minor Changes
+
+- replace `pyAPNS2` with `aioapns` to enable Django>=4.0 compatibility and resolve unmaintained `hyper` dependency
+- increase minimum Python version to 3.6
 
-- TODO
 
 ## [2.0] - 2023-01-09
 

django_walletpass/models.py

@@ -228,10 +228,13 @@ class Pass(models.Model):
     )
     updated_at = models.DateTimeField(auto_now=True)
 
+    def get_registrations(self):
+         return self.registrations.all()
+
     def push_notification(self):
         klass = import_string(WALLETPASS_CONF['WALLETPASS_PUSH_CLASS'])
         push_module = klass()
-        for registration in self.registrations.all():
+        for registration in self.get_registrations():
             push_module.push_notification_from_instance(registration)
 
     def new_pass_builder(self, directory=None):

django_walletpass/services.py

@@ -1,70 +1,46 @@
+import asyncio
 import logging
 from ssl import SSLError
-from hyper.tls import init_context
-from hyper.http20.exceptions import StreamResetError
-from apns2.client import APNsClient
-from apns2.credentials import Credentials
-from apns2.credentials import TokenCredentials
-from apns2.payload import Payload
+
+from aioapns import APNs, NotificationRequest
+from aioapns.exceptions import ConnectionClosed, ConnectionError
 from django_walletpass.models import Registration
 from django_walletpass.settings import dwpconfig as WALLETPASS_CONF
 
 logger = logging.getLogger('walletpass.services')
 
 
 class PushBackend:
-    def push_notification(self, credentials, push_sandbox, pass_type_id, push_token):
-        payload = Payload()
+    def __init__(self):
+        self.loop = asyncio.get_event_loop()
+
+    async def push_notification(self, client, token):
+
         try:
-            client = APNsClient(
-                credentials,
-                use_sandbox=push_sandbox,
-                use_alternative_port=False,
-            )
-            client.send_notification(
-                push_token,
-                payload,
-                pass_type_id,
-            )
+            request = NotificationRequest(device_token=token, message={"aps": {}},)
+            await client.send_notification(request)
+
         except SSLError as e:
             logger.error("django_walletpass SSLError: %s", e)
-        except StreamResetError as e:
+        except (ConnectionError, ConnectionClosed) as e:
             logger.error("django_walletpass StreamResetError. Bad cert or token? %s", e)
         # Errors should never pass silently.
         except Exception as e:  # pylint: disable=broad-except
             # Unless explicitly silenced.
             logger.error("django_walletpass uncaught error %s", e)
 
-    def get_credentials(self):
-        if WALLETPASS_CONF['PUSH_AUTH_STRATEGY'] == 'token':
-            return TokenCredentials(
-                auth_key_path=WALLETPASS_CONF['TOKEN_AUTH_KEY_PATH'],
-                auth_key_id=WALLETPASS_CONF['TOKEN_AUTH_KEY_ID'],
-                team_id=WALLETPASS_CONF['TEAM_ID'],
-            )
-
-        # legacy cert/key auth
-        context = init_context(
-            cert=(
-                WALLETPASS_CONF['CERT_PATH'],
-                WALLETPASS_CONF['KEY_PATH'],
-            ),
-            # cert_path=WALLETPASS_CONF['APPLE_WWDRCA_PEM_PATH'],
-            cert_password=WALLETPASS_CONF['KEY_PASSWORD'],
-        )
-
-        return Credentials(context)
-
     def push_notification_with_token(self, token):
-        self.push_notification(
-            self.get_credentials(),
-            push_sandbox=WALLETPASS_CONF['PUSH_SANDBOX'],
-            pass_type_id=WALLETPASS_CONF['PASS_TYPE_ID'],
-            push_token=token,
+        client = APNs(
+            key=WALLETPASS_CONF["TOKEN_AUTH_KEY_PATH"],
+            key_id=WALLETPASS_CONF["TOKEN_AUTH_KEY_ID"],
+            team_id=WALLETPASS_CONF["TEAM_ID"],
+            topic=WALLETPASS_CONF["PASS_TYPE_ID"],
+            use_sandbox=WALLETPASS_CONF["PUSH_SANDBOX"],
         )
+        return self.loop.run_until_complete(self.push_notification(client, token))
 
     def push_notification_from_instance(self, registration_instance):
-        self.push_notification_with_token(registration_instance.push_token)
+        return self.push_notification_with_token(registration_instance.push_token)
 
     def push_notification_from_pk(self, registration_pk):
         registration = Registration.objects.get(pk=registration_pk)

django_walletpass/tests/main.py

@@ -1,6 +1,7 @@
+from unittest import mock
 from django.test import TestCase
 from django_walletpass import crypto
-from django_walletpass.models import PassBuilder
+from django_walletpass.models import Pass, PassBuilder, Registration
 from django_walletpass.settings import dwpconfig as WALLETPASS_CONF
 
 
@@ -56,3 +57,16 @@ def test_build_pkpass(self):
 
         self.assertNotEqual(builder.manifest_dict, builder3.manifest_dict)
         self.assertNotEqual(builder.pass_data, builder3.pass_data)
+
+
+class ModelTestCase(TestCase):
+    @mock.patch("django_walletpass.models.Pass.get_registrations")
+    @mock.patch("django_walletpass.services.APNs.send_notification")
+    def test_push_notification(self, send_notification_mock, get_registrations_mock):
+        get_registrations_mock.return_value = [Registration()]
+        pass_ = Pass(pk=1)
+        with mock.patch("django_walletpass.services.APNs.__init__", return_value=None):
+            pass_.push_notification()
+            send_notification_mock.assert_called_with(mock.ANY)
+            request = send_notification_mock.call_args_list[0][0][0]
+            self.assertEqual(request.message, {"aps": {}})

setup.py

@@ -9,8 +9,8 @@
 
 setup(
     name='django-walletpass',
-    python_requires='>=3.5.0',
-    version='2.0',
+    python_requires='>=3.6.0',
+    version='3.0',
     author='Develatio Technologies S.L.',
     author_email='[email protected]',
     packages=find_packages(),
@@ -20,7 +20,7 @@
     install_requires=[
         'Django>=2.0',
         'cryptography>=2.4.2',
-        'apns2>=0.7.1',
+        'aioapns~=2.2',
         'pyopenssl',
         'djangorestframework>=3.8',
     ],