Import python-django-push-notifications_1.4.1.orig.tar.gz

[dgit import orig python-django-push-notifications_1.4.1.orig.tar.gz]
This commit is contained in:
Michael Fladischer 2017-03-03 08:37:39 +01:00
commit 7c850180f5
37 changed files with 2077 additions and 0 deletions

17
.gitignore vendored Normal file
View File

@ -0,0 +1,17 @@
# python compiled
__pycache__
*.pyc
# distutils
MANIFEST
build
# IDE
.idea
*.iml
# virtualenv
.env
# tox
.tox

12
.travis.yml Normal file
View File

@ -0,0 +1,12 @@
language: python
addons:
apt:
sources:
- deadsnakes
packages:
- python3.5
install:
- pip install tox
script:
- tox
sudo: false

44
AUTHORS Normal file
View File

@ -0,0 +1,44 @@
This library was created by Jerome Leclanche <jerome@leclan.ch>, for use on the
Anthill application (https://www.anthill.com).
Special thanks to everyone who contributed:
Adam "Cezar" Jenkins
Alan Descoins
Ales Dokhanin
Alistair Broomhead
Andrey Zevakin
Antonin Lenfant
Arthur Silva
Avichal Pandey
Brad Pitcher
Daniel Kronovet
David Pretty
Dilvane Zanardine
Florian Finke
Florian Purchess
Francois Lebel
halak
Innocenty Enikeew
Jack Feng
Jamaal Scarlett
Jay Camp
Jeremy Morgan
Jerome Leclanche
Julien Dubiel
Lital Natan
Luke Burden
Marconi Moreto
Matthew Hershberger
Maxim Kamenkov
Mohamad Nour Chawich
Nicolas Delaby
Remigiusz Dymecki
Ruslan Kovtun
Sander Heling
Sergei Evdokimov
Sujit Nair
Thomas Iovine
Valentin Hăloiu
Wyan Jow
@hoongun

101
CHANGELOG.rst Normal file
View File

@ -0,0 +1,101 @@
v1.4.1 (2016-01-11)
===================
* APNS: Increased max device token size to 100 bytes (WWDC 2015, iOS 9)
* BUGFIX: Fix an index error in the admin
v1.4.0 (2015-12-13)
===================
* BACKWARDS-INCOMPATIBLE: Drop support for Python<3.4
* DJANGO: Support Django 1.9
* GCM: Handle canonical IDs
* GCM: Allow full range of GCMDevice.device_id values
* GCM: Do not allow duplicate registration_ids
* DRF: Work around empty boolean defaults issue (django-rest-framework#1101)
* BUGFIX: Do not throw GCMError in bulk messages from the admin
* BUGFIX: Avoid generating an extra migration on Python 3
* BUGFIX: Only send in bulk to active devices
* BUGFIX: Display models correctly in the admin on both Python 2 and 3
v1.3.1 (2015-06-30)
===================
This is an errata release.
v1.3.0 (2015-06-30)
===================
* BACKWARDS-INCOMPATIBLE: Drop support for Python<2.7
* BACKWARDS-INCOMPATIBLE: Drop support for Django<1.8
* NEW FEATURE: Added a Django Rest Framework API. Requires DRF>=3.0.
* APNS: Add support for setting the ca_certs file with new APNS_CA_CERTIFICATES setting
* GCM: Deactivate GCMDevices when their notifications cause NotRegistered or InvalidRegistration
* GCM: Indiscriminately handle all keyword arguments in gcm_send_message and gcm_send_bulk_message
* GCM: Never fall back to json in gcm_send_message
* BUGFIX: Fixed migration issues from 1.2.0 upgrade.
* BUGFIX: Better detection of SQLite/GIS MySQL in various checks
* BUGFIX: Assorted Python 3 bugfixes
* BUGFIX: Fix display of device_id in admin
v1.2.1 (2015-04-11)
===================
* APNS, GCM: Add a db_index to the device_id field
* APNS: Use the native UUIDField on Django 1.8
* APNS: Fix timeout handling on Python 3
* APNS: Restore error checking on apns_send_bulk_message
* GCM: Expose the time_to_live argument in gcm_send_bulk_message
* GCM: Fix return value when gcm bulk is split in batches
* GCM: Improved error checking reliability
* GCM: Properly pass kwargs in GCMDeviceQuerySet.send_message()
* BUGFIX: Fix HexIntegerField for Django 1.3
v1.2.0 (2014-10-07)
===================
* BACKWARDS-INCOMPATIBLE: Added support for Django 1.7 migrations. South users will have to upgrade to South 1.0 or Django 1.7.
* APNS: APNS MAX_NOTIFICATION_SIZE is now a setting and its default has been increased to 2048
* APNS: Always connect with TLSv1 instead of SSLv3
* APNS: Implemented support for APNS Feedback Service
* APNS: Support for optional "category" dict
* GCM: Improved error handling in bulk mode
* GCM: Added support for time_to_live parameter
* BUGFIX: Fixed various issues relating HexIntegerField
* BUGFIX: Fixed issues in the admin with custom user models
v1.1.0 (2014-06-29)
===================
* BACKWARDS-INCOMPATIBLE: The arguments for device.send_message() have changed. See README.rst for details.
* Added a date_created field to GCMDevice and APNSDevice. This field keeps track of when the Device was created.
This requires a `manage.py migrate`.
* Updated APNS protocol support
* Allow sending empty sounds on APNS
* Several APNS bugfixes
* Fixed BigIntegerField support on PostGIS
* Assorted migrations bugfixes
* Added a test suite
v1.0.1 (2013-01-16)
===================
* Migrations have been reset. If you were using migrations pre-1.0 you should upgrade to 1.0 instead and only
upgrade to 1.0.1 when you are ready to reset your migrations.
v1.0 (2013-01-15)
=================
* Full Python 3 support
* GCM device_id is now a custom field based on BigIntegerField and always unsigned (it should be input as hex)
* Django versions older than 1.5 now require 'six' to be installed
* Drop uniqueness on gcm registration_id due to compatibility issues with MySQL
* Fix some issues with migrations
* Add some basic tests
* Integrate with travis-ci
* Add an AUTHORS file
v0.9 (2013-12-17)
=================
* Enable installation with pip
* Add wheel support
* Add full documentation
* Various bug fixes
v0.8 (2013-03-15)
=================
* Initial release

24
CONTRIBUTING.md Normal file
View File

@ -0,0 +1,24 @@
### Coding style
Please adhere to the coding style throughout the project.
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.
3. When in doubt, [PEP8](https://www.python.org/dev/peps/pep-0008/). Follow its naming conventions.
4. Know when to make exceptions.
Also see: [How to name things in programming](http://www.slideshare.net/pirhilton/how-to-name-things-the-hardest-problem-in-programming)
### Commits and Pull Requests
Keep the commit log as healthy as the code. It is one of the first places new contributors will look at the project.
1. No more than one change per commit. There should be no changes in a commit which are unrelated to its message.
2. Every commit should pass all tests on its own.
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.
Also see: [Github Help: Using Pull Requests](https://help.github.com/articles/using-pull-requests/)

19
LICENSE Normal file
View File

@ -0,0 +1,19 @@
Copyright (c) Jerome Leclanche <jerome@leclan.ch>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.

3
MANIFEST.in Normal file
View File

@ -0,0 +1,3 @@
include MANIFEST.in
include README.rst
include LICENSE

219
README.rst Normal file
View File

@ -0,0 +1,219 @@
django-push-notifications
=========================
.. image:: https://api.travis-ci.org/jleclanche/django-push-notifications.png
:target: https://travis-ci.org/jleclanche/django-push-notifications
A minimal Django app that implements Device models that can send messages through APNS and GCM.
The app implements two models: ``GCMDevice`` and ``APNSDevice``. Those models share the same attributes:
- ``name`` (optional): A name for the device.
- ``active`` (default True): A boolean that determines whether the device will be sent notifications.
- ``user`` (optional): A foreign key to auth.User, if you wish to link the device to a specific user.
- ``device_id`` (optional): A UUID for the device obtained from Android/iOS APIs, if you wish to uniquely identify it.
- ``registration_id`` (required): The GCM registration id or the APNS token for the device.
The app also implements an admin panel, through which you can test single and bulk notifications. Select one or more
GCM or APNS devices and in the action dropdown, select "Send test message" or "Send test message in bulk", accordingly.
Note that sending a non-bulk test message to more than one device will just iterate over the devices and send multiple
single messages.
Dependencies
------------
Django 1.8 is required. Support for older versions is available in the release 1.2.1.
Tastypie support should work on Tastypie 0.11.0 and newer.
Django REST Framework support should work on DRF version 3.0 and newer.
Setup
-----
You can install the library directly from pypi using pip:
.. code-block:: shell
$ pip install django-push-notifications
Edit your settings.py file:
.. code-block:: python
INSTALLED_APPS = (
...
"push_notifications"
)
PUSH_NOTIFICATIONS_SETTINGS = {
"GCM_API_KEY": "<your api key>",
"APNS_CERTIFICATE": "/path/to/your/certificate.pem",
}
.. note::
If you are planning on running your project with ``DEBUG=True``, then make sure you have set the
*development* certificate as your ``APNS_CERTIFICATE``. Otherwise the app will not be able to connect to the correct host. See settings_ for details.
You can learn more about APNS certificates `here <https://developer.apple.com/library/ios/documentation/NetworkingInternet/Conceptual/RemoteNotificationsPG/Chapters/ProvisioningDevelopment.html>`_.
Native Django migrations are in use. ``manage.py migrate`` will install and migrate all models.
.. _settings:
Settings list
-------------
All settings are contained in a ``PUSH_NOTIFICATIONS_SETTINGS`` dict.
In order to use GCM, you are required to include ``GCM_API_KEY``.
For APNS, you are required to include ``APNS_CERTIFICATE``.
- ``APNS_CERTIFICATE``: Absolute path to your APNS certificate file. Certificates with passphrases are not supported.
- ``APNS_CA_CERTIFICATES``: Absolute path to a CA certificates file for APNS. Optional - do not set if not needed. Defaults to None.
- ``GCM_API_KEY``: Your API key for GCM.
- ``APNS_HOST``: The hostname used for the APNS sockets.
- When ``DEBUG=True``, this defaults to ``gateway.sandbox.push.apple.com``.
- When ``DEBUG=False``, this defaults to ``gateway.push.apple.com``.
- ``APNS_PORT``: The port used along with APNS_HOST. Defaults to 2195.
- ``GCM_POST_URL``: The full url that GCM notifications will be POSTed to. Defaults to https://android.googleapis.com/gcm/send.
- ``GCM_MAX_RECIPIENTS``: The maximum amount of recipients that can be contained per bulk message. If the ``registration_ids`` list is larger than that number, multiple bulk messages will be sent. Defaults to 1000 (the maximum amount supported by GCM).
Sending messages
----------------
GCM and APNS services have slightly different semantics. The app tries to offer a common interface for both when using the models.
.. code-block:: python
from push_notifications.models import APNSDevice, GCMDevice
device = GCMDevice.objects.get(registration_id=gcm_reg_id)
# The first argument will be sent as "message" to the intent extras Bundle
# Retrieve it with intent.getExtras().getString("message")
device.send_message("You've got mail")
# If you want to customize, send an extra dict and a None message.
# the extras dict will be mapped into the intent extras Bundle.
# For dicts where all values are keys this will be sent as url parameters,
# but for more complex nested collections the extras dict will be sent via
# the bulk message api.
device.send_message(None, extra={"foo": "bar"})
device = APNSDevice.objects.get(registration_id=apns_token)
device.send_message("You've got mail") # Alert message may only be sent as text.
device.send_message(None, badge=5) # No alerts but with badge.
device.send_message(None, badge=1, extra={"foo": "bar"}) # Silent message with badge and added custom data.
.. note::
APNS does not support sending payloads that exceed 2048 bytes (increased from 256 in 2014).
The message is only one part of the payload, if
once constructed the payload exceeds the maximum size, an ``APNSDataOverflow`` exception will be raised before anything is sent.
Sending messages in bulk
------------------------
.. code-block:: python
from push_notifications.models import APNSDevice, GCMDevice
devices = GCMDevice.objects.filter(user__first_name="James")
devices.send_message("Happy name day!")
Sending messages in bulk makes use of the bulk mechanics offered by GCM and APNS. It is almost always preferable to send
bulk notifications instead of single ones.
Administration
--------------
APNS devices which are not receiving push notifications can be set to inactive by two methods. The web admin interface for
APNS devices has a "prune devices" option. Any selected devices which are not receiving notifications will be set to inactive [1]_.
There is also a management command to prune all devices failing to receive notifications:
.. code-block:: shell
$ python manage.py prune_devices
This removes all devices which are not receiving notifications.
For more information, please refer to the APNS feedback service_.
.. _service: https://developer.apple.com/library/ios/documentation/NetworkingInternet/Conceptual/RemoteNotificationsPG/Chapters/CommunicatingWIthAPS.html
Exceptions
----------
- ``NotificationError(Exception)``: Base exception for all notification-related errors.
- ``gcm.GCMError(NotificationError)``: An error was returned by GCM. This is never raised when using bulk notifications.
- ``apns.APNSError(NotificationError)``: Something went wrong upon sending APNS notifications.
- ``apns.APNSDataOverflow(APNSError)``: The APNS payload exceeds its maximum size and cannot be sent.
Tastypie support
----------------
The app includes tastypie-compatible resources in push_notifications.api.tastypie. These can be used as-is, or as base classes
for more involved APIs.
The following resources are available:
- ``APNSDeviceResource``
- ``GCMDeviceResource``
- ``APNSDeviceAuthenticatedResource``
- ``GCMDeviceAuthenticatedResource``
The base device resources will not ask for authentication, while the authenticated ones will link the logged in user to
the device they register.
Subclassing the authenticated resources in order to add a ``SameUserAuthentication`` and a user ``ForeignKey`` is recommended.
When registered, the APIs will show up at ``<api_root>/device/apns`` and ``<api_root>/device/gcm``, respectively.
Django REST Framework (DRF) support
-----------------------------------
ViewSets are available for both APNS and GCM devices in two permission flavors:
- ``APNSDeviceViewSet`` and ``GCMDeviceViewSet``
- Permissions as specified in settings (``AllowAny`` by default, which is not recommended)
- A device may be registered without associating it with a user
- ``APNSDeviceAuthorizedViewSet`` and ``GCMDeviceAuthorizedViewSet``
- Permissions are ``IsAuthenticated`` and custom permission ``IsOwner``, which will only allow the ``request.user`` to get and update devices that belong to that user
- Requires a user to be authenticated, so all devices will be associated with a user
When creating an ``APNSDevice``, the ``registration_id`` is validated to be a 64-character or 200-character hexadecimal string. Since 2016, device tokens are to be increased from 32 bytes to 100 bytes.
Routes can be added one of two ways:
- Routers_ (include all views)
.. _Routers: http://www.django-rest-framework.org/tutorial/6-viewsets-and-routers#using-routers
::
from push_notifications.api.rest_framework import APNSDeviceAuthorizedViewSet, GCMDeviceAuthorizedViewSet
from rest_framework.routers import DefaultRouter
router = DefaultRouter()
router.register(r'device/apns', APNSDeviceAuthorizedViewSet)
router.register(r'device/gcm', GCMDeviceAuthorizedViewSet)
urlpatterns = patterns('',
# URLs will show up at <api_root>/device/apns
url(r'^', include(router.urls)),
# ...
)
- Using as_view_ (specify which views to include)
.. _as_view: http://www.django-rest-framework.org/tutorial/6-viewsets-and-routers#binding-viewsets-to-urls-explicitly
::
from push_notifications.api.rest_framework import APNSDeviceAuthorizedViewSet
urlpatterns = patterns('',
# Only allow creation of devices by authenticated users
url(r'^device/apns/?$', APNSDeviceAuthorizedViewSet.as_view({'post': 'create'}), name='create_apns_device'),
# ...
)
Python 3 support
----------------
``django-push-notifications`` is fully compatible with Python 3.4 & 3.5
.. [1] Any devices which are not selected, but are not receiving notifications will not be deactivated on a subsequent call to "prune devices" unless another attempt to send a message to the device fails after the call to the feedback service.

View File

@ -0,0 +1,8 @@
__author__ = "Jerome Leclanche"
__email__ = "jerome@leclan.ch"
__version__ = "1.4.1"
class NotificationError(Exception):
pass

View File

@ -0,0 +1,81 @@
from django.contrib import admin, messages
from django.contrib.auth import get_user_model
from django.utils.translation import ugettext_lazy as _
from .gcm import GCMError
from .models import APNSDevice, GCMDevice, get_expired_tokens
User = get_user_model()
class DeviceAdmin(admin.ModelAdmin):
list_display = ("__str__", "device_id", "user", "active", "date_created")
search_fields = ("name", "device_id", "user__%s" % (User.USERNAME_FIELD))
list_filter = ("active", )
actions = ("send_message", "send_bulk_message", "prune_devices", "enable", "disable")
def send_messages(self, request, queryset, bulk=False):
"""
Provides error handling for DeviceAdmin send_message and send_bulk_message methods.
"""
ret = []
errors = []
r = ""
for device in queryset:
try:
if bulk:
r = queryset.send_message("Test bulk notification")
else:
r = device.send_message("Test single notification")
if r:
ret.append(r)
except GCMError as e:
errors.append(str(e))
if bulk:
break
if errors:
self.message_user(request, _("Some messages could not be processed: %r" % (", ".join(errors))), level=messages.ERROR)
if ret:
if not bulk:
ret = ", ".join(ret)
if errors:
msg = _("Some messages were sent: %s" % (ret))
else:
msg = _("All messages were sent: %s" % (ret))
self.message_user(request, msg)
def send_message(self, request, queryset):
self.send_messages(request, queryset)
send_message.short_description = _("Send test message")
def send_bulk_message(self, request, queryset):
self.send_messages(request, queryset, True)
send_bulk_message.short_description = _("Send test message in bulk")
def enable(self, request, queryset):
queryset.update(active=True)
enable.short_description = _("Enable selected devices")
def disable(self, request, queryset):
queryset.update(active=False)
disable.short_description = _("Disable selected devices")
def prune_devices(self, request, queryset):
# Note that when get_expired_tokens() is called, Apple's
# feedback service resets, so, calling it again won't return
# the device again (unless a message is sent to it again). So,
# if the user doesn't select all the devices for pruning, we
# could very easily leave an expired device as active. Maybe
# this is just a bad API.
expired = get_expired_tokens()
devices = queryset.filter(registration_id__in=expired)
for d in devices:
d.active = False
d.save()
admin.site.register(APNSDevice, DeviceAdmin)
admin.site.register(GCMDevice, DeviceAdmin)

View File

@ -0,0 +1,12 @@
from django.conf import settings
if "tastypie" in settings.INSTALLED_APPS:
# Tastypie resources are importable from the api package level (backwards compatibility)
from .tastypie import APNSDeviceResource, GCMDeviceResource, APNSDeviceAuthenticatedResource, GCMDeviceAuthenticatedResource
__all__ = [
"APNSDeviceResource",
"GCMDeviceResource",
"APNSDeviceAuthenticatedResource",
"GCMDeviceAuthenticatedResource"
]

View File

@ -0,0 +1,127 @@
from __future__ import absolute_import
from rest_framework import permissions
from rest_framework.serializers import ModelSerializer, ValidationError
from rest_framework.validators import UniqueValidator
from rest_framework.viewsets import ModelViewSet
from rest_framework.fields import IntegerField
from push_notifications.models import APNSDevice, GCMDevice
from push_notifications.fields import hex_re
from push_notifications.fields import UNSIGNED_64BIT_INT_MAX_VALUE
# Fields
class HexIntegerField(IntegerField):
"""
Store an integer represented as a hex string of form "0x01".
"""
def to_internal_value(self, data):
# validate hex string and convert it to the unsigned
# integer representation for internal use
try:
data = int(data, 16)
except ValueError:
raise ValidationError("Device ID is not a valid hex number")
return super(HexIntegerField, self).to_internal_value(data)
def to_representation(self, value):
return value
# Serializers
class DeviceSerializerMixin(ModelSerializer):
class Meta:
fields = ("name", "registration_id", "device_id", "active", "date_created")
read_only_fields = ("date_created", )
# See https://github.com/tomchristie/django-rest-framework/issues/1101
extra_kwargs = {"active": {"default": True}}
class APNSDeviceSerializer(ModelSerializer):
class Meta(DeviceSerializerMixin.Meta):
model = APNSDevice
def validate_registration_id(self, value):
# iOS device tokens are 256-bit hexadecimal (64 characters). In 2016 Apple is increasing
# iOS device tokens to 100 bytes hexadecimal (200 characters).
if hex_re.match(value) is None or len(value) not in (64, 200):
raise ValidationError("Registration ID (device token) is invalid")
return value
class GCMDeviceSerializer(ModelSerializer):
device_id = HexIntegerField(
help_text="ANDROID_ID / TelephonyManager.getDeviceId() (e.g: 0x01)",
style={'input_type': 'text'},
required=False
)
class Meta(DeviceSerializerMixin.Meta):
model = GCMDevice
extra_kwargs = {
# Work around an issue with validating the uniqueness of
# registration ids of up to 4k
'registration_id': {
'validators': [
UniqueValidator(queryset=GCMDevice.objects.all())
]
}
}
def validate_device_id(self, value):
# device ids are 64 bit unsigned values
if value > UNSIGNED_64BIT_INT_MAX_VALUE:
raise ValidationError("Device ID is out of range")
return value
# Permissions
class IsOwner(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
# must be the owner to view the object
return obj.user == request.user
# Mixins
class DeviceViewSetMixin(object):
lookup_field = "registration_id"
def perform_create(self, serializer):
if self.request.user.is_authenticated():
serializer.save(user=self.request.user)
return super(DeviceViewSetMixin, self).perform_create(serializer)
class AuthorizedMixin(object):
permission_classes = (permissions.IsAuthenticated, IsOwner)
def get_queryset(self):
# filter all devices to only those belonging to the current user
return self.queryset.filter(user=self.request.user)
# ViewSets
class APNSDeviceViewSet(DeviceViewSetMixin, ModelViewSet):
queryset = APNSDevice.objects.all()
serializer_class = APNSDeviceSerializer
class APNSDeviceAuthorizedViewSet(AuthorizedMixin, APNSDeviceViewSet):
pass
class GCMDeviceViewSet(DeviceViewSetMixin, ModelViewSet):
queryset = GCMDevice.objects.all()
serializer_class = GCMDeviceSerializer
class GCMDeviceAuthorizedViewSet(AuthorizedMixin, GCMDeviceViewSet):
pass

View File

@ -0,0 +1,42 @@
from tastypie.authorization import Authorization
from tastypie.authentication import BasicAuthentication
from tastypie.resources import ModelResource
from push_notifications.models import APNSDevice, GCMDevice
class APNSDeviceResource(ModelResource):
class Meta:
authorization = Authorization()
queryset = APNSDevice.objects.all()
resource_name = "device/apns"
class GCMDeviceResource(ModelResource):
class Meta:
authorization = Authorization()
queryset = GCMDevice.objects.all()
resource_name = "device/gcm"
class APNSDeviceAuthenticatedResource(APNSDeviceResource):
# user = ForeignKey(UserResource, "user")
class Meta(APNSDeviceResource.Meta):
authentication = BasicAuthentication()
# authorization = SameUserAuthorization()
def obj_create(self, bundle, **kwargs):
# See https://github.com/toastdriven/django-tastypie/issues/854
return super(APNSDeviceAuthenticatedResource, self).obj_create(bundle, user=bundle.request.user, **kwargs)
class GCMDeviceAuthenticatedResource(GCMDeviceResource):
# user = ForeignKey(UserResource, "user")
class Meta(GCMDeviceResource.Meta):
authentication = BasicAuthentication()
# authorization = SameUserAuthorization()
def obj_create(self, bundle, **kwargs):
# See https://github.com/toastdriven/django-tastypie/issues/854
return super(GCMDeviceAuthenticatedResource, self).obj_create(bundle, user=bundle.request.user, **kwargs)

238
push_notifications/apns.py Normal file
View File

@ -0,0 +1,238 @@
"""
Apple Push Notification Service
Documentation is available on the iOS Developer Library:
https://developer.apple.com/library/ios/documentation/NetworkingInternet/Conceptual/RemoteNotificationsPG/Chapters/ApplePushService.html
"""
import codecs
import json
import ssl
import struct
import socket
import time
from contextlib import closing
from binascii import unhexlify
from django.core.exceptions import ImproperlyConfigured
from . import NotificationError
from .settings import PUSH_NOTIFICATIONS_SETTINGS as SETTINGS
class APNSError(NotificationError):
pass
class APNSServerError(APNSError):
def __init__(self, status, identifier):
super(APNSServerError, self).__init__(status, identifier)
self.status = status
self.identifier = identifier
class APNSDataOverflow(APNSError):
pass
def _apns_create_socket(address_tuple):
certfile = SETTINGS.get("APNS_CERTIFICATE")
if not certfile:
raise ImproperlyConfigured(
'You need to set PUSH_NOTIFICATIONS_SETTINGS["APNS_CERTIFICATE"] to send messages through APNS.'
)
try:
with open(certfile, "r") as f:
f.read()
except Exception as e:
raise ImproperlyConfigured("The APNS certificate file at %r is not readable: %s" % (certfile, e))
ca_certs = SETTINGS.get("APNS_CA_CERTIFICATES")
sock = socket.socket()
sock = ssl.wrap_socket(sock, ssl_version=ssl.PROTOCOL_TLSv1, certfile=certfile, ca_certs=ca_certs)
sock.connect(address_tuple)
return sock
def _apns_create_socket_to_push():
return _apns_create_socket((SETTINGS["APNS_HOST"], SETTINGS["APNS_PORT"]))
def _apns_create_socket_to_feedback():
return _apns_create_socket((SETTINGS["APNS_FEEDBACK_HOST"], SETTINGS["APNS_FEEDBACK_PORT"]))
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
frame_fmt = "!BIBH%ssBH%ssBHIBHIBHB" % (len(token), len(payload))
frame = struct.pack(
frame_fmt,
2, frame_len,
1, len(token), token,
2, len(payload), payload,
3, 4, identifier,
4, 4, expiration,
5, 1, priority)
return frame
def _apns_check_errors(sock):
timeout = SETTINGS["APNS_ERROR_TIMEOUT"]
if timeout is None:
return # assume everything went fine!
saved_timeout = sock.gettimeout()
try:
sock.settimeout(timeout)
data = sock.recv(6)
if data:
command, status, identifier = struct.unpack("!BBI", data)
# apple protocol says command is always 8. See http://goo.gl/ENUjXg
assert command == 8, "Command must be 8!"
if status != 0:
raise APNSServerError(status, identifier)
except socket.timeout: # py3, see http://bugs.python.org/issue10272
pass
except ssl.SSLError as e: # py2
if "timed out" not in e.message:
raise
finally:
sock.settimeout(saved_timeout)
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):
data = {}
aps_data = {}
if action_loc_key or loc_key or loc_args:
alert = {"body": alert} if alert else {}
if action_loc_key:
alert["action-loc-key"] = action_loc_key
if loc_key:
alert["loc-key"] = loc_key
if loc_args:
alert["loc-args"] = loc_args
if alert is not None:
aps_data["alert"] = alert
if badge is not None:
aps_data["badge"] = badge
if sound is not None:
aps_data["sound"] = sound
if category is not None:
aps_data["category"] = category
if content_available:
aps_data["content-available"] = 1
data["aps"] = aps_data
data.update(extra)
# convert to json, avoiding unnecessary whitespace with separators (keys sorted for tests)
json_data = json.dumps(data, separators=(",", ":"), sort_keys=True).encode("utf-8")
max_size = SETTINGS["APNS_MAX_NOTIFICATION_SIZE"]
if len(json_data) > max_size:
raise APNSDataOverflow("Notification body cannot exceed %i bytes" % (max_size))
# if expiration isn't specified use 1 month from now
expiration_time = expiration if expiration is not None else int(time.time()) + 2592000
frame = _apns_pack_frame(token, json_data, identifier, expiration_time, priority)
if socket:
socket.write(frame)
else:
with closing(_apns_create_socket_to_push()) as socket:
socket.write(frame)
_apns_check_errors(socket)
def _apns_read_and_unpack(socket, data_format):
length = struct.calcsize(data_format)
data = socket.recv(length)
if data:
return struct.unpack_from(data_format, data, 0)
else:
return None
def _apns_receive_feedback(socket):
expired_token_list = []
# read a timestamp (4 bytes) and device token length (2 bytes)
header_format = '!LH'
has_data = True
while has_data:
try:
# read the header tuple
header_data = _apns_read_and_unpack(socket, header_format)
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
device_token = _apns_read_and_unpack(socket, token_format)
if device_token is not None:
# _apns_read_and_unpack returns a tuple, but
# it's just one item, so get the first.
expired_token_list.append((timestamp, device_token[0]))
else:
has_data = False
except socket.timeout: # py3, see http://bugs.python.org/issue10272
pass
except ssl.SSLError as e: # py2
if "timed out" not in e.message:
raise
return expired_token_list
def apns_send_message(registration_id, alert, **kwargs):
"""
Sends an APNS notification to a single registration_id.
This will send the notification as form data.
If sending multiple notifications, it is more efficient to use
apns_send_bulk_message()
Note that if set alert should always be a string. If it is not set,
it won't be included in the notification. You will need to pass None
to this for silent notifications.
"""
_apns_send(registration_id, alert, **kwargs)
def apns_send_bulk_message(registration_ids, alert, **kwargs):
"""
Sends an APNS notification to one or more registration_ids.
The registration_ids argument needs to be a list.
Note that if set alert should always be a string. If it is not set,
it won't be included in the notification. You will need to pass None
to this for silent notifications.
"""
with closing(_apns_create_socket_to_push()) as socket:
for identifier, registration_id in enumerate(registration_ids):
_apns_send(registration_id, alert, identifier=identifier, socket=socket, **kwargs)
_apns_check_errors(socket)
def apns_fetch_inactive_ids():
"""
Queries the APNS server for id's that are no longer active since
the last fetch
"""
with closing(_apns_create_socket_to_feedback()) as socket:
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'))
return inactive_ids

View File

@ -0,0 +1,123 @@
import re
import struct
from django import forms
from django.core.validators import MaxValueValidator
from django.core.validators import MinValueValidator
from django.core.validators import RegexValidator
from django.db import models, connection
from django.utils import six
from django.utils.translation import ugettext_lazy as _
UNSIGNED_64BIT_INT_MIN_VALUE = 0
UNSIGNED_64BIT_INT_MAX_VALUE = 2 ** 64 - 1
__all__ = ["HexadecimalField", "HexIntegerField"]
hex_re = re.compile(r"^(([0-9A-f])|(0x[0-9A-f]))+$")
signed_integer_engines = [
"django.db.backends.postgresql_psycopg2",
"django.contrib.gis.db.backends.postgis",
"django.db.backends.sqlite3"
]
def _using_signed_storage():
return connection.settings_dict["ENGINE"] in signed_integer_engines
def _signed_to_unsigned_integer(value):
return struct.unpack("Q", struct.pack("q", value))[0]
def _unsigned_to_signed_integer(value):
return struct.unpack("q", struct.pack("Q", value))[0]
def _hex_string_to_unsigned_integer(value):
return int(value, 16)
def _unsigned_integer_to_hex_string(value):
return hex(value).rstrip("L")
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")]
super(HexadecimalField, self).__init__(*args, **kwargs)
def prepare_value(self, value):
# converts bigint from db to hex before it is displayed in admin
if value and not isinstance(value, six.string_types) \
and connection.vendor in ("mysql", "sqlite"):
value = _unsigned_integer_to_hex_string(value)
return super(forms.CharField, self).prepare_value(value)
class HexIntegerField(models.BigIntegerField):
"""
This field stores a hexadecimal *string* of up to 64 bits as an unsigned integer
on *all* backends including postgres.
Reasoning: Postgres only supports signed bigints. Since we don't care about
signedness, we store it as signed, and cast it to unsigned when we deal with
the actual value (with struct)
On sqlite and mysql, native unsigned bigint types are used. In all cases, the
value we deal with in python is always in hex.
"""
validators = [
MinValueValidator(UNSIGNED_64BIT_INT_MIN_VALUE),
MaxValueValidator(UNSIGNED_64BIT_INT_MAX_VALUE)
]
def db_type(self, connection):
engine = connection.settings_dict["ENGINE"]
if "mysql" in engine:
return "bigint unsigned"
elif "sqlite" in engine:
return "UNSIGNED BIG INT"
else:
return super(HexIntegerField, self).db_type(connection=connection)
def get_prep_value(self, value):
""" Return the integer value to be stored from the hex string """
if value is None or value == "":
return None
if isinstance(value, six.string_types):
value = _hex_string_to_unsigned_integer(value)
if _using_signed_storage():
value = _unsigned_to_signed_integer(value)
return value
def from_db_value(self, value, expression, connection, context):
""" Return an unsigned int representation from all db backends """
if value is None:
return value
if _using_signed_storage():
value = _signed_to_unsigned_integer(value)
return value
def to_python(self, value):
""" Return a str representation of the hexadecimal """
if isinstance(value, six.string_types):
return value
if value is None:
return value
return _unsigned_integer_to_hex_string(value)
def formfield(self, **kwargs):
defaults = {"form_class": HexadecimalField}
defaults.update(kwargs)
# yes, that super call is right
return super(models.IntegerField, self).formfield(**defaults)
def run_validators(self, value):
# make sure validation is performed on integer value not string value
value = _hex_string_to_unsigned_integer(value)
return super(models.BigIntegerField, self).run_validators(value)

194
push_notifications/gcm.py Normal file
View File

@ -0,0 +1,194 @@
"""
Google Cloud Messaging
Previously known as C2DM
Documentation is available on the Android Developer website:
https://developer.android.com/google/gcm/index.html
"""
import json
from .models import GCMDevice
try:
from urllib.request import Request, urlopen
from urllib.parse import urlencode
except ImportError:
# Python 2 support
from urllib2 import Request, urlopen
from urllib import urlencode
from django.core.exceptions import ImproperlyConfigured
from . import NotificationError
from .settings import PUSH_NOTIFICATIONS_SETTINGS as SETTINGS
class GCMError(NotificationError):
pass
def _chunks(l, n):
"""
Yield successive chunks from list \a l with a minimum size \a n
"""
for i in range(0, len(l), n):
yield l[i:i + 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.')
headers = {
"Content-Type": content_type,
"Authorization": "key=%s" % (key),
"Content-Length": str(len(data)),
}
request = Request(SETTINGS["GCM_POST_URL"], data, headers)
return urlopen(request).read().decode("utf-8")
def _gcm_send_plain(registration_id, data, **kwargs):
"""
Sends a GCM notification to a single registration_id.
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
"""
values = {"registration_id": registration_id}
for k, v in data.items():
values["data.%s" % (k)] = v.encode("utf-8")
for k, v in kwargs.items():
if v:
if isinstance(v, bool):
# Encode bools into ints
v = 1
values[k] = v
data = urlencode(sorted(values.items())).encode("utf-8") # sorted items for tests
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):
# 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
# server database. Otherwise, get the value of Error
if result.startswith("id"):
lines = result.split("\n")
if len(lines) > 1 and lines[1].startswith("registration_id"):
new_id = lines[1].split("=")[-1]
_gcm_handle_canonical_id(new_id, registration_id)
elif result.startswith("Error="):
if result in ("Error=NotRegistered", "Error=InvalidRegistration"):
# Deactivate the problematic device
device = GCMDevice.objects.filter(registration_id=values["registration_id"])
device.update(active=0)
return result
raise GCMError(result)
return result
def _gcm_send_json(registration_ids, data, **kwargs):
"""
Sends a GCM notification to one or more registration_ids. The registration_ids
needs to be a list.
This will send the notification as json data.
"""
values = {"registration_ids": registration_ids}
if data is not None:
values["data"] = data
for k, v in kwargs.items():
if v:
values[k] = v
data = json.dumps(values, separators=(",", ":"), sort_keys=True).encode("utf-8") # keys sorted for tests
response = json.loads(_gcm_send(data, "application/json"))
if response["failure"] or response["canonical_ids"]:
ids_to_remove, old_new_ids = [], []
throw_error = False
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
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).
new_id = result.get("registration_id")
if new_id:
old_new_ids.append((registration_ids[index], new_id))
if ids_to_remove:
removed = GCMDevice.objects.filter(registration_id__in=ids_to_remove)
removed.update(active=0)
for old_id, new_id in old_new_ids:
_gcm_handle_canonical_id(new_id, old_id)
if throw_error:
raise GCMError(response)
return response
def _gcm_handle_canonical_id(canonical_id, current_id):
"""
Handle situation when GCM server response contains canonical ID
"""
if GCMDevice.objects.filter(registration_id=canonical_id, active=True).exists():
GCMDevice.objects.filter(registration_id=current_id).update(active=False)
else:
GCMDevice.objects.filter(registration_id=current_id).update(registration_id=canonical_id)
def gcm_send_message(registration_id, data, **kwargs):
"""
Sends a GCM notification to a single registration_id.
If sending multiple notifications, it is more efficient to use
gcm_send_bulk_message() with a list of registration_ids
A reference of extra keyword arguments sent to the server is available here:
https://developers.google.com/cloud-messaging/server-ref#downstream
"""
return _gcm_send_plain(registration_id, data, **kwargs)
def gcm_send_bulk_message(registration_ids, data, **kwargs):
"""
Sends a GCM notification to one or more registration_ids. The registration_ids
needs to be a list.
This will send the notification as json data.
A reference of extra keyword arguments sent to the server is available here:
https://developers.google.com/cloud-messaging/server-ref#downstream
"""
# GCM only allows up to 1000 reg ids per bulk message
# https://developer.android.com/google/gcm/gcm.html#request
max_recipients = SETTINGS.get("GCM_MAX_RECIPIENTS")
if len(registration_ids) > max_recipients:
ret = []
for chunk in _chunks(registration_ids, max_recipients):
ret.append(_gcm_send_json(chunk, data, **kwargs))
return ret
return _gcm_send_json(registration_ids, data, **kwargs)

View File

@ -0,0 +1,16 @@
from django.core.management.base import BaseCommand
class Command(BaseCommand):
can_import_settings = True
help = 'Deactivate APNS devices that are not receiving notifications'
def handle(self, *args, **options):
from push_notifications.models import APNSDevice, get_expired_tokens
expired = get_expired_tokens()
devices = APNSDevice.objects.filter(registration_id__in=expired)
for d in devices:
self.stdout.write('deactivating [%s]' % d.registration_id)
d.active = False
d.save()
self.stdout.write('deactivated %d devices' % len(devices))

View File

@ -0,0 +1,48 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import models, migrations
import push_notifications.fields
from django.conf import settings
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='APNSDevice',
fields=[
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
('name', models.CharField(max_length=255, null=True, verbose_name='Name', blank=True)),
('active', models.BooleanField(default=True, help_text='Inactive devices will not be sent notifications', verbose_name='Is active')),
('date_created', models.DateTimeField(auto_now_add=True, verbose_name='Creation date', null=True)),
('device_id', models.UUIDField(help_text='UDID / UIDevice.identifierForVendor()', max_length=32, null=True, verbose_name='Device ID', blank=True, db_index=True)),
('registration_id', models.CharField(unique=True, max_length=64, verbose_name='Registration ID')),
('user', models.ForeignKey(blank=True, to=settings.AUTH_USER_MODEL, null=True)),
],
options={
'verbose_name': 'APNS device',
},
bases=(models.Model,),
),
migrations.CreateModel(
name='GCMDevice',
fields=[
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
('name', models.CharField(max_length=255, null=True, verbose_name='Name', blank=True)),
('active', models.BooleanField(default=True, help_text='Inactive devices will not be sent notifications', verbose_name='Is active')),
('date_created', models.DateTimeField(auto_now_add=True, verbose_name='Creation date', null=True)),
('device_id', push_notifications.fields.HexIntegerField(help_text='ANDROID_ID / TelephonyManager.getDeviceId() (always as hex)', null=True, verbose_name='Device ID', blank=True, db_index=True)),
('registration_id', models.TextField(verbose_name='Registration ID')),
('user', models.ForeignKey(blank=True, to=settings.AUTH_USER_MODEL, null=True)),
],
options={
'verbose_name': 'GCM device',
},
bases=(models.Model,),
),
]

View File

@ -0,0 +1,20 @@
# -*- coding: utf-8 -*-
# Generated by Django 1.9.1 on 2016-01-06 08:50
from __future__ import unicode_literals
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('push_notifications', '0001_initial'),
]
operations = [
migrations.AlterField(
model_name='apnsdevice',
name='registration_id',
field=models.CharField(max_length=200, unique=True, verbose_name='Registration ID'),
),
]

View File

@ -0,0 +1,100 @@
from __future__ import unicode_literals
from django.conf import settings
from django.db import models
from django.utils.encoding import python_2_unicode_compatible
from django.utils.translation import ugettext_lazy as _
from .fields import HexIntegerField
@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"))
user = models.ForeignKey(settings.AUTH_USER_MODEL, blank=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")
class GCMDeviceManager(models.Manager):
def get_queryset(self):
return GCMDeviceQuerySet(self.model)
class GCMDeviceQuerySet(models.query.QuerySet):
def send_message(self, message, **kwargs):
if self:
from .gcm import gcm_send_bulk_message
data = kwargs.pop("extra", {})
if message is not None:
data["message"] = message
reg_ids = list(self.filter(active=True).values_list('registration_id', flat=True))
return gcm_send_bulk_message(registration_ids=reg_ids, data=data, **kwargs)
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)"))
registration_id = models.TextField(verbose_name=_("Registration ID"))
objects = GCMDeviceManager()
class Meta:
verbose_name = _("GCM device")
def send_message(self, message, **kwargs):
from .gcm import gcm_send_message
data = kwargs.pop("extra", {})
if message is not None:
data["message"] = message
return gcm_send_message(registration_id=self.registration_id, data=data, **kwargs)
class APNSDeviceManager(models.Manager):
def get_queryset(self):
return APNSDeviceQuerySet(self.model)
class APNSDeviceQuerySet(models.query.QuerySet):
def send_message(self, message, **kwargs):
if self:
from .apns import apns_send_bulk_message
reg_ids = list(self.filter(active=True).values_list('registration_id', flat=True))
return apns_send_bulk_message(registration_ids=reg_ids, alert=message, **kwargs)
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)
objects = APNSDeviceManager()
class Meta:
verbose_name = _("APNS device")
def send_message(self, message, **kwargs):
from .apns import apns_send_message
return apns_send_message(registration_id=self.registration_id, alert=message, **kwargs)
# This is an APNS-only function right now, but maybe GCM will implement it
# in the future. But the definition of 'expired' may not be the same. Whatevs
def get_expired_tokens():
from .apns import apns_fetch_inactive_ids
return apns_fetch_inactive_ids()

