fix NotificationResult import, add test for err_func param

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"):