Added support for Python3 and Django1.5+, added tests

.gitignore

@@ -0,0 +1,36 @@
+*.py[cod]
+
+# C extensions
+*.so
+
+# Packages
+*.egg
+*.egg-info
+dist
+build
+eggs
+parts
+bin
+var
+sdist
+develop-eggs
+.installed.cfg
+lib
+lib64
+__pycache__
+
+# Installer logs
+pip-log.txt
+
+# Unit test / coverage reports
+.coverage
+.tox
+nosetests.xml
+
+# Translations
+*.mo
+
+# Mr Developer
+.mr.developer.cfg
+.project
+.pydevproject

notification/backends/__init__.py

@@ -32,7 +32,7 @@ def load_backends():
             # import the module and get the module from sys.modules
             __import__(backend_mod)
             mod = sys.modules[backend_mod]
-        except ImportError, e:
+        except ImportError as e:
             raise exceptions.ImproperlyConfigured(
                 "Error importing notification backend {}: \"{}\"".format(backend_mod, e)
             )

notification/compat.py

@@ -0,0 +1,19 @@
+import django
+from django.conf import settings
+
+# Django 1.5 add support for custom auth user model
+if django.VERSION >= (1, 5):
+    AUTH_USER_MODEL = settings.AUTH_USER_MODEL
+else:
+    AUTH_USER_MODEL = 'auth.User'
+
+try:
+    from django.contrib.auth import get_user_model
+except ImportError:
+    from django.contrib.auth.models import User
+    get_user_model = lambda: User
+
+try:
+    from urllib import quote
+except ImportError:
+    from urllib.parse import quote

notification/engine.py

@@ -2,13 +2,13 @@
 import time
 import logging
 import traceback
-
-import cPickle as pickle
+import base64
 
 from django.conf import settings
 from django.core.mail import mail_admins
 from django.contrib.auth.models import User
 from django.contrib.sites.models import Site
+from django.utils.six.moves import cPickle as pickle
 
 from notification.lockfile import FileLock, AlreadyLocked, LockTimeout
 from notification.models import NoticeQueueBatch
@@ -44,7 +44,7 @@ def send_all(*args):
         # nesting the try statement to be Python 2.4
         try:
             for queued_batch in NoticeQueueBatch.objects.all():
-                notices = pickle.loads(str(queued_batch.pickled_data).decode("base64"))
+                notices = pickle.loads(base64.b64decode(queued_batch.pickled_data))
                 for user, label, extra_context, sender in notices:
                     try:
                         user = User.objects.get(pk=user)

notification/lockfile.py

@@ -12,7 +12,7 @@
 ... except AlreadyLocked:
 ...     print "somefile", "is locked already."
 ... except LockFailed:
-...     print "somefile", "can\\"t be locked."
+...     print "somefile", "can\\'t be locked."
 ... else:
 ...     print "got lock"
 got lock
@@ -52,11 +52,11 @@
 import sys
 import socket
 import os
-import thread
 import threading
 import time
 import errno
-import urllib
+
+from .compat import quote
 
 # Work with PEP8 and non-PEP8 versions of threading module.
 if not hasattr(threading, "current_thread"):
@@ -174,7 +174,7 @@ def __init__(self, path, threaded=True):
         self.pid = os.getpid()
         if threaded:
             name = threading.current_thread().get_name()
-            tname = "%s-" % urllib.quote(name, safe="")
+            tname = "%s-" % quote(name, safe="")
         else:
             tname = ""
         dirname = os.path.dirname(self.lock_file)