View File

@ -0,0 +1,21 @@
from django.conf import settings
PUSH_NOTIFICATIONS_SETTINGS = getattr(settings, "PUSH_NOTIFICATIONS_SETTINGS", {})
# GCM
PUSH_NOTIFICATIONS_SETTINGS.setdefault("GCM_POST_URL", "https://android.googleapis.com/gcm/send")
PUSH_NOTIFICATIONS_SETTINGS.setdefault("GCM_MAX_RECIPIENTS", 1000)
# APNS
PUSH_NOTIFICATIONS_SETTINGS.setdefault("APNS_PORT", 2195)
PUSH_NOTIFICATIONS_SETTINGS.setdefault("APNS_FEEDBACK_PORT", 2196)
PUSH_NOTIFICATIONS_SETTINGS.setdefault("APNS_ERROR_TIMEOUT", None)
PUSH_NOTIFICATIONS_SETTINGS.setdefault("APNS_MAX_NOTIFICATION_SIZE", 2048)
if settings.DEBUG:
PUSH_NOTIFICATIONS_SETTINGS.setdefault("APNS_HOST", "gateway.sandbox.push.apple.com")
PUSH_NOTIFICATIONS_SETTINGS.setdefault("APNS_FEEDBACK_HOST", "feedback.sandbox.push.apple.com")
else:
PUSH_NOTIFICATIONS_SETTINGS.setdefault("APNS_HOST", "gateway.push.apple.com")
PUSH_NOTIFICATIONS_SETTINGS.setdefault("APNS_FEEDBACK_HOST", "feedback.push.apple.com")

