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:
commit
7c850180f5
|
@ -0,0 +1,17 @@
|
|||
# python compiled
|
||||
__pycache__
|
||||
*.pyc
|
||||
|
||||
# distutils
|
||||
MANIFEST
|
||||
build
|
||||
|
||||
# IDE
|
||||
.idea
|
||||
*.iml
|
||||
|
||||
# virtualenv
|
||||
.env
|
||||
|
||||
# tox
|
||||
.tox
|
|
@ -0,0 +1,12 @@
|
|||
language: python
|
||||
addons:
|
||||
apt:
|
||||
sources:
|
||||
- deadsnakes
|
||||
packages:
|
||||
- python3.5
|
||||
install:
|
||||
- pip install tox
|
||||
script:
|
||||
- tox
|
||||
sudo: false
|
|
@ -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
|
|
@ -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
|
|
@ -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/)
|
|
@ -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.
|
|
@ -0,0 +1,3 @@
|
|||
include MANIFEST.in
|
||||
include README.rst
|
||||
include LICENSE
|
|
@ -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.
|
|
@ -0,0 +1,8 @@
|
|||
|
||||
__author__ = "Jerome Leclanche"
|
||||
__email__ = "jerome@leclan.ch"
|
||||
__version__ = "1.4.1"
|
||||
|
||||
|
||||
class NotificationError(Exception):
|
||||
pass
|
|
@ -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)
|
|
@ -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"
|
||||
]
|
|
@ -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
|
|
@ -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)
|
|
@ -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
|
|
@ -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)
|
|
@ -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)
|
|
@ -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))
|
|
@ -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,),
|
||||
),
|
||||
]
|
|
@ -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'),
|
||||
),
|
||||
]
|
|
@ -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()
|
|
@ -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")
|
|
@ -0,0 +1 @@
|
|||
Django
|
|
@ -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__,
|
||||
)
|
|
@ -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 *
|
|
@ -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"}]}'
|
|
@ -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()
|
|
@ -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"
|
|
@ -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([])
|
|
@ -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®istration_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®istration_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")
|
|
@ -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)
|
|
@ -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®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=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_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_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,
|
||||
)
|
|
@ -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())
|
|
@ -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/*
|
Loading…
Reference in New Issue