@@ -306,7 +306,7 @@ def __init__(self, path, threaded=True):
         """
         LockBase.__init__(self, path, threaded)
         if threaded:
-            tname = "%x-" % thread.get_ident()
+            tname = "%x-" % threading.get_ident()
         else:
             tname = ""
         # Lock file itself is a directory.  Place the unique file name into

notification/models.py

@@ -1,13 +1,18 @@
-import cPickle as pickle
+from __future__ import unicode_literals
+from __future__ import print_function
+
+import base64
 
 from django.db import models
 from django.db.models.query import QuerySet
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
 from django.utils.translation import ugettext_lazy as _
 from django.utils.translation import get_language, activate
+from django.utils.encoding import python_2_unicode_compatible
+from django.utils.six.moves import cPickle as pickle
 
-from django.contrib.auth.models import User
+from .compat import AUTH_USER_MODEL
 
 from notification import backends
 
@@ -28,6 +33,7 @@ def create_notice_type(label, display, description, **kwargs):
     NoticeType.create(label, display, description, **kwargs)
 
 
+@python_2_unicode_compatible
 class NoticeType(models.Model):
 
     label = models.CharField(_("label"), max_length=40)
@@ -37,7 +43,7 @@ class NoticeType(models.Model):
     # by default only on for media with sensitivity less than or equal to this number
     default = models.IntegerField(_("default"))
 
-    def __unicode__(self):
+    def __str__(self):
         return self.label
 
     class Meta:
@@ -66,11 +72,11 @@ def create(cls, label, display, description, default=2, verbosity=1):
             if updated:
                 notice_type.save()
                 if verbosity > 1:
-                    print "Updated %s NoticeType" % label
+                    print("Updated %s NoticeType" % label)
         except cls.DoesNotExist:
             cls(label=label, display=display, description=description, default=default).save()
             if verbosity > 1:
-                print "Created %s NoticeType" % label
+                print("Created %s NoticeType" % label)
 
 
 class NoticeSetting(models.Model):
@@ -79,7 +85,7 @@ class NoticeSetting(models.Model):
     of a given type to a given medium.
     """
 
-    user = models.ForeignKey(User, verbose_name=_("user"))
+    user = models.ForeignKey(AUTH_USER_MODEL, verbose_name=_("user"))
     notice_type = models.ForeignKey(NoticeType, verbose_name=_("notice type"))
     medium = models.CharField(_("medium"), max_length=1, choices=NOTICE_MEDIA)
     send = models.BooleanField(_("send"))
@@ -204,4 +210,4 @@ def queue(users, label, extra_context=None, sender=None):
     notices = []
     for user in users:
         notices.append((user, label, extra_context, sender))
-    NoticeQueueBatch(pickled_data=pickle.dumps(notices).encode("base64")).save()
+    NoticeQueueBatch(pickled_data=base64.b64encode(pickle.dumps(notices))).save()

notification/tests/models.py

@@ -0,0 +1,7 @@
+from django.db import models
+from ..compat import AUTH_USER_MODEL
+
+
+class Language(models.Model):
+    user = models.ForeignKey(AUTH_USER_MODEL)
+    language = models.CharField("language", max_length=10)

notification/tests/templates/404.html

@@ -0,0 +1,9 @@
+<!DOCTYPE html>
+<html>
+<head>
+    <title></title>
+</head>
+<body>
+
+</body>
+</html>

notification/tests/templates/account/base.html

@@ -0,0 +1,9 @@
+<!DOCTYPE html>
+<html>
+<head>
+    <title></title>
+</head>
+<body>
+
+</body>
+</html>

notification/tests/test_commands.py

@@ -0,0 +1,23 @@
+from django.test import TestCase
+from django.test.utils import override_settings
+from django.core import management, mail
+
+from ..compat import get_user_model
+from ..models import create_notice_type, NoticeType, queue
+
+
+class TestManagementCmd(TestCase):
+    def setUp(self):
+        self.user = get_user_model().objects.create_user("test_user", "test@user.com", "123456")
+        self.user2 = get_user_model().objects.create_user("test_user2", "test2@user.com", "123456")
+        create_notice_type("label", "display", "description")
+        self.notice_type = NoticeType.objects.get(label="label")
+
+    @override_settings(SITE_ID=1)
+    def test_emit_notices(self):
+        users = [self.user, self.user2]
+        queue(users, "label")
+        management.call_command("emit_notices")
+        self.assertEqual(len(mail.outbox), 2)
+        self.assertIn(self.user.email, mail.outbox[0].to)
+        self.assertIn(self.user2.email, mail.outbox[1].to)

notification/tests/test_models.py