1
requirements.txt Normal file
View File

@ -0,0 +1 @@
Django

2
setup.cfg Normal file
View File

@ -0,0 +1,2 @@
[wheel]
universal = 1

40
setup.py Executable file
View File

@ -0,0 +1,40 @@
#!/usr/bin/env python
import os.path
import push_notifications
from distutils.core import setup
README = open(os.path.join(os.path.dirname(__file__), "README.rst")).read()
CLASSIFIERS = [
"Development Status :: 5 - Production/Stable",
"Environment :: Web Environment",
"Framework :: Django",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python",
"Programming Language :: Python :: 2.7",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.4",
"Programming Language :: Python :: 3.5",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: System :: Networking",
]
setup(
name="django-push-notifications",
packages=[
"push_notifications",
"push_notifications/api",
"push_notifications/migrations",
"push_notifications/management",
"push_notifications/management/commands",
],
author=push_notifications.__author__,
author_email=push_notifications.__email__,
classifiers=CLASSIFIERS,
description="Send push notifications to mobile devices through GCM or APNS in Django.",
download_url="https://github.com/jleclanche/django-push-notifications/tarball/master",
long_description=README,
url="https://github.com/jleclanche/django-push-notifications",
version=push_notifications.__version__,
)

12
tests/__init__.py Normal file
View File

