fix NotificationResult import, add test for err_func param

Author: pomali

push_notifications/apns_async.py

@@ -1,9 +1,10 @@
 import asyncio
-from dataclasses import asdict, dataclass
 import time
+from dataclasses import asdict, dataclass
 from typing import Awaitable, Callable, Dict, Optional, Union
 
-from aioapns import APNs, NotificationRequest, ConnectionError, NotificationResult
+from aioapns import APNs, ConnectionError, NotificationRequest
+from aioapns.common import NotificationResult
 
 from . import models
 from .conf import get_manager
@@ -134,9 +135,12 @@ def send_message(
 		self,
 		request: NotificationRequest,
 	):
+		print("a")
 		loop = asyncio.get_event_loop()
 		res1 = self.client.send_notification(request)
+		print("b", res1)
 		res = loop.run_until_complete(res1)
+		print("c", res)
 		return res
 
 	def _create_notification_request_from_args(

tests/test_apns_async_push_payload.py

@@ -5,6 +5,7 @@
 import pytest
 from django.test import TestCase
 
+from aioapns.common import NotificationResult
 
 try:
 	from push_notifications.apns_async import TokenCredentials, apns_send_message
@@ -53,11 +54,7 @@ def test_push_payload_with_thread_id(self, mock_apns):
 			sound="chime",
 			extra={"custom_data": 12345},
 			expiration=int(time.time()) + 3,
-			creds=TokenCredentials(
-				key="aaa",
-				key_id="bbb",
-				team_id="ccc",
-			),
+			creds=TokenCredentials(key="aaa", key_id="bbb", team_id="ccc"),
 		)
 		args, kwargs = mock_apns.return_value.send_notification.call_args
 		req = args[0]
@@ -159,6 +156,34 @@ def test_collapse_id(self, mock_apns):
 		self.assertEqual(req.message["aps"]["alert"], "sample")
 		self.assertEqual(req.collapse_key, "456789")
 
+	@mock.patch("aioapns.client.APNsCertConnectionPool", autospec=True)
+	@mock.patch("aioapns.client.APNsKeyConnectionPool", autospec=True)
+	def test_aioapns_err_func(self, mock_cert_pool, mock_key_pool):
+		mock_cert_pool.return_value.send_notification = mock.AsyncMock()
+		result =  NotificationResult(
+			"123", "400"
+		)
+		mock_cert_pool.return_value.send_notification.return_value = result
+		err_func = mock.AsyncMock()
+		with pytest.raises(Exception):
+			apns_send_message(
+				"123",
+				"sample",
+				creds=TokenCredentials(
+					key="aaa",
+					key_id="bbb",
+					team_id="ccc",
+				),
+				topic="default",
+				err_func=err_func,
+			)
+		mock_cert_pool.assert_called_once()
+		mock_cert_pool.return_value.send_notification.assert_called_once()
+		mock_cert_pool.return_value.send_notification.assert_awaited_once()
+		err_func.assert_called_with(
+			mock.ANY, result
+		)
+
 	# def test_bad_priority(self):
 	# 	with mock.patch("apns2.credentials.init_context"):
 	# 		with mock.patch("apns2.client.APNsClient.connect"):