@@ -0,0 +1,123 @@
+import base64
+
+from django.test import TestCase
+from django.test.utils import override_settings
+from django.conf import settings
+from django.contrib.sites.models import Site
+from django.core import mail
+from django.utils.six.moves import cPickle as pickle
+
+from ..models import NoticeType, NoticeSetting, NoticeQueueBatch
+from ..models import NOTICE_MEDIA, LanguageStoreNotAvailable
+from ..models import get_notification_language, create_notice_type, send_now, send, queue
+from ..compat import get_user_model
+
+from .models import Language
+
+
+class BaseTest(TestCase):
+    def setUp(self):
+        self.user = get_user_model().objects.create_user("test_user", "test@user.com", "123456")
+        self.user2 = get_user_model().objects.create_user("test_user2", "test2@user.com", "123456")
+        create_notice_type('label', 'display', 'description')
+        self.notice_type = NoticeType.objects.get(label='label')
+
+    def tearDown(self):
+        self.user.delete()
+        self.user2.delete()
+        self.notice_type.delete()
+
+
+class TestNoticeType(TestCase):
+    def test_create_notice_type(self):
+        label = "friends_invite"
+        create_notice_type(label, "Invitation Received", "you received an invitation")
+        n = NoticeType.objects.get(label=label)
+        self.assertEqual(str(n), label)
+
+    def test_create(self):
+        label = "friends_invite"
+        NoticeType.create(label, "Invitation Received", "you received an invitation", default=2,
+                          verbosity=2)
+        n = NoticeType.objects.get(label=label)
+        self.assertEqual(str(n), label)
+        # update
+        NoticeType.create(label, "Invitation for you", "you got an invitation", default=1,
+                          verbosity=2)
+        n = NoticeType.objects.get(pk=n.pk)
+        self.assertEqual(n.display, "Invitation for you")
+        self.assertEqual(n.description, "you got an invitation")
+        self.assertEqual(n.default, 1)
+
+
+class TestNoticeSetting(BaseTest):
+    def test_for_user(self):
+        email_id = NOTICE_MEDIA[0][0]
+        ns = NoticeSetting.objects.create(user=self.user, notice_type=self.notice_type,
+                                          medium=email_id, send=False)
+        self.assertEqual(NoticeSetting.for_user(self.user, self.notice_type, email_id), ns)
+
+        # test default fallback
+        NoticeSetting.for_user(self.user2, self.notice_type, email_id)
+        ns2 = NoticeSetting.objects.get(user=self.user2, notice_type=self.notice_type,
+                                        medium=email_id)
+        self.assertTrue(ns2.send)
+
+
+class TestProcedures(BaseTest):
+    def setUp(self):
+        super(TestProcedures, self).setUp()
+        self.lang = Language.objects.create(user=self.user, language='en_US')
+        mail.outbox = []
+
+    def tearDown(self):
+        super(TestProcedures, self).tearDown()
+        self.lang.delete()
+        NoticeQueueBatch.objects.all().delete()
+
+    @override_settings(NOTIFICATION_LANGUAGE_MODULE="tests.Language")
+    def test_get_notification_language(self):
+        self.assertEqual(get_notification_language(self.user), 'en_US')
+        self.assertRaises(LanguageStoreNotAvailable, get_notification_language, self.user2)
+        del settings.NOTIFICATION_LANGUAGE_MODULE
+        self.assertRaises(LanguageStoreNotAvailable, get_notification_language, self.user)
+
+    @override_settings(SITE_ID=1, NOTIFICATION_LANGUAGE_MODULE="tests.Language")
+    def test_send_now(self):
+        Site.objects.create(domain='localhost', name='localhost')
+        users = [self.user, self.user2]
+        send_now(users, 'label')
+        self.assertEqual(len(mail.outbox), 2)
+        self.assertIn(self.user.email, mail.outbox[0].to)
+        self.assertIn(self.user2.email, mail.outbox[1].to)
+
+    @override_settings(SITE_ID=1)
+    def test_send(self):
+        self.assertRaises(AssertionError, send, queue=True, now=True)
+
+        users = [self.user, self.user2]
+        send(users, 'label', now=True)
+        self.assertEqual(len(mail.outbox), 2)
+        self.assertIn(self.user.email, mail.outbox[0].to)
+        self.assertIn(self.user2.email, mail.outbox[1].to)
+
+        send(users, 'label', queue=True)
+        self.assertEqual(NoticeQueueBatch.objects.count(), 1)
+        batch = NoticeQueueBatch.objects.all()[0]
+        notices = pickle.loads(base64.b64decode(batch.pickled_data))
+        self.assertEqual(len(notices), 2)
+
+    @override_settings(SITE_ID=1)
+    def test_send_default(self):
+        # default behaviout, send_now
+        users = [self.user, self.user2]
+        send(users, 'label')
+        self.assertEqual(len(mail.outbox), 2)
+        self.assertEqual(NoticeQueueBatch.objects.count(), 0)
+
+    @override_settings(SITE_ID=1)
+    def test_queue_queryset(self):
+        users = get_user_model().objects.all()
+        queue(users, 'label')
+        self.assertEqual(len(mail.outbox), 0)
+        self.assertEqual(NoticeQueueBatch.objects.count(), 1)