@ -0,0 +1,12 @@
from test_models import *
from test_gcm_push_payload import *
from test_apns_push_payload import *
from test_management_commands import *
# conditionally test rest_framework api if the DRF package is installed
try:
import rest_framework
except ImportError:
pass
else:
from test_rest_framework import *

17
tests/mock_responses.py Normal file
View File

@ -0,0 +1,17 @@
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"}]}'

57
tests/runtests.py Executable file
View File

@ -0,0 +1,57 @@
#!/usr/bin/env python
import os
import sys
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))
sys.path.insert(0, test_folder)
sys.path.insert(0, src_folder)
# define settings
import django.conf
os.environ[django.conf.ENVIRONMENT_VARIABLE] = "settings"
# set up environment
from django.test.utils import setup_test_environment
setup_test_environment()
# See https://docs.djangoproject.com/en/dev/releases/1.7/#app-loading-changes
import django
if django.VERSION >= (1, 7, 0):
django.setup()
# set up database
from django.db import connection
connection.creation.create_test_db()
def tear_down():
"""
tear down test environment
"""
# destroy test database
from django.db import connection
connection.creation.destroy_test_db("not_needed")
# teardown environment
from django.test.utils import teardown_test_environment
teardown_test_environment()
# fire in the hole!
if __name__ == "__main__":
setup()
import tests
unittest.main(module=tests)
tear_down()

