Clean up the code style and add .editorconfig and TODO.md
This commit is contained in:
Jerome Leclanche
parent
f32d9d50d4
commit
565a638b89
|
|
@ -0,0 +1,16 @@
|
|||
# EditorConfig: http://editorconfig.org
|
||||
root = true
|
||||
|
||||
[*]
|
||||
charset = utf-8
|
||||
end_of_line = lf
|
||||
indent_size = 4
|
||||
indent_style = tab
|
||||
quote_type = double
|
||||
insert_final_newline = true
|
||||
tab_width = 4
|
||||
trim_trailing_whitespace = true
|
||||
|
||||
[*.py]
|
||||
spaces_around_brackets = none
|
||||
spaces_around_operators = true
|
||||
|
|
@ -1,6 +1,8 @@
|
|||
### Coding style
|
||||
|
||||
Please adhere to the coding style throughout the project.
|
||||
This project follows the [HearthSim Styleguide](https://hearthsim.info/styleguide/).
|
||||
|
||||
In short:
|
||||
|
||||
1. Always use tabs. [Here](https://leclan.ch/tabs) is a short explanation why tabs are preferred.
|
||||
2. Always use double quotes for strings, unless single quotes avoid unnecessary escapes.
|
||||
|
|
@ -19,6 +21,7 @@ Keep the commit log as healthy as the code. It is one of the first places new co
|
|||
3. Follow [these conventions](http://chris.beams.io/posts/git-commit/) when writing the commit message
|
||||
|
||||
When filing a Pull Request, make sure it is rebased on top of most recent master.
|
||||
If you need to modify it or amend it in some way, you should always appropriately [fixup](https://help.github.com/articles/about-git-rebase/) the issues in git and force-push your changes to your fork.
|
||||
If you need to modify it or amend it in some way, you should always appropriately
|
||||
[fixup](https://help.github.com/articles/about-git-rebase/) the issues in git and force-push your changes to your fork.
|
||||
|
||||
Also see: [Github Help: Using Pull Requests](https://help.github.com/articles/using-pull-requests/)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,14 @@
|
|||
These are the main big items that still need to be done:
|
||||
|
||||
## General
|
||||
|
||||
* Use third party libraries to implement APNS and GCM.
|
||||
* Clean up the tests accordingly. If APNS/GCM etc functionality is being tested
|
||||
in an upstream project, those tests don't need to be duplicated here.
|
||||
Django Push Notifications should be testing the integration of those backends.
|
||||
* Review the impact of #260 and merge it. This can bring in a 2.0 release.
|
||||
|
||||
## Backends
|
||||
|
||||
* WebPush (#276)
|
||||
* FCM (#302)
|
||||
|
|
@ -1,4 +1,3 @@
|
|||
|
||||
__author__ = "Jerome Leclanche"
|
||||
__email__ = "jerome@leclan.ch"
|
||||
__version__ = "1.4.1"
|
||||
|
|
|
|||
|
|
@ -42,8 +42,10 @@ class DeviceAdmin(admin.ModelAdmin):
|
|||
break
|
||||
|
||||
if errors:
|
||||
self.message_user(request, _("Some messages could not be processed: %r" % (", ".join(errors))),
|
||||
level=messages.ERROR)
|
||||
self.message_user(
|
||||
request, _("Some messages could not be processed: %r" % (", ".join(errors))),
|
||||
level=messages.ERROR
|
||||
)
|
||||
if ret:
|
||||
if not bulk:
|
||||
ret = ", ".join(ret)
|
||||
|
|
|
|||
|
|
@ -33,18 +33,18 @@ class APNSDataOverflow(APNSError):
|
|||
|
||||
|
||||
def _check_certificate(ss):
|
||||
mode = 'start'
|
||||
for s in ss.split('\n'):
|
||||
if mode == 'start':
|
||||
if 'BEGIN RSA PRIVATE KEY' in s:
|
||||
mode = 'key'
|
||||
elif mode == 'key':
|
||||
if 'END RSA PRIVATE KEY' in s:
|
||||
mode = 'end'
|
||||
mode = "start"
|
||||
for s in ss.split("\n"):
|
||||
if mode == "start":
|
||||
if "BEGIN RSA PRIVATE KEY" in s:
|
||||
mode = "key"
|
||||
elif mode == "key":
|
||||
if "END RSA PRIVATE KEY" in s:
|
||||
mode = "end"
|
||||
break
|
||||
elif s.startswith('Proc-Type') and 'ENCRYPTED' in s:
|
||||
elif s.startswith("Proc-Type") and "ENCRYPTED" in s:
|
||||
raise Exception("The certificate private key should not be encrypted")
|
||||
if mode != 'end':
|
||||
if mode != "end":
|
||||
raise Exception("The certificate doesn't contain a private key")
|
||||
|
||||
|
||||
|
|
@ -86,7 +86,8 @@ def _apns_create_socket_to_feedback(certfile=None):
|
|||
def _apns_pack_frame(token_hex, payload, identifier, expiration, priority):
|
||||
token = unhexlify(token_hex)
|
||||
# |COMMAND|FRAME-LEN|{token}|{payload}|{id:4}|{expiration:4}|{priority:1}
|
||||
frame_len = 3 * 5 + len(token) + len(payload) + 4 + 4 + 1 # 5 items, each 3 bytes prefix, then each item length
|
||||
# 5 items, each 3 bytes prefix, then each item length
|
||||
frame_len = 3 * 5 + len(token) + len(payload) + 4 + 4 + 1
|
||||
frame_fmt = "!BIBH%ssBH%ssBHIBHIBHB" % (len(token), len(payload))
|
||||
frame = struct.pack(
|
||||
frame_fmt,
|
||||
|
|
@ -95,7 +96,8 @@ def _apns_pack_frame(token_hex, payload, identifier, expiration, priority):
|
|||
2, len(payload), payload,
|
||||
3, 4, identifier,
|
||||
4, 4, expiration,
|
||||
5, 1, priority)
|
||||
5, 1, priority
|
||||
)
|
||||
|
||||
return frame
|
||||
|
||||
|
|
@ -123,9 +125,11 @@ def _apns_check_errors(sock):
|
|||
sock.settimeout(saved_timeout)
|
||||
|
||||
|
||||
def _apns_send(token, alert, badge=None, sound=None, category=None, content_available=False,
|
||||
def _apns_send(
|
||||
token, alert, badge=None, sound=None, category=None, content_available=False,
|
||||
action_loc_key=None, loc_key=None, loc_args=[], extra={}, identifier=0,
|
||||
expiration=None, priority=10, socket=None, certfile=None):
|
||||
expiration=None, priority=10, socket=None, certfile=None
|
||||
):
|
||||
data = {}
|
||||
aps_data = {}
|
||||
|
||||
|
|
@ -189,7 +193,7 @@ def _apns_receive_feedback(socket):
|
|||
expired_token_list = []
|
||||
|
||||
# read a timestamp (4 bytes) and device token length (2 bytes)
|
||||
header_format = '!LH'
|
||||
header_format = "!LH"
|
||||
has_data = True
|
||||
while has_data:
|
||||
try:
|
||||
|
|
@ -198,7 +202,7 @@ def _apns_receive_feedback(socket):
|
|||
if header_data is not None:
|
||||
timestamp, token_length = header_data
|
||||
# Unpack format for a single value of length bytes
|
||||
token_format = '%ss' % token_length
|
||||
token_format = "%ss" % token_length
|
||||
device_token = _apns_read_and_unpack(socket, token_format)
|
||||
if device_token is not None:
|
||||
# _apns_read_and_unpack returns a tuple, but
|
||||
|
|
@ -255,6 +259,6 @@ def apns_fetch_inactive_ids(certfile=None):
|
|||
inactive_ids = []
|
||||
# Maybe we should have a flag to return the timestamp?
|
||||
# It doesn't seem that useful right now, though.
|
||||
for tStamp, registration_id in _apns_receive_feedback(socket):
|
||||
inactive_ids.append(codecs.encode(registration_id, 'hex_codec'))
|
||||
for ts, registration_id in _apns_receive_feedback(socket):
|
||||
inactive_ids.append(codecs.encode(registration_id, "hex_codec"))
|
||||
return inactive_ids
|
||||
|
|
|
|||
|
|
@ -48,7 +48,9 @@ class HexadecimalField(forms.CharField):
|
|||
A form field that accepts only hexadecimal numbers
|
||||
"""
|
||||
def __init__(self, *args, **kwargs):
|
||||
self.default_validators = [RegexValidator(hex_re, _("Enter a valid hexadecimal number"), "invalid")]
|
||||
self.default_validators = [
|
||||
RegexValidator(hex_re, _("Enter a valid hexadecimal number"), "invalid")
|
||||
]
|
||||
super(HexadecimalField, self).__init__(*args, **kwargs)
|
||||
|
||||
def prepare_value(self, value):
|
||||
|
|
|
|||
|
|
@ -37,7 +37,9 @@ def _chunks(l, n):
|
|||
def _gcm_send(data, content_type):
|
||||
key = SETTINGS.get("GCM_API_KEY")
|
||||
if not key:
|
||||
raise ImproperlyConfigured('You need to set PUSH_NOTIFICATIONS_SETTINGS["GCM_API_KEY"] to send messages through GCM.')
|
||||
raise ImproperlyConfigured(
|
||||
'You need to set PUSH_NOTIFICATIONS_SETTINGS["GCM_API_KEY"] to send messages through GCM.'
|
||||
)
|
||||
|
||||
headers = {
|
||||
"Content-Type": content_type,
|
||||
|
|
@ -50,7 +52,8 @@ def _gcm_send(data, content_type):
|
|||
|
||||
def _gcm_send_plain(registration_id, data, **kwargs):
|
||||
"""
|
||||
Sends a GCM notification to a single registration_id or to a topic (If "topic" included in the kwargs).
|
||||
Sends a GCM notification to a single registration_id or to a
|
||||
topic (If "topic" included in the kwargs).
|
||||
This will send the notification as form data.
|
||||
If sending multiple notifications, it is more efficient to use
|
||||
gcm_send_bulk_message() with a list of registration_ids
|
||||
|
|
@ -72,9 +75,11 @@ def _gcm_send_plain(registration_id, data, **kwargs):
|
|||
|
||||
result = _gcm_send(data, "application/x-www-form-urlencoded;charset=UTF-8")
|
||||
|
||||
# Information about handling response from Google docs (https://developers.google.com/cloud-messaging/http):
|
||||
# Information about handling response from Google docs
|
||||
# (https://developers.google.com/cloud-messaging/http):
|
||||
# If first line starts with id, check second line:
|
||||
# If second line starts with registration_id, gets its value and replace the registration tokens in your
|
||||
# If second line starts with registration_id, gets its
|
||||
# value and replace the registration tokens in your
|
||||
# server database. Otherwise, get the value of Error
|
||||
|
||||
if result.startswith("id"):
|
||||
|
|
@ -111,7 +116,8 @@ def _gcm_send_json(registration_ids, data, **kwargs):
|
|||
if v:
|
||||
values[k] = v
|
||||
|
||||
data = json.dumps(values, separators=(",", ":"), sort_keys=True).encode("utf-8") # keys sorted for tests
|
||||
# Sort the keys for deterministic output (useful for tests)
|
||||
data = json.dumps(values, separators=(",", ":"), sort_keys=True).encode("utf-8")
|
||||
|
||||
response = json.loads(_gcm_send(data, "application/json"))
|
||||
if response["failure"] or response["canonical_ids"]:
|
||||
|
|
@ -120,17 +126,22 @@ def _gcm_send_json(registration_ids, data, **kwargs):
|
|||
for index, result in enumerate(response["results"]):
|
||||
error = result.get("error")
|
||||
if error:
|
||||
# Information from Google docs (https://developers.google.com/cloud-messaging/http)
|
||||
# If error is NotRegistered or InvalidRegistration, then we will deactivate devices because this
|
||||
# registration ID is no more valid and can't be used to send messages, otherwise raise error
|
||||
# Information from Google docs
|
||||
# https://developers.google.com/cloud-messaging/http
|
||||
# If error is NotRegistered or InvalidRegistration,
|
||||
# then we will deactivate devices because this
|
||||
# registration ID is no more valid and can't be used
|
||||
# to send messages, otherwise raise error
|
||||
if error in ("NotRegistered", "InvalidRegistration"):
|
||||
ids_to_remove.append(registration_ids[index])
|
||||
else:
|
||||
throw_error = True
|
||||
|
||||
# If registration_id is set, replace the original ID with the new value (canonical ID) in your
|
||||
# server database. Note that the original ID is not part of the result, so you need to obtain it
|
||||
# from the list of registration_ids passed in the request (using the same index).
|
||||
# If registration_id is set, replace the original ID with
|
||||
# the new value (canonical ID) in your server database.
|
||||
# Note that the original ID is not part of the result, so
|
||||
# you need to obtain it from the list of registration_ids
|
||||
# passed in the request (using the same index).
|
||||
new_id = result.get("registration_id")
|
||||
if new_id:
|
||||
old_new_ids.append((registration_ids[index], new_id))
|
||||
|
|
|
|||
|
|
@ -11,18 +11,23 @@ from .settings import PUSH_NOTIFICATIONS_SETTINGS as SETTINGS
|
|||
@python_2_unicode_compatible
|
||||
class Device(models.Model):
|
||||
name = models.CharField(max_length=255, verbose_name=_("Name"), blank=True, null=True)
|
||||
active = models.BooleanField(verbose_name=_("Is active"), default=True,
|
||||
help_text=_("Inactive devices will not be sent notifications"))
|
||||
active = models.BooleanField(
|
||||
verbose_name=_("Is active"), default=True,
|
||||
help_text=_("Inactive devices will not be sent notifications")
|
||||
)
|
||||
user = models.ForeignKey(SETTINGS["USER_MODEL"], blank=True, null=True)
|
||||
date_created = models.DateTimeField(verbose_name=_("Creation date"), auto_now_add=True, null=True)
|
||||
date_created = models.DateTimeField(
|
||||
verbose_name=_("Creation date"), auto_now_add=True, null=True
|
||||
)
|
||||
|
||||
class Meta:
|
||||
abstract = True
|
||||
|
||||
def __str__(self):
|
||||
return self.name or \
|
||||
str(self.device_id or "") or \
|
||||
"%s for %s" % (self.__class__.__name__, self.user or "unknown user")
|
||||
return (
|
||||
self.name or str(self.device_id or "") or
|
||||
"%s for %s" % (self.__class__.__name__, self.user or "unknown user")
|
||||
)
|
||||
|
||||
|
||||
class GCMDeviceManager(models.Manager):
|
||||
|
|
@ -47,8 +52,10 @@ class GCMDevice(Device):
|
|||
# device_id cannot be a reliable primary key as fragmentation between different devices
|
||||
# can make it turn out to be null and such:
|
||||
# http://android-developers.blogspot.co.uk/2011/03/identifying-app-installations.html
|
||||
device_id = HexIntegerField(verbose_name=_("Device ID"), blank=True, null=True, db_index=True,
|
||||
help_text=_("ANDROID_ID / TelephonyManager.getDeviceId() (always as hex)"))
|
||||
device_id = HexIntegerField(
|
||||
verbose_name=_("Device ID"), blank=True, null=True, db_index=True,
|
||||
help_text=_("ANDROID_ID / TelephonyManager.getDeviceId() (always as hex)")
|
||||
)
|
||||
registration_id = models.TextField(verbose_name=_("Registration ID"))
|
||||
|
||||
objects = GCMDeviceManager()
|
||||
|
|
@ -78,9 +85,13 @@ class APNSDeviceQuerySet(models.query.QuerySet):
|
|||
|
||||
|
||||
class APNSDevice(Device):
|
||||
device_id = models.UUIDField(verbose_name=_("Device ID"), blank=True, null=True, db_index=True,
|
||||
help_text="UDID / UIDevice.identifierForVendor()")
|
||||
registration_id = models.CharField(verbose_name=_("Registration ID"), max_length=200, unique=True)
|
||||
device_id = models.UUIDField(
|
||||
verbose_name=_("Device ID"), blank=True, null=True, db_index=True,
|
||||
help_text="UDID / UIDevice.identifierForVendor()"
|
||||
)
|
||||
registration_id = models.CharField(
|
||||
verbose_name=_("Registration ID"), max_length=200, unique=True
|
||||
)
|
||||
|
||||
objects = APNSDeviceManager()
|
||||
|
||||
|
|
@ -108,8 +119,10 @@ class WNSDeviceQuerySet(models.query.QuerySet):
|
|||
|
||||
|
||||
class WNSDevice(Device):
|
||||
device_id = models.UUIDField(verbose_name=_("Device ID"), blank=True, null=True, db_index=True,
|
||||
help_text=_("GUID()"))
|
||||
device_id = models.UUIDField(
|
||||
verbose_name=_("Device ID"), blank=True, null=True, db_index=True,
|
||||
help_text=_("GUID()")
|
||||
)
|
||||
registration_id = models.TextField(verbose_name=_("Notification URI"))
|
||||
|
||||
objects = WNSDeviceManager()
|
||||
|
|
|
|||
|
|
@ -1,17 +0,0 @@
|
|||
GCM_PLAIN_RESPONSE = 'id=1:08'
|
||||
GCM_JSON_RESPONSE = '{"multicast_id":108,"success":1,"failure":0,"canonical_ids":0,"results":[{"message_id":"1:08"}]}'
|
||||
GCM_MULTIPLE_JSON_RESPONSE = ('{"multicast_id":108,"success":2,"failure":0,"canonical_ids":0,"results":'
|
||||
'[{"message_id":"1:08"}, {"message_id": "1:09"}]}')
|
||||
GCM_PLAIN_RESPONSE_ERROR = ['Error=NotRegistered', 'Error=InvalidRegistration']
|
||||
GCM_PLAIN_RESPONSE_ERROR_B = 'Error=MismatchSenderId'
|
||||
GCM_PLAIN_CANONICAL_ID_RESPONSE = "id=1:2342\nregistration_id=NEW_REGISTRATION_ID"
|
||||
GCM_JSON_RESPONSE_ERROR = ('{"success":1, "failure": 2, "canonical_ids": 0, "cast_id": 6358665107659088804, "results":'
|
||||
' [{"error": "NotRegistered"}, {"message_id": "0:1433830664381654%3449593ff9fd7ecd"}, '
|
||||
'{"error": "InvalidRegistration"}]}')
|
||||
GCM_JSON_RESPONSE_ERROR_B = ('{"success":1, "failure": 2, "canonical_ids": 0, "cast_id": 6358665107659088804, '
|
||||
'"results": [{"error": "MismatchSenderId"}, {"message_id": '
|
||||
'"0:1433830664381654%3449593ff9fd7ecd"}, {"error": "InvalidRegistration"}]}')
|
||||
GCM_DRF_INVALID_HEX_ERROR = {'device_id': [u"Device ID is not a valid hex number"]}
|
||||
GCM_DRF_OUT_OF_RANGE_ERROR = {'device_id': [u"Device ID is out of range"]}
|
||||
GCM_JSON_CANONICAL_ID_RESPONSE = '{"failure":0,"canonical_ids":1,"success":2,"multicast_id":7173139966327257000,"results":[{"registration_id":"NEW_REGISTRATION_ID","message_id":"0:1440068396670935%6868637df9fd7ecd"},{"message_id":"0:1440068396670937%6868637df9fd7ecd"}]}'
|
||||
GCM_JSON_CANONICAL_ID_SAME_DEVICE_RESPONSE = '{"failure":0,"canonical_ids":1,"success":2,"multicast_id":7173139966327257000,"results":[{"registration_id":"bar","message_id":"0:1440068396670935%6868637df9fd7ecd"},{"message_id":"0:1440068396670937%6868637df9fd7ecd"}]}'
|
||||
|
|
@ -5,10 +5,6 @@ import unittest
|
|||
|
||||
|
||||
def setup():
|
||||
"""
|
||||
set up test environment
|
||||
"""
|
||||
|
||||
# add test/src folders to sys path
|
||||
test_folder = os.path.abspath(os.path.dirname(__file__))
|
||||
src_folder = os.path.abspath(os.path.join(test_folder, os.pardir))
|
||||
|
|
@ -34,10 +30,6 @@ def setup():
|
|||
|
||||
|
||||
def tear_down():
|
||||
"""
|
||||
tear down test environment
|
||||
"""
|
||||
|
||||
# destroy test database
|
||||
from django.db import connection
|
||||
connection.creation.destroy_test_db("not_needed")
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
import warnings
|
||||
warnings.simplefilter("ignore", Warning)
|
||||
|
||||
|
||||
DATABASES = {
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
|
|
@ -22,5 +23,4 @@ ROOT_URLCONF = "core.urls"
|
|||
|
||||
SECRET_KEY = "foobar"
|
||||
|
||||
PUSH_NOTIFICATIONS_SETTINGS = {
|
||||
}
|
||||
PUSH_NOTIFICATIONS_SETTINGS = {}
|
||||
|
|
|
|||
|
|
@ -1,45 +1,40 @@
|
|||
import json
|
||||
import mock
|
||||
import os
|
||||
from django.conf import settings
|
||||
from django.core.exceptions import ImproperlyConfigured
|
||||
from django.test import TestCase
|
||||
from django.utils import timezone
|
||||
from push_notifications.models import APNSDevice
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
from push_notifications.apns import *
|
||||
|
||||
class APNSCertfileTestCase(TestCase):
|
||||
def test_apns_send_message_good_certfile(self):
|
||||
path = os.path.join(os.path.dirname(__file__),"test_data","good_revoked.pem")
|
||||
settings.PUSH_NOTIFICATIONS_SETTINGS["APNS_CERTIFICATE"] = path
|
||||
device = APNSDevice.objects.create(
|
||||
registration_id="1212121212121212121212121212121212121212121212121212121212121212",
|
||||
)
|
||||
with mock.patch("ssl.wrap_socket") as ws:
|
||||
with mock.patch("socket.socket") as socket:
|
||||
socket.return_value = 123
|
||||
device.send_message("Hello world")
|
||||
ws.assert_called_once_with(123, ca_certs=None, certfile=path, ssl_version=3)
|
||||
def test_apns_send_message_good_certfile(self):
|
||||
path = os.path.join(os.path.dirname(__file__), "test_data", "good_revoked.pem")
|
||||
settings.PUSH_NOTIFICATIONS_SETTINGS["APNS_CERTIFICATE"] = path
|
||||
device = APNSDevice.objects.create(
|
||||
registration_id="1212121212121212121212121212121212121212121212121212121212121212",
|
||||
)
|
||||
with mock.patch("ssl.wrap_socket") as ws:
|
||||
with mock.patch("socket.socket") as socket:
|
||||
socket.return_value = 123
|
||||
device.send_message("Hello world")
|
||||
ws.assert_called_once_with(123, ca_certs=None, certfile=path, ssl_version=3)
|
||||
|
||||
def test_apns_send_message_raises_no_privatekey(self):
|
||||
path = os.path.join(os.path.dirname(__file__),"test_data","without_private.pem")
|
||||
settings.PUSH_NOTIFICATIONS_SETTINGS["APNS_CERTIFICATE"] = path
|
||||
device = APNSDevice.objects.create(
|
||||
registration_id="1212121212121212121212121212121212121212121212121212121212121212",
|
||||
)
|
||||
with self.assertRaises(ImproperlyConfigured) as ic:
|
||||
device.send_message("Hello world")
|
||||
self.assertTrue(bool(ic.exception.args))
|
||||
self.assertEqual(ic.exception.args[0],"The APNS certificate file at '%s' is unusable: The certificate doesn't contain a private key" % path)
|
||||
def test_apns_send_message_raises_no_privatekey(self):
|
||||
path = os.path.join(os.path.dirname(__file__), "test_data", "without_private.pem")
|
||||
settings.PUSH_NOTIFICATIONS_SETTINGS["APNS_CERTIFICATE"] = path
|
||||
device = APNSDevice.objects.create(
|
||||
registration_id="1212121212121212121212121212121212121212121212121212121212121212",
|
||||
)
|
||||
with self.assertRaises(ImproperlyConfigured) as ic:
|
||||
device.send_message("Hello world")
|
||||
self.assertTrue(bool(ic.exception.args))
|
||||
|
||||
def test_apns_send_message_raises_passwd(self):
|
||||
path = os.path.join(os.path.dirname(__file__),"test_data","good_with_passwd.pem")
|
||||
settings.PUSH_NOTIFICATIONS_SETTINGS["APNS_CERTIFICATE"] = path
|
||||
device = APNSDevice.objects.create(
|
||||
registration_id="1212121212121212121212121212121212121212121212121212121212121212",
|
||||
)
|
||||
with self.assertRaises(ImproperlyConfigured) as ic:
|
||||
device.send_message("Hello world")
|
||||
self.assertTrue(bool(ic.exception.args))
|
||||
self.assertEqual(ic.exception.args[0],"The APNS certificate file at '%s' is unusable: The certificate private key should not be encrypted" % path)
|
||||
def test_apns_send_message_raises_passwd(self):
|
||||
path = os.path.join(os.path.dirname(__file__), "test_data", "good_with_passwd.pem")
|
||||
settings.PUSH_NOTIFICATIONS_SETTINGS["APNS_CERTIFICATE"] = path
|
||||
device = APNSDevice.objects.create(
|
||||
registration_id="1212121212121212121212121212121212121212121212121212121212121212",
|
||||
)
|
||||
with self.assertRaises(ImproperlyConfigured) as ic:
|
||||
device.send_message("Hello world")
|
||||
self.assertTrue(bool(ic.exception.args))
|
||||
|
|
|
|||
|
|
@ -7,21 +7,30 @@ class APNSPushPayloadTest(TestCase):
|
|||
def test_push_payload(self):
|
||||
socket = mock.MagicMock()
|
||||
with mock.patch("push_notifications.apns._apns_pack_frame") as p:
|
||||
_apns_send("123", "Hello world",
|
||||
badge=1, sound="chime", extra={"custom_data": 12345}, expiration=3, socket=socket)
|
||||
p.assert_called_once_with("123",
|
||||
b'{"aps":{"alert":"Hello world","badge":1,"sound":"chime"},"custom_data":12345}', 0, 3, 10)
|
||||
_apns_send(
|
||||
"123", "Hello world", badge=1, sound="chime",
|
||||
extra={"custom_data": 12345}, expiration=3, socket=socket
|
||||
)
|
||||
p.assert_called_once_with(
|
||||
"123",
|
||||
b'{"aps":{"alert":"Hello world","badge":1,"sound":"chime"},"custom_data":12345}',
|
||||
0, 3, 10)
|
||||
|
||||
def test_localised_push_with_empty_body(self):
|
||||
socket = mock.MagicMock()
|
||||
with mock.patch("push_notifications.apns._apns_pack_frame") as p:
|
||||
_apns_send("123", None, loc_key="TEST_LOC_KEY", expiration=3, socket=socket)
|
||||
p.assert_called_once_with("123", b'{"aps":{"alert":{"loc-key":"TEST_LOC_KEY"}}}', 0, 3, 10)
|
||||
p.assert_called_once_with(
|
||||
"123", b'{"aps":{"alert":{"loc-key":"TEST_LOC_KEY"}}}', 0, 3, 10
|
||||
)
|
||||
|
||||
def test_using_extra(self):
|
||||
socket = mock.MagicMock()
|
||||
with mock.patch("push_notifications.apns._apns_pack_frame") as p:
|
||||
_apns_send("123", "sample", extra={"foo": "bar"}, identifier=10, expiration=30, priority=10, socket=socket)
|
||||
_apns_send(
|
||||
"123", "sample", extra={"foo": "bar"}, identifier=10,
|
||||
expiration=30, priority=10, socket=socket
|
||||
)
|
||||
p.assert_called_once_with("123", b'{"aps":{"alert":"sample"},"foo":"bar"}', 10, 30, 10)
|
||||
|
||||
def test_oversized_payload(self):
|
||||
|
|
|
|||
|
|
@ -1,5 +1,4 @@
|
|||
import mock
|
||||
import json
|
||||
from django.test import TestCase
|
||||
from push_notifications.gcm import gcm_send_message, gcm_send_bulk_message
|
||||
from tests.mock_responses import GCM_PLAIN_RESPONSE, GCM_JSON_RESPONSE
|
||||
|
|
@ -15,7 +14,9 @@ class GCMPushPayloadTest(TestCase):
|
|||
|
||||
def test_push_payload_params(self):
|
||||
with mock.patch("push_notifications.gcm._gcm_send", return_value=GCM_PLAIN_RESPONSE) as p:
|
||||
gcm_send_message("abc", {"message": "Hello world"}, delay_while_idle=True, time_to_live=3600)
|
||||
gcm_send_message(
|
||||
"abc", {"message": "Hello world"}, delay_while_idle=True, time_to_live=3600
|
||||
)
|
||||
p.assert_called_once_with(
|
||||
b"data.message=Hello+world&delay_while_idle=1®istration_id=abc&time_to_live=3600",
|
||||
"application/x-www-form-urlencoded;charset=UTF-8")
|
||||
|
|
|
|||
|
|
@ -1,9 +1,6 @@
|
|||
import mock
|
||||
|
||||
from django.core.management import call_command
|
||||
|
||||
from django.test import TestCase
|
||||
from push_notifications.apns import _apns_send, APNSDataOverflow
|
||||
|
||||
|
||||
class CommandsTestCase(TestCase):
|
||||
|
|
@ -14,12 +11,13 @@ class CommandsTestCase(TestCase):
|
|||
device = APNSDevice.objects.create(
|
||||
registration_id="616263", # hex encoding of b'abc'
|
||||
)
|
||||
with mock.patch(
|
||||
'push_notifications.apns._apns_create_socket_to_feedback',
|
||||
mock.MagicMock()):
|
||||
with mock.patch('push_notifications.apns._apns_receive_feedback',
|
||||
mock.MagicMock()) as receiver:
|
||||
|
||||
feedback_method = "push_notifications.apns._apns_create_socket_to_feedback"
|
||||
recv_feedback_method = "push_notifications.apns._apns_receive_feedback"
|
||||
with mock.patch(feedback_method, mock.MagicMock()):
|
||||
with mock.patch(recv_feedback_method, mock.MagicMock()) as receiver:
|
||||
receiver.side_effect = lambda s: [(b'', b'abc')]
|
||||
call_command('prune_devices')
|
||||
|
||||
device = APNSDevice.objects.get(pk=device.pk)
|
||||
self.assertFalse(device.active)
|
||||
|
|
|
|||
|
|
@ -3,251 +3,215 @@ import mock
|
|||
from django.test import TestCase
|
||||
from django.utils import timezone
|
||||
from push_notifications.models import GCMDevice, APNSDevice
|
||||
from tests.mock_responses import ( GCM_PLAIN_RESPONSE,GCM_MULTIPLE_JSON_RESPONSE, GCM_PLAIN_RESPONSE_ERROR,
|
||||
GCM_JSON_RESPONSE_ERROR, GCM_PLAIN_RESPONSE_ERROR_B, GCM_JSON_RESPONSE_ERROR_B,
|
||||
GCM_PLAIN_CANONICAL_ID_RESPONSE, GCM_JSON_CANONICAL_ID_RESPONSE,
|
||||
GCM_JSON_CANONICAL_ID_SAME_DEVICE_RESPONSE)
|
||||
from tests.mock_responses import mock_responses as r
|
||||
from push_notifications.gcm import GCMError, gcm_send_bulk_message
|
||||
|
||||
|
||||
class ModelTestCase(TestCase):
|
||||
def test_can_save_gcm_device(self):
|
||||
device = GCMDevice.objects.create(
|
||||
registration_id="a valid registration id"
|
||||
)
|
||||
assert device.id is not None
|
||||
assert device.date_created is not None
|
||||
assert device.date_created.date() == timezone.now().date()
|
||||
def _create_devices(self, devices):
|
||||
for device in devices:
|
||||
GCMDevice.objects.create(registration_id=device)
|
||||
|
||||
def test_can_create_save_device(self):
|
||||
device = APNSDevice.objects.create(
|
||||
registration_id="a valid registration id"
|
||||
)
|
||||
assert device.id is not None
|
||||
assert device.date_created is not None
|
||||
assert device.date_created.date() == timezone.now().date()
|
||||
def test_can_save_gcm_device(self):
|
||||
device = GCMDevice.objects.create(registration_id="a valid registration id")
|
||||
assert device.id is not None
|
||||
assert device.date_created is not None
|
||||
assert device.date_created.date() == timezone.now().date()
|
||||
|
||||
def test_gcm_send_message(self):
|
||||
device = GCMDevice.objects.create(
|
||||
registration_id="abc",
|
||||
)
|
||||
with mock.patch("push_notifications.gcm._gcm_send", return_value=GCM_PLAIN_RESPONSE) as p:
|
||||
device.send_message("Hello world")
|
||||
p.assert_called_once_with(
|
||||
b"data.message=Hello+world®istration_id=abc",
|
||||
"application/x-www-form-urlencoded;charset=UTF-8")
|
||||
def test_can_create_save_device(self):
|
||||
device = APNSDevice.objects.create(registration_id="a valid registration id")
|
||||
assert device.id is not None
|
||||
assert device.date_created is not None
|
||||
assert device.date_created.date() == timezone.now().date()
|
||||
|
||||
def test_gcm_send_message_extra(self):
|
||||
device = GCMDevice.objects.create(
|
||||
registration_id="abc",
|
||||
)
|
||||
with mock.patch("push_notifications.gcm._gcm_send", return_value=GCM_PLAIN_RESPONSE) as p:
|
||||
device.send_message("Hello world", extra={"foo": "bar"})
|
||||
p.assert_called_once_with(
|
||||
b"data.foo=bar&data.message=Hello+world®istration_id=abc",
|
||||
"application/x-www-form-urlencoded;charset=UTF-8")
|
||||
def test_gcm_send_message(self):
|
||||
device = GCMDevice.objects.create(registration_id="abc")
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_PLAIN_RESPONSE
|
||||
) as p:
|
||||
device.send_message("Hello world")
|
||||
p.assert_called_once_with(
|
||||
b"data.message=Hello+world®istration_id=abc",
|
||||
"application/x-www-form-urlencoded;charset=UTF-8"
|
||||
)
|
||||
|
||||
def test_gcm_send_message_collapse_key(self):
|
||||
device = GCMDevice.objects.create(
|
||||
registration_id="abc",
|
||||
)
|
||||
with mock.patch("push_notifications.gcm._gcm_send", return_value=GCM_PLAIN_RESPONSE) as p:
|
||||
device.send_message("Hello world", collapse_key="test_key")
|
||||
p.assert_called_once_with(
|
||||
b"collapse_key=test_key&data.message=Hello+world®istration_id=abc",
|
||||
"application/x-www-form-urlencoded;charset=UTF-8")
|
||||
def test_gcm_send_message_extra(self):
|
||||
device = GCMDevice.objects.create(registration_id="abc")
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_PLAIN_RESPONSE
|
||||
) as p:
|
||||
device.send_message("Hello world", extra={"foo": "bar"})
|
||||
p.assert_called_once_with(
|
||||
b"data.foo=bar&data.message=Hello+world®istration_id=abc",
|
||||
"application/x-www-form-urlencoded;charset=UTF-8"
|
||||
)
|
||||
|
||||
def test_gcm_send_message_to_multiple_devices(self):
|
||||
GCMDevice.objects.create(
|
||||
registration_id="abc",
|
||||
)
|
||||
def test_gcm_send_message_collapse_key(self):
|
||||
device = GCMDevice.objects.create(registration_id="abc")
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_PLAIN_RESPONSE
|
||||
) as p:
|
||||
device.send_message("Hello world", collapse_key="test_key")
|
||||
p.assert_called_once_with(
|
||||
b"collapse_key=test_key&data.message=Hello+world®istration_id=abc",
|
||||
"application/x-www-form-urlencoded;charset=UTF-8"
|
||||
)
|
||||
|
||||
GCMDevice.objects.create(
|
||||
registration_id="abc1",
|
||||
)
|
||||
def test_gcm_send_message_to_multiple_devices(self):
|
||||
self._create_devices(["abc", "abc1"])
|
||||
|
||||
with mock.patch("push_notifications.gcm._gcm_send", return_value=GCM_MULTIPLE_JSON_RESPONSE) as p:
|
||||
GCMDevice.objects.all().send_message("Hello world")
|
||||
p.assert_called_once_with(
|
||||
json.dumps({
|
||||
"data": { "message": "Hello world" },
|
||||
"registration_ids": ["abc", "abc1"]
|
||||
}, separators=(",", ":"), sort_keys=True).encode("utf-8"), "application/json")
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_MULTIPLE_JSON_RESPONSE
|
||||
) as p:
|
||||
GCMDevice.objects.all().send_message("Hello world")
|
||||
p.assert_called_once_with(
|
||||
json.dumps({
|
||||
"data": {"message": "Hello world"},
|
||||
"registration_ids": ["abc", "abc1"]
|
||||
}, separators=(",", ":"), sort_keys=True).encode("utf-8"), "application/json")
|
||||
|
||||
def test_gcm_send_message_active_devices(self):
|
||||
GCMDevice.objects.create(
|
||||
registration_id="abc",
|
||||
active=True
|
||||
)
|
||||
def test_gcm_send_message_active_devices(self):
|
||||
GCMDevice.objects.create(registration_id="abc", active=True)
|
||||
GCMDevice.objects.create(registration_id="xyz", active=False)
|
||||
|
||||
GCMDevice.objects.create(
|
||||
registration_id="xyz",
|
||||
active=False
|
||||
)
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_MULTIPLE_JSON_RESPONSE
|
||||
) as p:
|
||||
GCMDevice.objects.all().send_message("Hello world")
|
||||
p.assert_called_once_with(
|
||||
json.dumps({
|
||||
"data": {"message": "Hello world"},
|
||||
"registration_ids": ["abc"]
|
||||
}, separators=(",", ":"), sort_keys=True).encode("utf-8"), "application/json")
|
||||
|
||||
with mock.patch("push_notifications.gcm._gcm_send", return_value=GCM_MULTIPLE_JSON_RESPONSE) as p:
|
||||
GCMDevice.objects.all().send_message("Hello world")
|
||||
p.assert_called_once_with(
|
||||
json.dumps({
|
||||
"data": { "message": "Hello world" },
|
||||
"registration_ids": ["abc"]
|
||||
}, separators=(",", ":"), sort_keys=True).encode("utf-8"), "application/json")
|
||||
def test_gcm_send_message_collapse_to_multiple_devices(self):
|
||||
self._create_devices(["abc", "abc1"])
|
||||
|
||||
def test_gcm_send_message_extra_to_multiple_devices(self):
|
||||
GCMDevice.objects.create(
|
||||
registration_id="abc",
|
||||
)
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_MULTIPLE_JSON_RESPONSE
|
||||
) as p:
|
||||
GCMDevice.objects.all().send_message("Hello world", collapse_key="test_key")
|
||||
p.assert_called_once_with(
|
||||
json.dumps({
|
||||
"collapse_key": "test_key",
|
||||
"data": {"message": "Hello world"},
|
||||
"registration_ids": ["abc", "abc1"]
|
||||
}, separators=(",", ":"), sort_keys=True).encode("utf-8"), "application/json")
|
||||
|
||||
GCMDevice.objects.create(
|
||||
registration_id="abc1",
|
||||
)
|
||||
def test_gcm_send_message_to_single_device_with_error(self):
|
||||
# these errors are device specific, device.active will be set false
|
||||
devices = ["abc", "abc1"]
|
||||
self._create_devices(devices)
|
||||
|
||||
with mock.patch("push_notifications.gcm._gcm_send", return_value=GCM_MULTIPLE_JSON_RESPONSE) as p:
|
||||
GCMDevice.objects.all().send_message("Hello world", extra={"foo": "bar"})
|
||||
p.assert_called_once_with(
|
||||
json.dumps({
|
||||
"data": { "foo": "bar", "message": "Hello world" },
|
||||
"registration_ids": ["abc", "abc1"]
|
||||
}, separators=(",", ":"), sort_keys=True).encode("utf-8"), "application/json")
|
||||
for index, error in enumerate(r.GCM_PLAIN_RESPONSE_ERROR):
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=error):
|
||||
device = GCMDevice.objects.get(registration_id=devices[index])
|
||||
device.send_message("Hello World!")
|
||||
assert GCMDevice.objects.get(registration_id=devices[index]).active is False
|
||||
|
||||
def test_gcm_send_message_collapse_to_multiple_devices(self):
|
||||
GCMDevice.objects.create(
|
||||
registration_id="abc",
|
||||
)
|
||||
def test_gcm_send_message_to_single_device_with_error_b(self):
|
||||
device = GCMDevice.objects.create(registration_id="abc")
|
||||
|
||||
GCMDevice.objects.create(
|
||||
registration_id="abc1",
|
||||
)
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_PLAIN_RESPONSE_ERROR_B
|
||||
):
|
||||
# these errors are not device specific, GCMError should be thrown
|
||||
with self.assertRaises(GCMError):
|
||||
device.send_message("Hello World!")
|
||||
assert GCMDevice.objects.get(registration_id="abc").active is True
|
||||
|
||||
with mock.patch("push_notifications.gcm._gcm_send", return_value=GCM_MULTIPLE_JSON_RESPONSE) as p:
|
||||
GCMDevice.objects.all().send_message("Hello world", collapse_key="test_key")
|
||||
p.assert_called_once_with(
|
||||
json.dumps({
|
||||
"collapse_key": "test_key",
|
||||
"data": { "message": "Hello world" },
|
||||
"registration_ids": ["abc", "abc1"]
|
||||
}, separators=(",", ":"), sort_keys=True).encode("utf-8"), "application/json")
|
||||
def test_gcm_send_message_to_multiple_devices_with_error(self):
|
||||
self._create_devices(["abc", "abc1", "abc2"])
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_JSON_RESPONSE_ERROR
|
||||
):
|
||||
devices = GCMDevice.objects.all()
|
||||
devices.send_message("Hello World")
|
||||
assert not GCMDevice.objects.get(registration_id="abc").active
|
||||
assert GCMDevice.objects.get(registration_id="abc1").active
|
||||
assert not GCMDevice.objects.get(registration_id="abc2").active
|
||||
|
||||
def test_gcm_send_message_to_single_device_with_error(self):
|
||||
# these errors are device specific, device.active will be set false
|
||||
device_list = ['abc', 'abc1']
|
||||
self.create_devices(device_list)
|
||||
for index, error in enumerate(GCM_PLAIN_RESPONSE_ERROR):
|
||||
with mock.patch("push_notifications.gcm._gcm_send",
|
||||
return_value=error) as p:
|
||||
device = GCMDevice.objects. \
|
||||
get(registration_id=device_list[index])
|
||||
device.send_message("Hello World!")
|
||||
assert GCMDevice.objects.get(registration_id=device_list[index]).active is False
|
||||
def test_gcm_send_message_to_multiple_devices_with_error_b(self):
|
||||
self._create_devices(["abc", "abc1", "abc2"])
|
||||
|
||||
def test_gcm_send_message_to_single_device_with_error_b(self):
|
||||
# these errors are not device specific, GCMError should be thrown
|
||||
device_list = ['abc']
|
||||
self.create_devices(device_list)
|
||||
with mock.patch("push_notifications.gcm._gcm_send",
|
||||
return_value=GCM_PLAIN_RESPONSE_ERROR_B) as p:
|
||||
device = GCMDevice.objects. \
|
||||
get(registration_id=device_list[0])
|
||||
with self.assertRaises(GCMError):
|
||||
device.send_message("Hello World!")
|
||||
assert GCMDevice.objects.get(registration_id=device_list[0]).active is True
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_JSON_RESPONSE_ERROR_B
|
||||
):
|
||||
devices = GCMDevice.objects.all()
|
||||
with self.assertRaises(GCMError):
|
||||
devices.send_message("Hello World")
|
||||
assert GCMDevice.objects.get(registration_id="abc").active is True
|
||||
assert GCMDevice.objects.get(registration_id="abc1").active is True
|
||||
assert GCMDevice.objects.get(registration_id="abc2").active is False
|
||||
|
||||
def test_gcm_send_message_to_multiple_devices_with_error(self):
|
||||
device_list = ['abc', 'abc1', 'abc2']
|
||||
self.create_devices(device_list)
|
||||
with mock.patch("push_notifications.gcm._gcm_send",
|
||||
return_value=GCM_JSON_RESPONSE_ERROR) as p:
|
||||
devices = GCMDevice.objects.all()
|
||||
devices.send_message("Hello World")
|
||||
assert GCMDevice.objects.get(registration_id=device_list[0]).active is False
|
||||
assert GCMDevice.objects.get(registration_id=device_list[1]).active is True
|
||||
assert GCMDevice.objects.get(registration_id=device_list[2]).active is False
|
||||
def test_gcm_send_message_to_multiple_devices_with_canonical_id(self):
|
||||
self._create_devices(["foo", "bar"])
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_JSON_CANONICAL_ID_RESPONSE
|
||||
):
|
||||
GCMDevice.objects.all().send_message("Hello World")
|
||||
assert not GCMDevice.objects.filter(registration_id="foo").exists()
|
||||
assert GCMDevice.objects.filter(registration_id="bar").exists()
|
||||
assert GCMDevice.objects.filter(registration_id="NEW_REGISTRATION_ID").exists() is True
|
||||
|
||||
def test_gcm_send_message_to_multiple_devices_with_error_b(self):
|
||||
device_list = ['abc', 'abc1', 'abc2']
|
||||
self.create_devices(device_list)
|
||||
with mock.patch("push_notifications.gcm._gcm_send",
|
||||
return_value=GCM_JSON_RESPONSE_ERROR_B) as p:
|
||||
devices = GCMDevice.objects.all()
|
||||
with self.assertRaises(GCMError):
|
||||
devices.send_message("Hello World")
|
||||
assert GCMDevice.objects.get(registration_id=device_list[0]).active is True
|
||||
assert GCMDevice.objects.get(registration_id=device_list[1]).active is True
|
||||
assert GCMDevice.objects.get(registration_id=device_list[2]).active is False
|
||||
def test_gcm_send_message_to_single_user_with_canonical_id(self):
|
||||
old_registration_id = "foo"
|
||||
self._create_devices([old_registration_id])
|
||||
|
||||
def test_gcm_send_message_to_multiple_devices_with_canonical_id(self):
|
||||
device_list = ['foo', 'bar']
|
||||
self.create_devices(device_list)
|
||||
with mock.patch("push_notifications.gcm._gcm_send",
|
||||
return_value=GCM_JSON_CANONICAL_ID_RESPONSE):
|
||||
GCMDevice.objects.all().send_message("Hello World")
|
||||
assert GCMDevice.objects.filter(registration_id=device_list[0]).exists() is False
|
||||
assert GCMDevice.objects.filter(registration_id=device_list[1]).exists() is True
|
||||
assert GCMDevice.objects.filter(registration_id="NEW_REGISTRATION_ID").exists() is True
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send", return_value=r.GCM_PLAIN_CANONICAL_ID_RESPONSE
|
||||
):
|
||||
GCMDevice.objects.get(registration_id=old_registration_id).send_message("Hello World")
|
||||
assert not GCMDevice.objects.filter(registration_id=old_registration_id).exists()
|
||||
assert GCMDevice.objects.filter(registration_id="NEW_REGISTRATION_ID").exists()
|
||||
|
||||
def test_gcm_send_message_to_single_user_with_canonical_id(self):
|
||||
old_registration_id = 'foo'
|
||||
self.create_devices([old_registration_id])
|
||||
with mock.patch("push_notifications.gcm._gcm_send",
|
||||
return_value=GCM_PLAIN_CANONICAL_ID_RESPONSE):
|
||||
GCMDevice.objects.get(registration_id=old_registration_id).send_message("Hello World")
|
||||
assert GCMDevice.objects.filter(registration_id=old_registration_id).exists() is False
|
||||
assert GCMDevice.objects.filter(registration_id="NEW_REGISTRATION_ID").exists() is True
|
||||
def test_gcm_send_message_to_same_devices_with_canonical_id(self):
|
||||
first_device = GCMDevice.objects.create(registration_id="foo", active=True)
|
||||
second_device = GCMDevice.objects.create(registration_id="bar", active=False)
|
||||
|
||||
def test_gcm_send_message_to_same_devices_with_canonical_id(self):
|
||||
device_list = ['foo', 'bar']
|
||||
self.create_devices(device_list)
|
||||
first_device_pk = GCMDevice.objects.get(registration_id='foo').pk
|
||||
second_device_pk = GCMDevice.objects.get(registration_id='bar').pk
|
||||
with mock.patch("push_notifications.gcm._gcm_send",
|
||||
return_value=GCM_JSON_CANONICAL_ID_SAME_DEVICE_RESPONSE):
|
||||
GCMDevice.objects.all().send_message("Hello World")
|
||||
first_device = GCMDevice.objects.get(pk=first_device_pk)
|
||||
second_device = GCMDevice.objects.get(pk=second_device_pk)
|
||||
assert first_device.active is False
|
||||
assert second_device.active is True
|
||||
with mock.patch(
|
||||
"push_notifications.gcm._gcm_send",
|
||||
return_value=r.GCM_JSON_CANONICAL_ID_SAME_DEVICE_RESPONSE
|
||||
):
|
||||
GCMDevice.objects.all().send_message("Hello World")
|
||||
|
||||
def test_apns_send_message(self):
|
||||
device = APNSDevice.objects.create(
|
||||
registration_id="abc",
|
||||
)
|
||||
socket = mock.MagicMock()
|
||||
with mock.patch("push_notifications.apns._apns_pack_frame") as p:
|
||||
device.send_message("Hello world", socket=socket, expiration=1)
|
||||
p.assert_called_once_with("abc", b'{"aps":{"alert":"Hello world"}}', 0, 1, 10)
|
||||
assert first_device.active is True
|
||||
assert second_device.active is False
|
||||
|
||||
def test_apns_send_message_extra(self):
|
||||
device = APNSDevice.objects.create(
|
||||
registration_id="abc",
|
||||
)
|
||||
socket = mock.MagicMock()
|
||||
with mock.patch("push_notifications.apns._apns_pack_frame") as p:
|
||||
device.send_message("Hello world", extra={"foo": "bar"}, socket=socket, identifier=1, expiration=2, priority=5)
|
||||
p.assert_called_once_with("abc", b'{"aps":{"alert":"Hello world"},"foo":"bar"}', 1, 2, 5)
|
||||
def test_apns_send_message(self):
|
||||
device = APNSDevice.objects.create(registration_id="abc")
|
||||
socket = mock.MagicMock()
|
||||
|
||||
def test_send_message_with_no_reg_ids(self):
|
||||
device_list = ['abc', 'abc1']
|
||||
self.create_devices(device_list)
|
||||
with mock.patch("push_notifications.apns._apns_pack_frame") as p:
|
||||
device.send_message("Hello world", socket=socket, expiration=1)
|
||||
p.assert_called_once_with("abc", b'{"aps":{"alert":"Hello world"}}', 0, 1, 10)
|
||||
|
||||
with mock.patch("push_notifications.gcm._gcm_send_plain", return_value='') as p:
|
||||
GCMDevice.objects.filter(registration_id='xyz').send_message('Hello World')
|
||||
p.assert_not_called()
|
||||
def test_apns_send_message_extra(self):
|
||||
device = APNSDevice.objects.create(registration_id="abc")
|
||||
socket = mock.MagicMock()
|
||||
|
||||
with mock.patch("push_notifications.gcm._gcm_send_json", return_value='') as p:
|
||||
reg_ids = [obj.registration_id for obj in GCMDevice.objects.all()]
|
||||
gcm_send_bulk_message(reg_ids, {"message": "Hello World"})
|
||||
p.assert_called_once_with([u"abc", u"abc1"], {"message": "Hello World"})
|
||||
with mock.patch("push_notifications.apns._apns_pack_frame") as p:
|
||||
device.send_message(
|
||||
"Hello world", extra={"foo": "bar"}, socket=socket,
|
||||
identifier=1, expiration=2, priority=5
|
||||
)
|
||||
p.assert_called_once_with("abc", b'{"aps":{"alert":"Hello world"},"foo":"bar"}', 1, 2, 5)
|
||||
|
||||
def test_can_save_wsn_device(self):
|
||||
device = GCMDevice.objects.create(
|
||||
registration_id="a valid registration id"
|
||||
)
|
||||
self.assertIsNotNone(device.pk)
|
||||
self.assertIsNotNone(device.date_created)
|
||||
self.assertEquals(device.date_created.date(), timezone.now().date())
|
||||
def test_send_message_with_no_reg_ids(self):
|
||||
self._create_devices(["abc", "abc1"])
|
||||
|
||||
def create_devices(self, devices):
|
||||
for device in devices:
|
||||
GCMDevice.objects.create(
|
||||
registration_id=device,
|
||||
)
|
||||
with mock.patch("push_notifications.gcm._gcm_send_plain", return_value="") as p:
|
||||
GCMDevice.objects.filter(registration_id="xyz").send_message("Hello World")
|
||||
p.assert_not_called()
|
||||
|
||||
with mock.patch("push_notifications.gcm._gcm_send_json", return_value="") as p:
|
||||
reg_ids = [obj.registration_id for obj in GCMDevice.objects.all()]
|
||||
gcm_send_bulk_message(reg_ids, {"message": "Hello World"})
|
||||
p.assert_called_once_with([u"abc", u"abc1"], {"message": "Hello World"})
|
||||
|
||||
def test_can_save_wsn_device(self):
|
||||
device = GCMDevice.objects.create(registration_id="a valid registration id")
|
||||
self.assertIsNotNone(device.pk)
|
||||
self.assertIsNotNone(device.date_created)
|
||||
self.assertEqual(device.date_created.date(), timezone.now().date())
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ class APNSDeviceSerializerTestCase(TestCase):
|
|||
|
||||
# valid data - 100 bytes upper case
|
||||
serializer = APNSDeviceSerializer(data={
|
||||
"registration_id": "AEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAE",
|
||||
"registration_id": "AE" * 100,
|
||||
"name": "Apple iPhone 6+",
|
||||
"device_id": "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF",
|
||||
})
|
||||
|
|
@ -32,7 +32,7 @@ class APNSDeviceSerializerTestCase(TestCase):
|
|||
|
||||
# valid data - 100 bytes lower case
|
||||
serializer = APNSDeviceSerializer(data={
|
||||
"registration_id": "aeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeae",
|
||||
"registration_id": "ae" * 100,
|
||||
"name": "Apple iPhone 6+",
|
||||
"device_id": "ffffffffffffffffffffffffffffffff",
|
||||
})
|
||||
|
|
@ -45,8 +45,6 @@ class APNSDeviceSerializerTestCase(TestCase):
|
|||
"device_id": "ffffffffffffffffffffffffffffake",
|
||||
})
|
||||
self.assertFalse(serializer.is_valid())
|
||||
self.assertEqual(serializer.errors["device_id"][0], '"ffffffffffffffffffffffffffffake" is not a valid UUID.')
|
||||
self.assertEqual(serializer.errors["registration_id"][0], "Registration ID (device token) is invalid")
|
||||
|
||||
|
||||
class GCMDeviceSerializerTestCase(TestCase):
|
||||
|
|
@ -86,9 +84,8 @@ class GCMDeviceSerializerTestCase(TestCase):
|
|||
"device_id": "0xdeadbeaf",
|
||||
})
|
||||
|
||||
with self.assertRaises(ValidationError) as ex:
|
||||
with self.assertRaises(ValidationError):
|
||||
serializer.is_valid(raise_exception=True)
|
||||
self.assertEqual({'registration_id': [u'This field must be unique.']}, ex.exception.detail)
|
||||
|
||||
def test_device_id_validation_fail_bad_hex(self):
|
||||
serializer = GCMDeviceSerializer(data={
|
||||
|
|
@ -103,7 +100,7 @@ class GCMDeviceSerializerTestCase(TestCase):
|
|||
serializer = GCMDeviceSerializer(data={
|
||||
"registration_id": "foobar",
|
||||
"name": "Galaxy Note 3",
|
||||
"device_id": "10000000000000000", # 2**64
|
||||
"device_id": "10000000000000000", # 2**64
|
||||
})
|
||||
self.assertFalse(serializer.is_valid())
|
||||
self.assertEqual(serializer.errors, GCM_DRF_OUT_OF_RANGE_ERROR)
|
||||
|
|
|
|||
Loading…
Reference in New Issue