diff --git a/README.rst b/README.rst index 7367be24..b65872c7 100644 --- a/README.rst +++ b/README.rst @@ -64,7 +64,7 @@ Edit your settings.py file: "WNS_PACKAGE_SECURITY_ID": "[your package security id, e.g: 'ms-app://e-3-4-6234...']", "WNS_SECRET_KEY": "[your app secret key, e.g.: 'KDiejnLKDUWodsjmewuSZkk']", "WP_PRIVATE_KEY": "/path/to/your/private.pem", - "WP_CLAIMS": {'sub': "mailto: development@example.com"} + "WP_CLAIMS": {'sub': "mailto:development@example.com"} } .. note:: @@ -122,278 +122,11 @@ For WNS, you need both the ``WNS_PACKAGE_SECURITY_KEY`` and the ``WNS_SECRET_KEY **WP settings** -- Install: - -.. code-block:: python - - pip install pywebpush - pip install py-vapid (Only for generating key) - -- Getting keys: - - - Create file (claim.json) like this: - -.. code-block:: bash - - { - "sub": "mailto: development@example.com", - "aud": "https://android.googleapis.com" - } - - - Generate public and private keys: - -.. code-block:: bash - - vapid --sign claim.json - - No private_key.pem file found. - Do you want me to create one for you? (Y/n)Y - Do you want me to create one for you? (Y/n)Y - Generating private_key.pem - Generating public_key.pem - Include the following headers in your request: - - Crypto-Key: p256ecdsa=BEFuGfKKEFp-kEBMxAIw7ng8HeH_QwnH5_h55ijKD4FRvgdJU1GVlDo8K5U5ak4cMZdQTUJlkA34llWF0xHya70 - - Authorization: WebPush eyJ0eXAiOiJKV1QiLCJhbGciOiJFUzI1NiJ9.eyJhdWQiOiJodHRwczovL2FuZHJvaWQuZ29vZ2xlYXBpcy5jb20iLCJleHAiOiIxNTA4NDkwODM2Iiwic3ViIjoibWFpbHRvOiBkZXZlbG9wbWVudEBleGFtcGxlLmNvbSJ9.r5CYMs86X3JZ4AEs76pXY5PxsnEhIFJ-0ckbibmFHZuyzfIpf1ZGIJbSI7knA4ufu7Hm8RFfEg5wWN1Yf-dR2A - - - Generate client public key (applicationServerKey) - -.. code-block:: bash - - vapid --applicationServerKey - - Application Server Key = BEFuGfKKEFp-kEBMxAIw7ng8HeH_QwnH5_h55ijKD4FRvgdJU1GVlDo8K5U5ak4cMZdQTUJlkA34llWF0xHya70 - - -- Configure settings: - - ``WP_PRIVATE_KEY``: Absolute path to your private certificate file: os.path.join(BASE_DIR, "private_key.pem") -- ``WP_CLAIMS``: Dictionary with the same sub info like claims file: {'sub': "mailto: development@example.com"} +- ``WP_CLAIMS``: Dictionary with default value for the sub, (subject), sent to the webpush service, This would be used by the service if they needed to reach out to you (the sender). Could be a url or mailto e.g. {'sub': "mailto:development@example.com"}. - ``WP_ERROR_TIMEOUT``: The timeout on WebPush POSTs. (Optional) -- ``WP_POST_URL``: A dictionary (key per browser supported) with the full url that webpush notifications will be POSTed to. (Optional) - - -- Configure client (javascript): - -.. code-block:: javascript - - // Utils functions: - - function urlBase64ToUint8Array (base64String) { - var padding = '='.repeat((4 - base64String.length % 4) % 4) - var base64 = (base64String + padding) - .replace(/\-/g, '+') - .replace(/_/g, '/') - - var rawData = window.atob(base64) - var outputArray = new Uint8Array(rawData.length) - for (var i = 0; i < rawData.length; ++i) { - outputArray[i] = rawData.charCodeAt(i) - } - return outputArray; - } - - function loadVersionBrowser () { - if ("userAgentData" in navigator) { - // navigator.userAgentData is not available in - // Firefox and Safari - const uaData = navigator.userAgentData; - // Outputs of navigator.userAgentData.brands[n].brand are e.g. - // Chrome: 'Google Chrome' - // Edge: 'Microsoft Edge' - // Opera: 'Opera' - let browsername; - let browserversion; - let chromeVersion = null; - for (var i = 0; i < uaData.brands.length; i++) { - let brand = uaData.brands[i].brand; - browserversion = uaData.brands[i].version; - if (brand.match(/opera|chrome|edge|safari|firefox|msie|trident/i) !== null) { - // If we have a chrome match, save the match, but try to find another match - // E.g. Edge can also produce a false Chrome match. - if (brand.match(/chrome/i) !== null) { - chromeVersion = browserversion; - } - // If this is not a chrome match return immediately - else { - browsername = brand.substr(brand.indexOf(' ')+1); - return { - name: browsername, - version: browserversion - } - } - } - } - // No non-Chrome match was found. If we have a chrome match, return it. - if (chromeVersion !== null) { - return { - name: "chrome", - version: chromeVersion - } - } - } - // If no userAgentData is not present, or if no match via userAgentData was found, - // try to extract the browser name and version from userAgent - const userAgent = navigator.userAgent; - var ua = userAgent, tem, M = ua.match(/(opera|chrome|safari|firefox|msie|trident(?=\/))\/?\s*(\d+)/i) || []; - if (/trident/i.test(M[1])) { - tem = /\brv[ :]+(\d+)/g.exec(ua) || []; - return {name: 'IE', version: (tem[1] || '')}; - } - if (M[1] === 'Chrome') { - tem = ua.match(/\bOPR\/(\d+)/); - if (tem != null) { - return {name: 'Opera', version: tem[1]}; - } - } - M = M[2] ? [M[1], M[2]] : [navigator.appName, navigator.appVersion, '-?']; - if ((tem = ua.match(/version\/(\d+)/i)) != null) { - M.splice(1, 1, tem[1]); - } - return { - name: M[0], - version: M[1] - }; - }; - var applicationServerKey = "BEFuGfKKEFp-kEBMxAIw7ng8HeH_QwnH5_h55ijKD4FRvgdJU1GVlDo8K5U5ak4cMZdQTUJlkA34llWF0xHya70"; - .... - - // In your ready listener - if ('serviceWorker' in navigator) { - // The service worker has to store in the root of the app - // http://stackoverflow.com/questions/29874068/navigator-serviceworker-is-never-ready - var browser = loadVersionBrowser(navigator.userAgent); - navigator.serviceWorker.register('navigatorPush.service.js?version=1.0.0').then(function (reg) { - reg.pushManager.subscribe({ - userVisibleOnly: true, - applicationServerKey: urlBase64ToUint8Array(applicationServerKey) - }).then(function (sub) { - var endpointParts = sub.endpoint.split('/'); - var registration_id = endpointParts[endpointParts.length - 1]; - var data = { - 'browser': browser.name.toUpperCase(), - 'p256dh': btoa(String.fromCharCode.apply(null, new Uint8Array(sub.getKey('p256dh')))), - 'auth': btoa(String.fromCharCode.apply(null, new Uint8Array(sub.getKey('auth')))), - 'name': 'XXXXX', - 'registration_id': registration_id - }; - requestPOSTToServer(data); - }) - }).catch(function (err) { - console.log(':^(', err); - }); - - - - - // Example navigatorPush.service.js file - - var getTitle = function (title) { - if (title === "") { - title = "TITLE DEFAULT"; - } - return title; - }; - var getNotificationOptions = function (message, message_tag) { - var options = { - body: message, - icon: '/img/icon_120.png', - tag: message_tag, - vibrate: [200, 100, 200, 100, 200, 100, 200] - }; - return options; - }; - - self.addEventListener('install', function (event) { - self.skipWaiting(); - }); - - self.addEventListener('push', function(event) { - try { - // Push is a JSON - var response_json = event.data.json(); - var title = response_json.title; - var message = response_json.message; - var message_tag = response_json.tag; - } catch (err) { - // Push is a simple text - var title = ""; - var message = event.data.text(); - var message_tag = ""; - } - self.registration.showNotification(getTitle(title), getNotificationOptions(message, message_tag)); - // Optional: Comunicating with our js application. Send a signal - self.clients.matchAll({includeUncontrolled: true, type: 'window'}).then(function (clients) { - clients.forEach(function (client) { - client.postMessage({ - "data": message_tag, - "data_title": title, - "data_body": message}); - }); - }); - }); - - // Optional: Added to that the browser opens when you click on the notification push web. - self.addEventListener('notificationclick', function(event) { - // Android doesn't close the notification when you click it - // See http://crbug.com/463146 - event.notification.close(); - // Check if there's already a tab open with this URL. - // If yes: focus on the tab. - // If no: open a tab with the URL. - event.waitUntil(clients.matchAll({type: 'window', includeUncontrolled: true}).then(function(windowClients) { - for (var i = 0; i < windowClients.length; i++) { - var client = windowClients[i]; - if ('focus' in client) { - return client.focus(); - } - } - }) - ); - }); - -The above code makes a call to ``requestPOSTToServer()``, which is not implemented. This is where you should make a call to your Django app to register the Web Push device. Your implementation will vary depending on the design of your web app, here's an example using a Django view and an Ajax call: - -Example Django view to handle registration: - -.. code-block:: python - - from django.http import JsonResponse - from push_notifications.models import WebPushDevice - - def register_wp_notifications(request): - WebPushDevice.objects.create( - registration_id=request.GET.get('registration_id'), - p256dh=request.GET.get('p256dh'), - auth=request.GET.get('auth'), - browser=request.GET.get('browser'), - user=request.user, - ) - data = { - 'result': True - } - return JsonResponse(data) - -Example JavaScript to send registration (requires jQuery): - -.. code-block:: javascript - - function requestPOSTToServer ( data ) { - $.ajax({ - url: '/PATH/DEFINED/IN/URLS.PY/', - data: { - 'browser': data.browser, - 'p256dh': data.p256dh, - 'auth': data.auth, - 'registration_id': data.registration_id - }, - dataType: 'json', - success: function (data) { - } - }); - } +For more information about how to configure WebPush, see `docs/WebPush `_. Sending messages diff --git a/docs/WebPush.rst b/docs/WebPush.rst new file mode 100644 index 00000000..bee495a2 --- /dev/null +++ b/docs/WebPush.rst @@ -0,0 +1,225 @@ +At a high-level, the key steps for implementing web push notifications after installing django-push-notifications[WP] are: + - Configure the VAPID keys, a private and public key for signing your push requests. + - Add client side logic to ask the user for permission to send push notifications and then sending returned client identifier information to a django view to create a WebPushDevice. + - Use a service worker to receive messages that have been pushed to the device and displaying them as notifications. + +These are in addition to the instalation steps for django-push-notifications[WP] + +Configure the VAPID keys +------------------------------ +- Install: + +.. code-block:: python + + pip install py-vapid (Only for generating key) + +- Generate public and private keys: + +.. code-block:: bash + + vapid --gen + + Generating private_key.pem + Generating public_key.pem + + +The private key generated is the file to use with the setting ``WP_PRIVATE_KEY`` +The public key will be used in your client side javascript, but first it must be formated as an Application Server Key + +- Generate client public key (applicationServerKey) + +.. code-block:: bash + + vapid --applicationServerKey + + Application Server Key = + + + +Client Side logic to ask user for permission and subscribe to WebPush +------------------------------ +The example subscribeUser function is best called in response to a user action, such as a button click. Some browsers will deny the request otherwise. + +.. code-block:: javascript + + // Utils functions: + + function urlBase64ToUint8Array (base64String) { + var padding = '='.repeat((4 - base64String.length % 4) % 4) + var base64 = (base64String + padding) + .replace(/\-/g, '+') + .replace(/_/g, '/') + + var rawData = window.atob(base64) + var outputArray = new Uint8Array(rawData.length) + + for (var i = 0; i < rawData.length; ++i) { + outputArray[i] = rawData.charCodeAt(i) + } + return outputArray; + } + + var applicationServerKey = ''; + + function subscribeUser() { + if ('Notification' in window && 'serviceWorker' in navigator) { + navigator.serviceWorker.ready.then(function (reg) { + reg.pushManager + .subscribe({ + userVisibleOnly: true, + applicationServerKey: urlBase64ToUint8Array( + applicationServerKey + ), + }) + .then(function (sub) { + var registration_id = sub.endpoint; + var data = { + p256dh: btoa( + String.fromCharCode.apply( + null, + new Uint8Array(sub.getKey('p256dh')) + ) + ), + auth: btoa( + String.fromCharCode.apply( + null, + new Uint8Array(sub.getKey('auth')) + ) + ), + registration_id: registration_id, + } + requestPOSTToServer(data) + }) + .catch(function (e) { + if (Notification.permission === 'denied') { + console.warn('Permission for notifications was denied') + } else { + console.error('Unable to subscribe to push', e) + } + }) + }) + } + } + + // Send the subscription data to your server + function requestPOSTToServer (data) { + const headers = new Headers(); + headers.set('Content-Type', 'application/json'); + const requestOptions = { + method: 'POST', + headers, + body: JSON.stringify(data), + }; + + return ( + fetch( + '', + requestOptions + ) + ).then((response) => response.json()) + } + +Server Side logic to create webpush +------------------------------ +Is is up to you how to add a view in your django application that can handle a POST of p256dh, auth, registration_id and create a WebPushDevice with those values assoicated with the appropriate user. +For example you could use rest_framework + +.. code-block:: python + + from rest_framework.routers import SimpleRouter + from push_notifications.api.rest_framework import WebPushDeviceViewSet + .... + api_router = SimpleRouter() + api_router.register(r'push/web', WebPushDeviceViewSet, basename='web_push') + ... + urlpatterns += [ + # Api + re_path('api/v1/', include(api_router.urls)), + ... + ] + +Or a generic function view (add your own boilerplate for errors and protections) + +.. code-block:: python + + import json + from push_notifications.models import WebPushDevice + def register_webpush(request): + data = json.loads(request.body) + WebPushDevice.objects.create( + user=request.user, + **data + ) + + +Service Worker to show messages +------------------------------ +You will need a service worker registered with your web app that can handle the notfications, for example + +.. code-block:: javascript + + // Example navigatorPush.service.js file + + var getTitle = function (title) { + if (title === "") { + title = "TITLE DEFAULT"; + } + return title; + }; + var getNotificationOptions = function (message, message_tag) { + var options = { + body: message, + icon: '/img/icon_120.png', + tag: message_tag, + vibrate: [200, 100, 200, 100, 200, 100, 200] + }; + return options; + }; + + self.addEventListener('install', function (event) { + self.skipWaiting(); + }); + + self.addEventListener('push', function(event) { + try { + // Push is a JSON + var response_json = event.data.json(); + var title = response_json.title; + var message = response_json.message; + var message_tag = response_json.tag; + } catch (err) { + // Push is a simple text + var title = ""; + var message = event.data.text(); + var message_tag = ""; + } + self.registration.showNotification(getTitle(title), getNotificationOptions(message, message_tag)); + // Optional: Comunicating with our js application. Send a signal + self.clients.matchAll({includeUncontrolled: true, type: 'window'}).then(function (clients) { + clients.forEach(function (client) { + client.postMessage({ + "data": message_tag, + "data_title": title, + "data_body": message}); + }); + }); + }); + + // Optional: Added to that the browser opens when you click on the notification push web. + self.addEventListener('notificationclick', function(event) { + // Android doesn't close the notification when you click it + // See http://crbug.com/463146 + event.notification.close(); + // Check if there's already a tab open with this URL. + // If yes: focus on the tab. + // If no: open a tab with the URL. + event.waitUntil(clients.matchAll({type: 'window', includeUncontrolled: true}).then(function(windowClients) { + for (var i = 0; i < windowClients.length; i++) { + var client = windowClients[i]; + if ('focus' in client) { + return client.focus(); + } + } + }) + ); + }); diff --git a/push_notifications/models.py b/push_notifications/models.py index d9e1174a..79faad0b 100644 --- a/push_notifications/models.py +++ b/push_notifications/models.py @@ -255,6 +255,4 @@ def device_id(self): def send_message(self, message, **kwargs): from .webpush import webpush_send_message - return webpush_send_message( - uri=self.registration_id, message=message, browser=self.browser, - auth=self.auth, p256dh=self.p256dh, application_id=self.application_id, **kwargs) + return webpush_send_message(self, message, **kwargs) diff --git a/push_notifications/webpush.py b/push_notifications/webpush.py index e594fff1..de379431 100644 --- a/push_notifications/webpush.py +++ b/push_notifications/webpush.py @@ -1,3 +1,5 @@ +import warnings + from pywebpush import WebPushException, webpush from .conf import get_manager @@ -5,9 +7,18 @@ def get_subscription_info(application_id, uri, browser, auth, p256dh): - url = get_manager().get_wp_post_url(application_id, browser) + if uri.startswith("https://"): + endpoint = uri + else: + url = get_manager().get_wp_post_url(application_id, browser) + endpoint = "{}/{}".format(url, uri) + warnings.warn( + "registration_id should be the full endpoint returned from pushManager.subscribe", + DeprecationWarning, + stacklevel=2, + ) return { - "endpoint": "{}/{}".format(url, uri), + "endpoint": endpoint, "keys": { "auth": auth, "p256dh": p256dh, @@ -15,25 +26,30 @@ def get_subscription_info(application_id, uri, browser, auth, p256dh): } -def webpush_send_message( - uri, message, browser, auth, p256dh, application_id=None, **kwargs -): - subscription_info = get_subscription_info(application_id, uri, browser, auth, p256dh) - +def webpush_send_message(device, message, **kwargs): + subscription_info = get_subscription_info( + device.application_id, device.registration_id, + device.browser, device.auth, device.p256dh) try: + results = {"results": [{"original_registration_id": device.registration_id}]} response = webpush( subscription_info=subscription_info, data=message, - vapid_private_key=get_manager().get_wp_private_key(application_id), - vapid_claims=get_manager().get_wp_claims(application_id).copy(), + vapid_private_key=get_manager().get_wp_private_key(device.application_id), + vapid_claims=get_manager().get_wp_claims(device.application_id).copy(), **kwargs ) - results = {"results": [{}]} - if not response.ok: - results["results"][0]["error"] = response.content - results["results"][0]["original_registration_id"] = response.content - else: + if response.ok: results["success"] = 1 + else: + results["failure"] = 1 + results["results"][0]["error"] = response.content return results except WebPushException as e: + if e.response is not None and e.response.status_code in [404, 410]: + results["failure"] = 1 + results["results"][0]["error"] = e.message + device.active = False + device.save() + return results raise WebPushError(e.message) diff --git a/setup.cfg b/setup.cfg index b68daaa1..19c8c08a 100644 --- a/setup.cfg +++ b/setup.cfg @@ -38,7 +38,6 @@ setup_requires = APNS = apns2>=0.3.0 importlib-metadata;python_version < "3.8" - pywebpush>=1.3.0 Django>=2.2 WP = pywebpush>=1.3.0 diff --git a/tests/settings.py b/tests/settings.py index 09c56c28..a5da8a5e 100644 --- a/tests/settings.py +++ b/tests/settings.py @@ -26,5 +26,5 @@ SECRET_KEY = "foobar" PUSH_NOTIFICATIONS_SETTINGS = { - "WP_CLAIMS": {"sub": "mailto: jazzband@example.com"} + "WP_CLAIMS": {"sub": "mailto:jazzband@example.com"} } diff --git a/tests/settings_unique.py b/tests/settings_unique.py index f88e39a3..e7429943 100644 --- a/tests/settings_unique.py +++ b/tests/settings_unique.py @@ -26,6 +26,6 @@ SECRET_KEY = "foobar" PUSH_NOTIFICATIONS_SETTINGS = { - "WP_CLAIMS": {"sub": "mailto: jazzband@example.com"}, + "WP_CLAIMS": {"sub": "mailto:jazzband@example.com"}, "UNIQUE_REG_ID": True } diff --git a/tests/test_legacy_config.py b/tests/test_legacy_config.py index 2dfc55c7..a11cd8c1 100644 --- a/tests/test_legacy_config.py +++ b/tests/test_legacy_config.py @@ -1,3 +1,5 @@ +from unittest import mock + from django.core.exceptions import ImproperlyConfigured from django.test import TestCase @@ -38,9 +40,17 @@ def test_get_error_timeout(self): ) def test_immutable_wp_claims(self): + self.endpoint = "https://updates.push.services.mozilla.com/wpush/v2/token" + self.mock_device = mock.Mock() + self.mock_device.application_id = None + self.mock_device.registration_id = self.endpoint + self.mock_device.auth = "authtest" + self.mock_device.p256dh = "p256dhtest" + self.mock_device.active = True + self.mock_device.save.return_value = True vapid_claims_pre = get_manager().get_wp_claims(None).copy() try: - webpush_send_message("", {}, "CHROME", "", "") + webpush_send_message(self.mock_device, "message") except WebPushError: pass vapid_claims_after = get_manager().get_wp_claims(None) diff --git a/tests/test_rest_framework.py b/tests/test_rest_framework.py index 24c8d7a4..0f5dd257 100644 --- a/tests/test_rest_framework.py +++ b/tests/test_rest_framework.py @@ -11,7 +11,7 @@ class APNSDeviceSerializerTestCase(TestCase): def test_validation(self): - # valid data - 32 bytes upper case + # valid data - 64 bytes upper case serializer = APNSDeviceSerializer(data={ "registration_id": "AEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAE", "name": "Apple iPhone 6+", @@ -20,7 +20,7 @@ def test_validation(self): }) self.assertTrue(serializer.is_valid()) - # valid data - 32 bytes lower case + # valid data - 64 bytes lower case serializer = APNSDeviceSerializer(data={ "registration_id": "aeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeae", "name": "Apple iPhone 6+", @@ -31,7 +31,7 @@ def test_validation(self): # valid data - 100 bytes upper case serializer = APNSDeviceSerializer(data={ - "registration_id": "AE" * 100, + "registration_id": "AE" * 50, "name": "Apple iPhone 6+", "device_id": "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF", }) @@ -39,7 +39,7 @@ def test_validation(self): # valid data - 100 bytes lower case serializer = APNSDeviceSerializer(data={ - "registration_id": "ae" * 100, + "registration_id": "ae" * 50, "name": "Apple iPhone 6+", "device_id": "ffffffffffffffffffffffffffffffff", }) @@ -47,7 +47,7 @@ def test_validation(self): # valid data - 200 bytes mixed case serializer = APNSDeviceSerializer(data={ - "registration_id": "aE" * 200, + "registration_id": "aE" * 100, "name": "Apple iPhone 6+", "device_id": "ffffffffffffffffffffffffffffffff", }) diff --git a/tests/test_webpush.py b/tests/test_webpush.py new file mode 100644 index 00000000..bbdbdf70 --- /dev/null +++ b/tests/test_webpush.py @@ -0,0 +1,93 @@ +from unittest import mock + +from django.test import TestCase +from pywebpush import WebPushException + +from push_notifications.exceptions import WebPushError +from push_notifications.webpush import ( + get_subscription_info, webpush_send_message +) + +# Mock Responses +mock_success_response = mock.MagicMock(status_code=200, ok=True) +mock_fail_resposne = mock.MagicMock(status_code=400, ok=False, content="Test Error") +mock_unsubscribe_response = mock.MagicMock( + status_code=410, ok=False, content="Unsubscribe") +mock_unsubscribe_response_404 = mock.MagicMock( + status_code=404, ok=False, content="Unsubscribe") + + +class WebPushSendMessageTestCase(TestCase): + def setUp(self): + self.endpoint = "https://updates.push.services.mozilla.com/wpush/v2/token" + self.mock_device = mock.Mock() + self.mock_device.application_id = None + self.mock_device.registration_id = self.endpoint + self.mock_device.auth = "authtest" + self.mock_device.p256dh = "p256dhtest" + self.mock_device.active = True + self.mock_device.save.return_value = True + + def test_get_subscription_info(self): + keys = {"auth": "authtest", "p256dh": "p256dhtest"} + endpoint = self.endpoint + original = get_subscription_info( + None, "token", "FIREFOX", keys["auth"], keys["p256dh"] + ) + + self.assertEqual( + original, + { + "endpoint": endpoint, + "keys": keys, + }, + ) + + patched = get_subscription_info( + None, + endpoint, + "", + keys["auth"], + keys["p256dh"], + ) + + self.assertEqual( + patched, + { + "endpoint": endpoint, + "keys": keys, + }, + ) + + @mock.patch("push_notifications.webpush.webpush", return_value=mock_success_response) + def test_webpush_send_message(self, webpush_mock): + results = webpush_send_message(self.mock_device, "message") + self.assertEqual(results["success"], 1) + + @mock.patch("push_notifications.webpush.webpush", return_value=mock_fail_resposne) + def test_webpush_send_message_failure(self, webpush_mock): + results = webpush_send_message(self.mock_device, "message") + self.assertEqual(results["failure"], 1) + + @mock.patch( + "push_notifications.webpush.webpush", + side_effect=WebPushException("Unsubscribe", + response=mock_unsubscribe_response)) + def test_webpush_send_message_unsubscribe(self, webpush_mock): + results = webpush_send_message(self.mock_device, "message") + self.assertEqual(results["failure"], 1) + + @mock.patch( + "push_notifications.webpush.webpush", + side_effect=WebPushException("Unsubscribe", + response=mock_unsubscribe_response_404)) + def test_webpush_send_message_404(self, webpush_mock): + results = webpush_send_message(self.mock_device, "message") + self.assertEqual(results["failure"], 1) + + @mock.patch( + "push_notifications.webpush.webpush", + side_effect=WebPushException("Error")) + def test_webpush_send_message_exception(self, webpush_mock): + with self.assertRaises(WebPushError): + webpush_send_message(self.mock_device, "message")