23
tests/settings.py Normal file
View File

@ -0,0 +1,23 @@
# assert warnings are enabled
import warnings
warnings.simplefilter("ignore", Warning)
DATABASES = {
"default": {
"ENGINE": "django.db.backends.sqlite3",
}
}
INSTALLED_APPS = [
"django.contrib.admin",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.sites",
"push_notifications",
]
SITE_ID = 1
ROOT_URLCONF = "core.urls"
SECRET_KEY = "foobar"

View File

@ -0,0 +1,31 @@
import mock
from django.test import TestCase
from push_notifications.apns import _apns_send, APNSDataOverflow
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)
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)
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)
p.assert_called_once_with("123", b'{"aps":{"alert":"sample"},"foo":"bar"}', 10, 30, 10)
def test_oversized_payload(self):
socket = mock.MagicMock()
with mock.patch("push_notifications.apns._apns_pack_frame") as p:
self.assertRaises(APNSDataOverflow, _apns_send, "123", "_" * 2049, socket=socket)
p.assert_has_calls([])

View File

@ -0,0 +1,28 @@
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
class GCMPushPayloadTest(TestCase):
def test_push_payload(self):
with mock.patch("push_notifications.gcm._gcm_send", return_value=GCM_PLAIN_RESPONSE) as p:
gcm_send_message("abc", {"message": "Hello world"})
p.assert_called_once_with(
b"data.message=Hello+world&registration_id=abc",
"application/x-www-form-urlencoded;charset=UTF-8")
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)
p.assert_called_once_with(
b"data.message=Hello+world&delay_while_idle=1&registration_id=abc&time_to_live=3600",
"application/x-www-form-urlencoded;charset=UTF-8")
def test_bulk_push_payload(self):
with mock.patch("push_notifications.gcm._gcm_send", return_value=GCM_JSON_RESPONSE) as p:
gcm_send_bulk_message(["abc", "123"], {"message": "Hello world"})
p.assert_called_once_with(
b'{"data":{"message":"Hello world"},"registration_ids":["abc","123"]}',
"application/json")

View File

@ -0,0 +1,25 @@
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):
def test_prune_devices(self):
from push_notifications.models import APNSDevice
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:
receiver.side_effect = lambda s: [(b'', b'abc')]
call_command('prune_devices')
device = APNSDevice.objects.get(pk=device.pk)
self.assertFalse(device.active)

232
tests/test_models.py Normal file
View File

@ -0,0 +1,232 @@
import json
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 push_notifications.gcm import GCMError
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 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(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&registration_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=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&registration_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&registration_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",
)
GCMDevice.objects.create(
registration_id="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")
def test_gcm_send_message_active_devices(self):
GCMDevice.objects.create(
registration_id="abc",
active=True
)
GCMDevice.objects.create(
registration_id="xyz",
active=False
)
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_extra_to_multiple_devices(self):
GCMDevice.objects.create(
registration_id="abc",
)
GCMDevice.objects.create(
registration_id="abc1",
)
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")
def test_gcm_send_message_collapse_to_multiple_devices(self):
GCMDevice.objects.create(
registration_id="abc",
)
GCMDevice.objects.create(
registration_id="abc1",
)
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_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_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
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_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_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
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):
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
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)
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 create_devices(self, devices):
for device in devices:
GCMDevice.objects.create(
registration_id=device,
)

View File

@ -0,0 +1,120 @@
from django.test import TestCase
from push_notifications.api.rest_framework import APNSDeviceSerializer, GCMDeviceSerializer
from rest_framework.serializers import ValidationError
from tests.mock_responses import GCM_DRF_INVALID_HEX_ERROR, GCM_DRF_OUT_OF_RANGE_ERROR
class APNSDeviceSerializerTestCase(TestCase):
def test_validation(self):
# valid data - 32 bytes upper case
serializer = APNSDeviceSerializer(data={
"registration_id": "AEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAE",
"name": "Apple iPhone 6+",
"device_id": "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF",
})
self.assertTrue(serializer.is_valid())
# valid data - 32 bytes lower case
serializer = APNSDeviceSerializer(data={
"registration_id": "aeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeae",
"name": "Apple iPhone 6+",
"device_id": "ffffffffffffffffffffffffffffffff",
})
self.assertTrue(serializer.is_valid())
# valid data - 100 bytes upper case
serializer = APNSDeviceSerializer(data={
"registration_id": "AEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAEAE",
"name": "Apple iPhone 6+",
"device_id": "FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF",
})
self.assertTrue(serializer.is_valid())
# valid data - 100 bytes lower case
serializer = APNSDeviceSerializer(data={
"registration_id": "aeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeaeae",
"name": "Apple iPhone 6+",
"device_id": "ffffffffffffffffffffffffffffffff",
})
self.assertTrue(serializer.is_valid())
# invalid data - device_id, registration_id
serializer = APNSDeviceSerializer(data={
"registration_id": "invalid device token contains no hex",
"name": "Apple iPhone 6+",
"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):
def test_device_id_validation_pass(self):
serializer = GCMDeviceSerializer(data={
"registration_id": "foobar",
"name": "Galaxy Note 3",
"device_id": "0x1031af3b",
})
self.assertTrue(serializer.is_valid())
def test_registration_id_unique(self):
"""Validate that a duplicate registration id raises a validation error."""
# add a device
serializer = GCMDeviceSerializer(data={
"registration_id": "foobar",
"name": "Galaxy Note 3",
"device_id": "0x1031af3b",
})
serializer.is_valid(raise_exception=True)
obj = serializer.save()
# ensure updating the same object works
serializer = GCMDeviceSerializer(obj, data={
"registration_id": "foobar",
"name": "Galaxy Note 5",
"device_id": "0x1031af3b",
})
serializer.is_valid(raise_exception=True)
obj = serializer.save()
# try to add a new device with the same token
serializer = GCMDeviceSerializer(data={
"registration_id": "foobar",
"name": "Galaxy Note 3",
"device_id": "0xdeadbeaf",
})
with self.assertRaises(ValidationError) as ex:
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={
"registration_id": "foobar",
"name": "Galaxy Note 3",
"device_id": "0x10r",
})
self.assertFalse(serializer.is_valid())
self.assertEqual(serializer.errors, GCM_DRF_INVALID_HEX_ERROR)
def test_device_id_validation_fail_out_of_range(self):
serializer = GCMDeviceSerializer(data={
"registration_id": "foobar",
"name": "Galaxy Note 3",
"device_id": "10000000000000000", # 2**64
})
self.assertFalse(serializer.is_valid())
self.assertEqual(serializer.errors, GCM_DRF_OUT_OF_RANGE_ERROR)
def test_device_id_validation_value_between_signed_unsigned_64b_int_maximums(self):
"""
2**63 < 0xe87a4e72d634997c < 2**64
"""
serializer = GCMDeviceSerializer(data={
"registration_id": "foobar",
"name": "Nexus 5",
"device_id": "e87a4e72d634997c",
})
self.assertTrue(serializer.is_valid())

20
tox.ini Normal file
View File

@ -0,0 +1,20 @@
[tox]
envlist = {py27,py34,py35}--django{18,19}--drf{32,33},flake8
[testenv]
commands = python ./tests/runtests.py
deps=
django18: Django>=1.8,<1.9
django19: Django>=1.9,<2.0
mock==1.0.1
drf32: djangorestframework>=3.2,<3.3
drf33: djangorestframework>=3.3,<3.4
[testenv:flake8]
commands = flake8 push_notifications
deps = flake8
[flake8]
ignore = F403,W191,E126,E128
max-line-length = 160
exclude = push_notifications/migrations/*