Hopiu/django-notifications
1
0
Fork 0
mirror of https://github.com/Hopiu/django-notifications.git synced 2026-03-16 21:30:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Lint all files

Browse source
This commit is contained in:
Alvaro Mariano 2023-06-24 00:10:23 +00:00
parent b035ee80cd
commit b75e5c66e5
42 changed files with 748 additions and 699 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.pre-commit-config.yaml
View file

@ -21,9 +21,11 @@ repos:
hooks:
- id: pylint
name: pylint
entry: pylint
entry: poetry run pylint
language: system
types: [python]
args: ["--rcfile", "pyproject.toml"]
exclude: "migrations"
- repo: https://github.com/PyCQA/bandit
rev: 1.7.5

2
manage.py
View file

@ -1,5 +1,5 @@
#!/usr/bin/env python
''' Django notification manage file '''
""" Django notification manage file """
import os
import sys

4
notifications/__init__.py
View file

@ -8,6 +8,6 @@
"""
# PEP 386-compliant version number: N.N[.N]+[{a|b|c|rc}N[.N]+][.postN][.devN]
__version__ = '1.8.0'
__version__ = "1.8.0"
default_app_config = 'notifications.apps.Config' # pylint: disable=invalid-name
default_app_config = "notifications.apps.Config" # pylint: disable=invalid-name

31
notifications/admin.py
View file

@ -1,29 +1,36 @@
''' Django notifications admin file '''
""" Django notifications admin file """
# -*- coding: utf-8 -*-
from django.contrib import admin
from django.utils.translation import gettext_lazy
from notifications.base.admin import AbstractNotificationAdmin
from swapper import load_model
Notification = load_model('notifications', 'Notification')
from notifications.base.admin import AbstractNotificationAdmin
Notification = load_model("notifications", "Notification")
def mark_unread(modeladmin, request, queryset):
def mark_unread(queryset, *args, **kwargs):
queryset.update(unread=True)
mark_unread.short_description = gettext_lazy('Mark selected notifications as unread')
mark_unread.short_description = gettext_lazy("Mark selected notifications as unread")
class NotificationAdmin(AbstractNotificationAdmin):
raw_id_fields = ('recipient',)
readonly_fields = ('action_object_url', 'actor_object_url', 'target_object_url')
list_display = ('recipient', 'actor',
'level', 'target', 'unread', 'public')
list_filter = ('level', 'unread', 'public', 'timestamp',)
raw_id_fields = ("recipient",)
readonly_fields = ("action_object_url", "actor_object_url", "target_object_url")
list_display = ("recipient", "actor", "level", "target", "unread", "public")
list_filter = (
"level",
"unread",
"public",
"timestamp",
)
actions = [mark_unread]
def get_queryset(self, request):
qs = super(NotificationAdmin, self).get_queryset(request)
return qs.prefetch_related('actor')
qs = super().get_queryset(request)
return qs.prefetch_related("actor")
admin.site.register(Notification, NotificationAdmin)

10
notifications/apps.py
View file

@ -1,4 +1,4 @@
''' Django notifications apps file '''
""" Django notifications apps file """
# -*- coding: utf-8 -*-
from django.apps import AppConfig
from django.utils.translation import gettext_lazy as _
@ -7,10 +7,4 @@ from django.utils.translation import gettext_lazy as _
class Config(AppConfig):
name = "notifications"
verbose_name = _("Notifications")
default_auto_field = 'django.db.models.AutoField'
def ready(self):
super(Config, self).ready()
# this is for backwards compatibility
import notifications.signals
notifications.notify = notifications.signals.notify
default_auto_field = "django.db.models.AutoField"

16
notifications/base/admin.py
View file

@ -2,11 +2,15 @@ from django.contrib import admin
class AbstractNotificationAdmin(admin.ModelAdmin):
raw_id_fields = ('recipient',)
list_display = ('recipient', 'actor',
'level', 'target', 'unread', 'public')
list_filter = ('level', 'unread', 'public', 'timestamp',)
raw_id_fields = ("recipient",)
list_display = ("recipient", "actor", "level", "target", "unread", "public")
list_filter = (
"level",
"unread",
"public",
"timestamp",
)
def get_queryset(self, request):
qs = super(AbstractNotificationAdmin, self).get_queryset(request)
return qs.prefetch_related('actor')
qs = super().get_queryset(request)
return qs.prefetch_related("actor")

168
notifications/base/models.py
View file

@ -2,26 +2,26 @@
# pylint: disable=too-many-lines
from django.conf import settings
from django.contrib.auth.models import Group
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ImproperlyConfigured
from django.db import models
from django.db.models.query import QuerySet
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from django.urls import NoReverseMatch, reverse
from django.utils import timesince, timezone
from django.utils.html import format_html
from django.utils.translation import gettext_lazy as _
from swapper import load_model
from notifications import settings as notifications_settings
from notifications.signals import notify
from notifications.utils import id2slug
from swapper import load_model
from django.contrib.contenttypes.fields import GenericForeignKey
from django.urls import reverse, NoReverseMatch
EXTRA_DATA = notifications_settings.get_config()['USE_JSONFIELD']
EXTRA_DATA = notifications_settings.get_config()["USE_JSONFIELD"]
def is_soft_delete():
return notifications_settings.get_config()['SOFT_DELETE']
return notifications_settings.get_config()["SOFT_DELETE"]
def assert_soft_delete():
@ -29,12 +29,13 @@ def assert_soft_delete():
# msg = """To use 'deleted' field, please set 'SOFT_DELETE'=True in settings.
# Otherwise NotificationQuerySet.unread and NotificationQuerySet.read do NOT filter by 'deleted' field.
# """
msg = 'REVERTME'
msg = "REVERTME"
raise ImproperlyConfigured(msg)
class NotificationQuerySet(models.query.QuerySet):
''' Notification QuerySet '''
"""Notification QuerySet"""
def unsent(self):
return self.filter(emailed=False)
@ -162,98 +163,96 @@ class AbstractNotification(models.Model):
HTML Representation::
<a href="http://oebfare.com/">brosner</a> commented on <a href="http://github.com/pinax/pinax">pinax/pinax</a> 2 hours ago # noqa
<a href="http://test.com/">brosner</a> commented on <a href="http://gh.com/pinax/pinax">pinax/pinax</a> 2 hours ago
"""
level = models.IntegerField(_('level'), choices=NotificationLevel.choices, default=NotificationLevel.INFO)
level = models.IntegerField(_("level"), choices=NotificationLevel.choices, default=NotificationLevel.INFO)
recipient = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='notifications',
verbose_name=_('recipient'),
related_name="notifications",
verbose_name=_("recipient"),
blank=False,
)
unread = models.BooleanField(_('unread'), default=True, blank=False, db_index=True)
unread = models.BooleanField(_("unread"), default=True, blank=False, db_index=True)
actor_content_type = models.ForeignKey(
ContentType,
on_delete=models.CASCADE,
related_name='notify_actor',
verbose_name=_('actor content type')
ContentType, on_delete=models.CASCADE, related_name="notify_actor", verbose_name=_("actor content type")
)
actor_object_id = models.CharField(_('actor object id'), max_length=255)
actor = GenericForeignKey('actor_content_type', 'actor_object_id')
actor.short_description = _('actor')
actor_object_id = models.CharField(_("actor object id"), max_length=255)
actor = GenericForeignKey("actor_content_type", "actor_object_id")
actor.short_description = _("actor")
verb = models.CharField(_('verb'), max_length=255)
description = models.TextField(_('description'), blank=True, null=True)
verb = models.CharField(_("verb"), max_length=255)
description = models.TextField(_("description"), blank=True, null=True)
target_content_type = models.ForeignKey(
ContentType,
on_delete=models.CASCADE,
related_name='notify_target',
verbose_name=_('target content type'),
related_name="notify_target",
verbose_name=_("target content type"),
blank=True,
null=True
null=True,
)
target_object_id = models.CharField(_('target object id'), max_length=255, blank=True, null=True)
target = GenericForeignKey('target_content_type', 'target_object_id')
target.short_description = _('target')
target_object_id = models.CharField(_("target object id"), max_length=255, blank=True, null=True)
target = GenericForeignKey("target_content_type", "target_object_id")
target.short_description = _("target")
action_object_content_type = models.ForeignKey(
ContentType,
on_delete=models.CASCADE,
related_name='notify_action_object',
verbose_name=_('action object content type'),
related_name="notify_action_object",
verbose_name=_("action object content type"),
blank=True,
null=True
null=True,
)
action_object_object_id = models.CharField(_('action object object id'), max_length=255, blank=True, null=True)
action_object = GenericForeignKey('action_object_content_type', 'action_object_object_id')
action_object.short_description = _('action object')
action_object_object_id = models.CharField(_("action object object id"), max_length=255, blank=True, null=True)
action_object = GenericForeignKey("action_object_content_type", "action_object_object_id")
action_object.short_description = _("action object")
timestamp = models.DateTimeField(_('timestamp'), default=timezone.now, db_index=True)
timestamp = models.DateTimeField(_("timestamp"), default=timezone.now, db_index=True)
public = models.BooleanField(_('public'), default=True, db_index=True)
deleted = models.BooleanField(_('deleted'), default=False, db_index=True)
emailed = models.BooleanField(_('emailed'), default=False, db_index=True)
public = models.BooleanField(_("public"), default=True, db_index=True)
deleted = models.BooleanField(_("deleted"), default=False, db_index=True)
emailed = models.BooleanField(_("emailed"), default=False, db_index=True)
data = models.JSONField(_('data'), blank=True, null=True)
data = models.JSONField(_("data"), blank=True, null=True)
objects = NotificationQuerySet.as_manager()
class Meta:
abstract = True
ordering = ('-timestamp',)
ordering = ("-timestamp",)
# speed up notifications count query
index_together = ('recipient', 'unread')
verbose_name = _('Notification')
verbose_name_plural = _('Notifications')
index_together = ("recipient", "unread")
verbose_name = _("Notification")
verbose_name_plural = _("Notifications")
def __str__(self):
ctx = {
'actor': self.actor,
'verb': self.verb,
'action_object': self.action_object,
'target': self.target,
'timesince': self.timesince()
"actor": self.actor,
"verb": self.verb,
"action_object": self.action_object,
"target": self.target,
"timesince": self.timesince(),
}
if self.target:
if self.action_object:
return _('%(actor)s %(verb)s %(action_object)s on %(target)s %(timesince)s ago') % ctx
return _('%(actor)s %(verb)s %(target)s %(timesince)s ago') % ctx
return _("%(actor)s %(verb)s %(action_object)s on %(target)s %(timesince)s ago") % ctx
return _("%(actor)s %(verb)s %(target)s %(timesince)s ago") % ctx
if self.action_object:
return _('%(actor)s %(verb)s %(action_object)s %(timesince)s ago') % ctx
return _('%(actor)s %(verb)s %(timesince)s ago') % ctx
return _("%(actor)s %(verb)s %(action_object)s %(timesince)s ago") % ctx
return _("%(actor)s %(verb)s %(timesince)s ago") % ctx
def timesince(self, now=None):
"""
Shortcut for the ``django.utils.timesince.timesince`` function of the
current timestamp.
"""
from django.utils.timesince import timesince as timesince_
return timesince_(self.timestamp, now)
return timesince.timesince(self.timestamp, now)
@property
def slug(self):
@ -271,27 +270,30 @@ class AbstractNotification(models.Model):
def actor_object_url(self):
try:
url = reverse("admin:{0}_{1}_change".format(self.actor_content_type.app_label,
self.actor_content_type.model),
args=(self.actor_object_id,))
url = reverse(
f"admin:{self.actor_content_type.app_label}_{self.actor_content_type.model}_change",
args=(self.actor_object_id,),
)
return format_html("<a href='{url}'>{id}</a>", url=url, id=self.actor_object_id)
except NoReverseMatch:
return self.actor_object_id
def action_object_url(self):
try:
url = reverse("admin:{0}_{1}_change".format(self.action_object_content_type.app_label,
self.action_content_type.model),
args=(self.action_object_id,))
url = reverse(
f"admin:{self.action_object_content_type.app_label}_{self.action_object_content_type.model}_change",
args=(self.action_object_id,),
)
return format_html("<a href='{url}'>{id}</a>", url=url, id=self.action_object_object_id)
except NoReverseMatch:
return self.action_object_object_id
def target_object_url(self):
try:
url = reverse("admin:{0}_{1}_change".format(self.target_content_type.app_label,
self.target_content_type.model),
args=(self.target_object_id,))
url = reverse(
f"admin:{self.target_content_type.app_label}_{self.target_content_type.model}_change",
args=(self.target_object_id,),
)
return format_html("<a href='{url}'>{id}</a>", url=url, id=self.target_object_id)
except NoReverseMatch:
return self.target_object_id
@ -302,19 +304,16 @@ def notify_handler(verb, **kwargs):
Handler function to create Notification instance upon action signal call.
"""
# Pull the options out of kwargs
kwargs.pop('signal', None)
recipient = kwargs.pop('recipient')
actor = kwargs.pop('sender')
optional_objs = [
(kwargs.pop(opt, None), opt)
for opt in ('target', 'action_object')
]
public = bool(kwargs.pop('public', True))
description = kwargs.pop('description', None)
timestamp = kwargs.pop('timestamp', timezone.now())
Notification = load_model('notifications', 'Notification')
level = kwargs.pop('level', NotificationLevel.INFO)
actor_for_concrete_model = kwargs.pop('actor_for_concrete_model', True)
kwargs.pop("signal", None)
recipient = kwargs.pop("recipient")
actor = kwargs.pop("sender")
optional_objs = [(kwargs.pop(opt, None), opt) for opt in ("target", "action_object")]
public = bool(kwargs.pop("public", True))
description = kwargs.pop("description", None)
timestamp = kwargs.pop("timestamp", timezone.now())
Notification = load_model("notifications", "Notification") # pylint: disable=invalid-name
level = kwargs.pop("level", NotificationLevel.INFO)
actor_for_concrete_model = kwargs.pop("actor_for_concrete_model", True)
# Check if User or Group
if isinstance(recipient, Group):
@ -341,10 +340,13 @@ def notify_handler(verb, **kwargs):
# Set optional objects
for obj, opt in optional_objs:
if obj is not None:
for_concrete_model = kwargs.pop(f'{opt}_for_concrete_model', True)
setattr(newnotify, '%s_object_id' % opt, obj.pk)
setattr(newnotify, '%s_content_type' % opt,
ContentType.objects.get_for_model(obj, for_concrete_model=for_concrete_model))
for_concrete_model = kwargs.pop(f"{opt}_for_concrete_model", True)
setattr(newnotify, f"{opt}_object_id", obj.pk)
setattr(
newnotify,
f"{opt}_content_type",
ContentType.objects.get_for_model(obj, for_concrete_model=for_concrete_model),
)
if kwargs and EXTRA_DATA:
# set kwargs as model column if available
@ -360,4 +362,4 @@ def notify_handler(verb, **kwargs):
# connect the signal
notify.connect(notify_handler, dispatch_uid='notifications.models.notification')
notify.connect(notify_handler, dispatch_uid="notifications.models.notification")

25
notifications/helpers.py
View file

@ -1,34 +1,37 @@
from django.forms import model_to_dict
from notifications.utils import id2slug
from notifications.settings import get_config
from notifications.utils import id2slug
def get_num_to_fetch(request):
default_num_to_fetch = get_config()['NUM_TO_FETCH']
default_num_to_fetch = get_config()["NUM_TO_FETCH"]
try:
# If they don't specify, make it 5.
num_to_fetch = request.GET.get('max', default_num_to_fetch)
num_to_fetch = request.GET.get("max", default_num_to_fetch)
num_to_fetch = int(num_to_fetch)
if not (1 <= num_to_fetch <= 100):
if not 1 <= num_to_fetch <= 100:
num_to_fetch = default_num_to_fetch
except ValueError: # If casting to an int fails.
num_to_fetch = default_num_to_fetch
return num_to_fetch
def get_notification_list(request, method_name='all'):
def get_notification_list(request, method_name="all"):
num_to_fetch = get_num_to_fetch(request)
notification_list = []
for notification in getattr(request.user.notifications, method_name)()[0:num_to_fetch]:
struct = model_to_dict(notification)
struct['slug'] = id2slug(notification.id)
struct["slug"] = id2slug(notification.id)
if notification.actor:
struct['actor'] = str(notification.actor)
struct["actor"] = str(notification.actor)
if notification.target:
struct['target'] = str(notification.target)
struct["target"] = str(notification.target)
if notification.action_object:
struct['action_object'] = str(notification.action_object)
struct["action_object"] = str(notification.action_object)
if notification.data:
struct['data'] = notification.data
struct["data"] = notification.data
notification_list.append(struct)
if request.GET.get('mark_as_read'):
if request.GET.get("mark_as_read"):
notification.mark_as_read()
return notification_list

76
notifications/migrations/0001_initial.py
View file

@ -1,39 +1,73 @@
# -*- coding: utf-8 -*-
from django.db import models, migrations
import django.utils.timezone
from django.conf import settings
import swapper
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
('contenttypes', '0001_initial'),
("contenttypes", "0001_initial"),
]
operations = [
migrations.CreateModel(
name='Notification',
name="Notification",
fields=[
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
('level', models.CharField(default='info', max_length=20, choices=[('success', 'success'), ('info', 'info'), ('warning', 'warning'), ('error', 'error')])),
('unread', models.BooleanField(default=True)),
('actor_object_id', models.CharField(max_length=255)),
('verb', models.CharField(max_length=255)),
('description', models.TextField(null=True, blank=True)),
('target_object_id', models.CharField(max_length=255, null=True, blank=True)),
('action_object_object_id', models.CharField(max_length=255, null=True, blank=True)),
('timestamp', models.DateTimeField(default=django.utils.timezone.now)),
('public', models.BooleanField(default=True)),
('action_object_content_type', models.ForeignKey(related_name='notify_action_object', blank=True, to='contenttypes.ContentType', null=True, on_delete=models.CASCADE)),
('actor_content_type', models.ForeignKey(related_name='notify_actor', to='contenttypes.ContentType', on_delete=models.CASCADE)),
('recipient', models.ForeignKey(related_name='notifications', to=settings.AUTH_USER_MODEL, on_delete=models.CASCADE)),
('target_content_type', models.ForeignKey(related_name='notify_target', blank=True, to='contenttypes.ContentType', null=True, on_delete=models.CASCADE)),
("id", models.AutoField(verbose_name="ID", serialize=False, auto_created=True, primary_key=True)),
(
"level",
models.CharField(
default="info",
max_length=20,
choices=[("success", "success"), ("info", "info"), ("warning", "warning"), ("error", "error")],
),
),
("unread", models.BooleanField(default=True)),
("actor_object_id", models.CharField(max_length=255)),
("verb", models.CharField(max_length=255)),
("description", models.TextField(null=True, blank=True)),
("target_object_id", models.CharField(max_length=255, null=True, blank=True)),
("action_object_object_id", models.CharField(max_length=255, null=True, blank=True)),
("timestamp", models.DateTimeField(default=django.utils.timezone.now)),
("public", models.BooleanField(default=True)),
(
"action_object_content_type",
models.ForeignKey(
related_name="notify_action_object",
blank=True,
to="contenttypes.ContentType",
null=True,
on_delete=models.CASCADE,
),
),
(
"actor_content_type",
models.ForeignKey(
related_name="notify_actor", to="contenttypes.ContentType", on_delete=models.CASCADE
),
),
(
"recipient",
models.ForeignKey(
related_name="notifications", to=settings.AUTH_USER_MODEL, on_delete=models.CASCADE
),
),
(
"target_content_type",
models.ForeignKey(
related_name="notify_target",
blank=True,
to="contenttypes.ContentType",
null=True,
on_delete=models.CASCADE,
),
),
],
options={
'swappable': swapper.swappable_setting('notifications', 'Notification'),
'ordering': ('-timestamp',),
"swappable": swapper.swappable_setting("notifications", "Notification"),
"ordering": ("-timestamp",),
},
bases=(models.Model,),
),

13
notifications/migrations/0002_auto_20150224_1134.py
View file

@ -1,23 +1,22 @@
# -*- coding: utf-8 -*-
from django.db import models, migrations
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('notifications', '0001_initial'),
("notifications", "0001_initial"),
]
operations = [
migrations.AddField(
model_name='notification',
name='deleted',
model_name="notification",
name="deleted",
field=models.BooleanField(default=False),
preserve_default=True,
),
migrations.AddField(
model_name='notification',
name='emailed',
model_name="notification",
name="emailed",
field=models.BooleanField(default=False),
preserve_default=True,
),

17
notifications/migrations/0003_notification_data.py
View file

@ -1,24 +1,17 @@
# -*- coding: utf-8 -*-
from django.db import models, migrations
try:
from jsonfield.fields import JSONField
except ModuleNotFoundError:
JSONField = models.JSONField
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('notifications', '0002_auto_20150224_1134'),
("notifications", "0002_auto_20150224_1134"),
]
operations = [
migrations.AddField(
model_name='notification',
name='data',
field=JSONField(null=True, blank=True),
model_name="notification",
name="data",
field=models.JSONField(null=True, blank=True),
preserve_default=True,
),
]

9
notifications/migrations/0004_auto_20150826_1508.py
View file

@ -1,18 +1,17 @@
# -*- coding: utf-8 -*-
from django.db import models, migrations
from django.db import migrations, models
from django.utils import timezone
class Migration(migrations.Migration):
dependencies = [
('notifications', '0003_notification_data'),
("notifications", "0003_notification_data"),
]
operations = [
migrations.AlterField(
model_name='notification',
name='timestamp',
model_name="notification",
name="timestamp",
field=models.DateTimeField(default=timezone.now),
),
]

13
notifications/migrations/0005_auto_20160504_1520.py
View file

@ -4,15 +4,18 @@ from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('notifications', '0004_auto_20150826_1508'),
("notifications", "0004_auto_20150826_1508"),
]
operations = [
migrations.AlterField(
model_name='notification',
name='level',
field=models.CharField(choices=[('success', 'success'), ('info', 'info'), ('warning', 'warning'), ('error', 'error')], default='info', max_length=20),
model_name="notification",
name="level",
field=models.CharField(
choices=[("success", "success"), ("info", "info"), ("warning", "warning"), ("error", "error")],
default="info",
max_length=20,
),
),
]

19
notifications/migrations/0006_indexes.py
View file

@ -4,30 +4,29 @@ from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('notifications', '0005_auto_20160504_1520'),
("notifications", "0005_auto_20160504_1520"),
]
operations = [
migrations.AlterField(
model_name='notification',
name='deleted',
model_name="notification",
name="deleted",
field=models.BooleanField(db_index=True, default=False),
),
migrations.AlterField(
model_name='notification',
name='emailed',
model_name="notification",
name="emailed",
field=models.BooleanField(db_index=True, default=False),
),
migrations.AlterField(
model_name='notification',
name='public',
model_name="notification",
name="public",
field=models.BooleanField(db_index=True, default=True),
),
migrations.AlterField(
model_name='notification',
name='unread',
model_name="notification",
name="unread",
field=models.BooleanField(db_index=True, default=True),
),
]

9
notifications/migrations/0007_add_timestamp_index.py
View file

@ -1,19 +1,18 @@
# Generated by Django 2.0.9 on 2018-10-26 10:41
from django.db import migrations, models
import django.utils.timezone
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('notifications', '0006_indexes'),
("notifications", "0006_indexes"),
]
operations = [
migrations.AlterField(
model_name='notification',
name='timestamp',
model_name="notification",
name="timestamp",
field=models.DateTimeField(db_index=True, default=django.utils.timezone.now),
),
]

7
notifications/migrations/0008_index_together_recipient_unread.py
View file

@ -5,15 +5,14 @@ from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
('notifications', '0007_add_timestamp_index'),
("notifications", "0007_add_timestamp_index"),
]
operations = [
migrations.AlterIndexTogether(
name='notification',
index_together={('recipient', 'unread')},
name="notification",
index_together={("recipient", "unread")},
),
]

30
notifications/migrations/0009_alter_notification_options_and_more.py
View file

@ -5,14 +5,8 @@ import django.utils.timezone
from django.conf import settings
from django.db import migrations, models
try:
from jsonfield.fields import JSONField
except ModuleNotFoundError:
JSONField = models.JSONField
class Migration(migrations.Migration):
dependencies = [
("contenttypes", "0002_remove_content_type_name"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
@ -68,16 +62,12 @@ class Migration(migrations.Migration):
migrations.AlterField(
model_name="notification",
name="data",
field=JSONField(
blank=True, null=True, verbose_name="data"
),
field=models.JSONField(blank=True, null=True, verbose_name="data"),
),
migrations.AlterField(
model_name="notification",
name="deleted",
field=models.BooleanField(
db_index=True, default=False, verbose_name="deleted"
),
field=models.BooleanField(db_index=True, default=False, verbose_name="deleted"),
),
migrations.AlterField(
model_name="notification",
@ -87,9 +77,7 @@ class Migration(migrations.Migration):
migrations.AlterField(
model_name="notification",
name="emailed",
field=models.BooleanField(
db_index=True, default=False, verbose_name="emailed"
),
field=models.BooleanField(db_index=True, default=False, verbose_name="emailed"),
),
migrations.AlterField(
model_name="notification",
@ -109,9 +97,7 @@ class Migration(migrations.Migration):
migrations.AlterField(
model_name="notification",
name="public",
field=models.BooleanField(
db_index=True, default=True, verbose_name="public"
),
field=models.BooleanField(db_index=True, default=True, verbose_name="public"),
),
migrations.AlterField(
model_name="notification",
@ -138,9 +124,7 @@ class Migration(migrations.Migration):
migrations.AlterField(
model_name="notification",
name="target_object_id",
field=models.CharField(
blank=True, max_length=255, null=True, verbose_name="target object id"
),
field=models.CharField(blank=True, max_length=255, null=True, verbose_name="target object id"),
),
migrations.AlterField(
model_name="notification",
@ -154,9 +138,7 @@ class Migration(migrations.Migration):
migrations.AlterField(
model_name="notification",
name="unread",
field=models.BooleanField(
db_index=True, default=True, verbose_name="unread"
),
field=models.BooleanField(db_index=True, default=True, verbose_name="unread"),
),
migrations.AlterField(
model_name="notification",

4
notifications/migrations/0012_auto_20230601_1905.py
View file

@ -1,9 +1,11 @@
# Generated by Django 4.2.1 on 2023-06-02 00:05
from django.db import migrations
from ..base.models import NotificationLevel
def copy_level(apps, schema_editor):
def copy_level(apps, *args):
Notification = apps.get_model("notifications", "Notification")
Notification.objects.filter(level="success").update(new_level=NotificationLevel.SUCCESS)
Notification.objects.filter(level="info").update(new_level=NotificationLevel.INFO)

7
notifications/models.py
View file

@ -1,13 +1,13 @@
from django.contrib.humanize.templatetags.humanize import naturalday, naturaltime
from swapper import swappable_setting
from .base.models import AbstractNotification
class Notification(AbstractNotification):
class Meta(AbstractNotification.Meta):
abstract = False
swappable = swappable_setting('notifications', 'Notification')
swappable = swappable_setting("notifications", "Notification")
def naturalday(self):
"""
@ -15,9 +15,8 @@ class Notification(AbstractNotification):
Take a parameter humanize_type. This parameter control the which humanize method use.
Return ``today``, ``yesterday`` ,``now``, ``2 seconds ago``etc.
"""
from django.contrib.humanize.templatetags.humanize import naturalday
return naturalday(self.timestamp)
def naturaltime(self):
from django.contrib.humanize.templatetags.humanize import naturaltime
return naturaltime(self.timestamp)

15
notifications/settings.py
View file

@ -1,19 +1,18 @@
''' Django notifications settings file '''
""" Django notifications settings file """
# -*- coding: utf-8 -*-
from django.conf import settings
CONFIG_DEFAULTS = {
'PAGINATE_BY': 20,
'USE_JSONFIELD': False,
'SOFT_DELETE': False,
'NUM_TO_FETCH': 10,
'CACHE_TIMEOUT': 2,
"PAGINATE_BY": 20,
"USE_JSONFIELD": False,
"SOFT_DELETE": False,
"NUM_TO_FETCH": 10,
"CACHE_TIMEOUT": 2,
}
def get_config():
user_config = getattr(settings, 'DJANGO_NOTIFICATIONS_CONFIG', {})
user_config = getattr(settings, "DJANGO_NOTIFICATIONS_CONFIG", {})
config = CONFIG_DEFAULTS.copy()
config.update(user_config)

2
notifications/signals.py
View file

@ -1,4 +1,4 @@
''' Django notifications signal file '''
""" Django notifications signal file """
# -*- coding: utf-8 -*-
from django.dispatch import Signal

12
notifications/templates/notifications/notice.html
View file

@ -2,23 +2,23 @@
<a class="close pull-right" href="{% url 'notifications:mark_as_read' notice.slug %}">
<i class="icon-close"></i>
</a>
<h4>
<i class="icon-mail{% if notice.unread %}-alt{% endif %}"></i>
{{ notice.actor }}
{{ notice.actor }}
{{ notice.verb }}
{% if notice.target %}
of {{ notice.target }}
{% endif %}
</h4>
<p>{{ notice.timesince }} ago</p>
<p>{{ notice.description|linebreaksbr }}</p>
<div class="notice-actions">
{% for action in notice.data.actions %}
<a class="btn" href="{{ action.href }}">{{ action.title }}</a>
{% endfor %}
</div>
</div>
</div>

89
notifications/templatetags/notifications_tags.py
View file

@ -1,27 +1,26 @@
''' Django notifications template tags file '''
""" Django notifications template tags file """
# -*- coding: utf-8 -*-
from django.template import Library
from django.utils.html import format_html
from django.core.cache import cache
from notifications import settings
from django.template import Library
from django.urls import reverse
from django.utils.html import format_html
from notifications import settings
register = Library()
def get_cached_notification_unread_count(user):
return cache.get_or_set(
'cache_notification_unread_count',
user.notifications.unread().count,
settings.get_config()['CACHE_TIMEOUT']
"cache_notification_unread_count", user.notifications.unread().count, settings.get_config()["CACHE_TIMEOUT"]
)
@register.simple_tag(takes_context=True)
def notifications_unread(context):
user = user_context(context)
if not user:
return ''
return ""
return get_cached_notification_unread_count(user)
@ -34,76 +33,66 @@ def has_notification(user):
# Requires vanilla-js framework - http://vanilla-js.com/
@register.simple_tag
def register_notify_callbacks(badge_class='live_notify_badge', # pylint: disable=too-many-arguments,missing-docstring
menu_class='live_notify_list',
refresh_period=15,
callbacks='',
api_name='list',
fetch=5,
nonce=None,
mark_as_read=False
):
def register_notify_callbacks(
badge_class="live_notify_badge", # pylint: disable=too-many-arguments,missing-docstring
menu_class="live_notify_list",
refresh_period=15,
callbacks="",
api_name="list",
fetch=5,
nonce=None,
mark_as_read=False,
):
refresh_period = int(refresh_period) * 1000
if api_name == 'list':
api_url = reverse('notifications:live_unread_notification_list')
elif api_name == 'count':
api_url = reverse('notifications:live_unread_notification_count')
if api_name == "list":
api_url = reverse("notifications:live_unread_notification_list")
elif api_name == "count":
api_url = reverse("notifications:live_unread_notification_count")
else:
return ""
definitions = """
definitions = f"""
notify_badge_class='{badge_class}';
notify_menu_class='{menu_class}';
notify_api_url='{api_url}';
notify_fetch_count='{fetch_count}';
notify_unread_url='{unread_url}';
notify_mark_all_unread_url='{mark_all_unread_url}';
notify_refresh_period={refresh};
notify_fetch_count='{fetch}';
notify_unread_url='{reverse("notifications:unread")}';
notify_mark_all_unread_url='{reverse("notifications:mark_all_as_read")}';
notify_refresh_period={refresh_period};
notify_mark_as_read={mark_as_read};
""".format(
badge_class=badge_class,
menu_class=menu_class,
refresh=refresh_period,
api_url=api_url,
unread_url=reverse('notifications:unread'),
mark_all_unread_url=reverse('notifications:mark_all_as_read'),
fetch_count=fetch,
mark_as_read=str(mark_as_read).lower()
)
"""
# add a nonce value to the script tag if one is provided
nonce_str = ' nonce="{nonce}"'.format(nonce=nonce) if nonce else ""
nonce_str = f' nonce="{nonce}"' if nonce else ""
script = '<script type="text/javascript"{nonce}>'.format(nonce=nonce_str) + definitions
for callback in callbacks.split(','):
script += "register_notifier(" + callback + ");"
script = f'<script type="text/javascript"{nonce_str}>' + definitions
for callback in callbacks.split(","):
script += f'register_notifier("{callback}");'
script += "</script>"
return format_html(script)
@register.simple_tag(takes_context=True)
def live_notify_badge(context, badge_class='live_notify_badge'):
def live_notify_badge(context, badge_class="live_notify_badge"):
user = user_context(context)
if not user:
return ''
return ""
html = "<span class='{badge_class}'>{unread}</span>".format(
badge_class=badge_class, unread=get_cached_notification_unread_count(user)
)
html = f"<span class='{badge_class}'>{get_cached_notification_unread_count(user)}</span>"
return format_html(html)
@register.simple_tag
def live_notify_list(list_class='live_notify_list'):
html = "<ul class='{list_class}'></ul>".format(list_class=list_class)
def live_notify_list(list_class="live_notify_list"):
html = f"<ul class='{list_class}'></ul>"
return format_html(html)
def user_context(context):
if 'user' not in context:
if "user" not in context:
return None
request = context['request']
request = context["request"]
user = request.user
if user.is_anonymous:

2
notifications/tests/factories/users.py
View file

@ -1,5 +1,5 @@
from django.conf import settings
import factory
from django.conf import settings
class Recipient(factory.django.DjangoModelFactory):

4
notifications/tests/sample_notifications/__init__.py
View file

@ -1 +1,3 @@
default_app_config = 'notifications.tests.sample_notifications.apps.SampleNotificationsConfig'
default_app_config = ( # pylint: disable=invalid-name
"notifications.tests.sample_notifications.apps.SampleNotificationsConfig"
)

3
notifications/tests/sample_notifications/admin.py
View file

@ -1,8 +1,9 @@
import swapper
from django.contrib import admin
from notifications.base.admin import AbstractNotificationAdmin
Notification = swapper.load_model('notifications', 'Notification')
Notification = swapper.load_model("notifications", "Notification")
@admin.register(Notification)

4
notifications/tests/sample_notifications/apps.py
View file

@ -2,5 +2,5 @@ from notifications.apps import Config as NotificationConfig
class SampleNotificationsConfig(NotificationConfig):
name = 'notifications.tests.sample_notifications'
label = 'sample_notifications'
name = "notifications.tests.sample_notifications"
label = "sample_notifications"

90
notifications/tests/sample_notifications/migrations/0001_initial.py
View file

@ -1,47 +1,85 @@
# Generated by Django 3.0.5 on 2020-04-11 12:15
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
import django.utils.timezone
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
('contenttypes', '0002_remove_content_type_name'),
("contenttypes", "0002_remove_content_type_name"),
]
operations = [
migrations.CreateModel(
name='Notification',
name="Notification",
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('level', models.CharField(choices=[('success', 'success'), ('info', 'info'), ('warning', 'warning'), ('error', 'error')], default='info', max_length=20)),
('unread', models.BooleanField(db_index=True, default=True)),
('actor_object_id', models.CharField(max_length=255)),
('verb', models.CharField(max_length=255)),
('description', models.TextField(blank=True, null=True)),
('target_object_id', models.CharField(blank=True, max_length=255, null=True)),
('action_object_object_id', models.CharField(blank=True, max_length=255, null=True)),
('timestamp', models.DateTimeField(db_index=True, default=django.utils.timezone.now)),
('public', models.BooleanField(db_index=True, default=True)),
('deleted', models.BooleanField(db_index=True, default=False)),
('emailed', models.BooleanField(db_index=True, default=False)),
('data', models.JSONField(blank=True, null=True)),
('details', models.CharField(blank=True, max_length=64, null=True)),
('action_object_content_type', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='notify_action_object', to='contenttypes.ContentType')),
('actor_content_type', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='notify_actor', to='contenttypes.ContentType')),
('recipient', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='notifications', to=settings.AUTH_USER_MODEL)),
('target_content_type', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='notify_target', to='contenttypes.ContentType')),
("id", models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
(
"level",
models.CharField(
choices=[("success", "success"), ("info", "info"), ("warning", "warning"), ("error", "error")],
default="info",
max_length=20,
),
),
("unread", models.BooleanField(db_index=True, default=True)),
("actor_object_id", models.CharField(max_length=255)),
("verb", models.CharField(max_length=255)),
("description", models.TextField(blank=True, null=True)),
("target_object_id", models.CharField(blank=True, max_length=255, null=True)),
("action_object_object_id", models.CharField(blank=True, max_length=255, null=True)),
("timestamp", models.DateTimeField(db_index=True, default=django.utils.timezone.now)),
("public", models.BooleanField(db_index=True, default=True)),
("deleted", models.BooleanField(db_index=True, default=False)),
("emailed", models.BooleanField(db_index=True, default=False)),
("data", models.JSONField(blank=True, null=True)),
("details", models.CharField(blank=True, max_length=64, null=True)),
(
"action_object_content_type",
models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name="notify_action_object",
to="contenttypes.ContentType",
),
),
(
"actor_content_type",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="notify_actor",
to="contenttypes.ContentType",
),
),
(
"recipient",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="notifications",
to=settings.AUTH_USER_MODEL,
),
),
(
"target_content_type",
models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name="notify_target",
to="contenttypes.ContentType",
),
),
],
options={
'ordering': ('-timestamp',),
'abstract': False,
'index_together': {('recipient', 'unread')},
"ordering": ("-timestamp",),
"abstract": False,
"index_together": {("recipient", "unread")},
},
),
]

1
notifications/tests/sample_notifications/models.py
View file

@ -1,4 +1,5 @@
from django.db import models
from notifications.base.models import AbstractNotification

0
notifications/tests/sample_notifications/templatetags/__init__.py
View file

1
notifications/tests/sample_notifications/templatetags/notifications_tags.py
View file

@ -1 +0,0 @@
from notifications.templatetags.notifications_tags import register

16
notifications/tests/sample_notifications/tests.py
View file

@ -1,26 +1,26 @@
import os
from unittest import skipUnless
from django.contrib.auth.models import User
import swapper
from django.contrib.auth import get_user_model
from notifications.signals import notify
from notifications.tests.tests import AdminTest as BaseAdminTest
from notifications.tests.tests import NotificationTest as BaseNotificationTest
Notification = swapper.load_model('notifications', 'Notification')
Notification = swapper.load_model("notifications", "Notification")
User = get_user_model()
@skipUnless(os.environ.get('SAMPLE_APP', False), 'Running tests on standard django-notifications models')
@skipUnless(os.environ.get("SAMPLE_APP", False), "Running tests on standard django-notifications models")
class AdminTest(BaseAdminTest):
@classmethod
def setUpClass(cls):
super().setUpClass()
BaseAdminTest.app_name = 'sample_notifications'
BaseAdminTest.app_name = "sample_notifications"
@skipUnless(os.environ.get('SAMPLE_APP', False), 'Running tests on standard django-notifications models')
@skipUnless(os.environ.get("SAMPLE_APP", False), "Running tests on standard django-notifications models")
class NotificationTest(BaseNotificationTest):
pass
@ -32,11 +32,11 @@ class TestExtraDataCustomAccessor(NotificationTest):
notify.send(
self.from_user,
recipient=self.to_user,
verb='commented',
verb="commented",
action_object=self.from_user,
url="/learn/ask-a-pro/q/test-question-9/299/",
other_content="Hello my 'world'",
details="test detail"
details="test detail",
)
def test_extra_data(self):

75
notifications/tests/settings.py
View file

@ -1,40 +1,39 @@
''' Django notification settings for tests '''
""" Django notification settings for tests """
# -*- coding: utf-8 -*-
import os
BASE_DIR = os.path.abspath(os.path.dirname(__file__))
SECRET_KEY = 'secret_key' # noqa
SECRET_KEY = "secret_key"
DEBUG = True
TESTING = True
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': 'test.sqlite3',
"default": {
"ENGINE": "django.db.backends.sqlite3",
"NAME": "test.sqlite3",
}
}
MIDDLEWARE = (
'django.contrib.sessions.middleware.SessionMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware'
"django.contrib.sessions.middleware.SessionMiddleware",
"django.contrib.auth.middleware.AuthenticationMiddleware",
"django.contrib.messages.middleware.MessageMiddleware",
)
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.messages',
'django.contrib.staticfiles',
'django.contrib.sessions',
'notifications.tests',
'notifications',
"django.contrib.admin",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.messages",
"django.contrib.staticfiles",
"django.contrib.sessions",
"notifications.tests",
"notifications",
]
ROOT_URLCONF = 'notifications.tests.urls'
STATIC_URL = '/static/'
ROOT_URLCONF = "notifications.tests.urls"
STATIC_URL = "/static/"
STATICFILES_DIRS = [
os.path.join(BASE_DIR, "static"),
]
@ -42,36 +41,36 @@ STATIC_ROOT = os.path.join(BASE_DIR, "static-files")
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'OPTIONS': {
'loaders' : [
'django.template.loaders.filesystem.Loader',
'django.template.loaders.app_directories.Loader',
"BACKEND": "django.template.backends.django.DjangoTemplates",
"DIRS": [],
"OPTIONS": {
"loaders": [
"django.template.loaders.filesystem.Loader",
"django.template.loaders.app_directories.Loader",
],
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
"context_processors": [
"django.template.context_processors.debug",
"django.template.context_processors.request",
"django.contrib.auth.context_processors.auth",
"django.contrib.messages.context_processors.messages",
],
},
},
]
LOGIN_REDIRECT_URL = 'test/'
LOGIN_URL = '/admin/login/'
LOGIN_REDIRECT_URL = "test/"
LOGIN_URL = "/admin/login/"
APPEND_SLASH = True
DJANGO_NOTIFICATIONS_CONFIG = {
'USE_JSONFIELD': True,
"USE_JSONFIELD": True,
}
USE_TZ = True
if os.environ.get('SAMPLE_APP', False):
INSTALLED_APPS.remove('notifications')
INSTALLED_APPS.append('notifications.tests.sample_notifications')
NOTIFICATIONS_NOTIFICATION_MODEL = 'sample_notifications.Notification'
TEMPLATES[0]['DIRS'] += [os.path.join(BASE_DIR, '../templates')]
if os.environ.get("SAMPLE_APP", False):
INSTALLED_APPS.remove("notifications")
INSTALLED_APPS.append("notifications.tests.sample_notifications")
NOTIFICATIONS_NOTIFICATION_MODEL = "sample_notifications.Notification"
TEMPLATES[0]["DIRS"] += [os.path.join(BASE_DIR, "../templates")]
ALLOWED_HOSTS = ["*"]

2
notifications/tests/static/notifications/live-test.js
View file

@ -4,4 +4,4 @@ function make_notification() {
var r = new XMLHttpRequest();
r.open("GET", '/test_make/', true);
r.send();
}
}

10
notifications/tests/test_migrations/test_level_migration.py
View file

@ -8,9 +8,9 @@ def test_main_migration0002(migrator):
"""Ensures that the second migration works."""
old_state = migrator.apply_initial_migration(("notifications", "0011_notification_new_level"))
OldUser = old_state.apps.get_model("auth", "User")
OldNotification = old_state.apps.get_model("notifications", "Notification")
OldContentType = old_state.apps.get_model("contenttypes", "ContentType")
OldUser = old_state.apps.get_model("auth", "User") # pylint: disable=invalid-name
OldNotification = old_state.apps.get_model("notifications", "Notification") # pylint: disable=invalid-name
OldContentType = old_state.apps.get_model("contenttypes", "ContentType") # pylint: disable=invalid-name
mark_follower = factory.create(OldUser, FACTORY_CLASS=user_factory.Recipient)
guido = factory.create(OldUser, FACTORY_CLASS=user_factory.Target)
@ -35,7 +35,7 @@ def test_main_migration0002(migrator):
assert OldNotification.objects.filter(new_level=NotificationLevel.INFO).count() == 4
new_state = migrator.apply_tested_migration(("notifications", "0012_auto_20230601_1905"))
NewNotification = new_state.apps.get_model("notifications", "Notification")
NewNotification = new_state.apps.get_model("notifications", "Notification") # pylint: disable=invalid-name
assert NewNotification.objects.count() == 4
assert NewNotification.objects.filter(new_level=NotificationLevel.SUCCESS).count() == 1
@ -44,7 +44,7 @@ def test_main_migration0002(migrator):
assert NewNotification.objects.filter(new_level=NotificationLevel.ERROR).count() == 1
new_state_2 = migrator.apply_tested_migration(("notifications", "0014_rename_new_level_notification_level"))
NewNotification2 = new_state_2.apps.get_model("notifications", "Notification")
NewNotification2 = new_state_2.apps.get_model("notifications", "Notification") # pylint: disable=invalid-name
assert NewNotification2.objects.count() == 4
assert NewNotification2.objects.filter(level=NotificationLevel.SUCCESS).count() == 1

488
notifications/tests/tests.py
View file

@ -1,32 +1,34 @@
'''
"""
This file demonstrates writing tests using the unittest module. These will pass
when you run "manage.py test".
Replace this with more appropriate tests for your application.
'''
"""
# -*- coding: utf-8 -*-
# pylint: disable=too-many-lines,missing-docstring
import json
from datetime import datetime, timezone
import pytz
from django.conf import settings
from django.contrib.auth.models import Group, User
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from django.core.exceptions import ImproperlyConfigured
from django.db import connection
from django.test import override_settings # noqa
from django.shortcuts import render
from django.template import Context, Template
from django.test import override_settings # noqa
from django.test import RequestFactory, TestCase
from django.test.utils import CaptureQueriesContext
from django.urls import reverse
from django.utils import timezone
from django.utils.timezone import localtime, utc
from django.utils.timezone import localtime
from swapper import load_model
from notifications.base.models import notify_handler
from notifications.signals import notify
from notifications.utils import id2slug
from swapper import load_model
Notification = load_model('notifications', 'Notification')
Notification = load_model("notifications", "Notification")
User = get_user_model()
MALICIOUS_NEXT_URLS = [
"http://bla.com",
@ -36,17 +38,19 @@ MALICIOUS_NEXT_URLS = [
"ftp://www.bla.com/file.exe",
]
class NotificationTest(TestCase):
''' Django notifications automated tests '''
"""Django notifications automated tests"""
@override_settings(USE_TZ=True)
@override_settings(TIME_ZONE='Asia/Shanghai')
@override_settings(TIME_ZONE="Asia/Shanghai")
def test_use_timezone(self):
from_user = User.objects.create(username="from", password="pwd", email="example@example.com")
to_user = User.objects.create(username="to", password="pwd", email="example@example.com")
notify.send(from_user, recipient=to_user, verb='commented', action_object=from_user)
notify.send(from_user, recipient=to_user, verb="commented", action_object=from_user)
notification = Notification.objects.get(recipient=to_user)
delta = (
timezone.now().replace(tzinfo=utc) - localtime(notification.timestamp, pytz.timezone(settings.TIME_ZONE))
delta = datetime.now().replace(tzinfo=timezone.utc) - localtime(
notification.timestamp, pytz.timezone(settings.TIME_ZONE)
)
self.assertTrue(delta.seconds < 60)
# The delta between the two events will still be less than a second despite the different timezones
@ -54,31 +58,33 @@ class NotificationTest(TestCase):
# test above was originally.
@override_settings(USE_TZ=False)
@override_settings(TIME_ZONE='Asia/Shanghai')
@override_settings(TIME_ZONE="Asia/Shanghai")
def test_disable_timezone(self):
from_user = User.objects.create(username="from2", password="pwd", email="example@example.com")
to_user = User.objects.create(username="to2", password="pwd", email="example@example.com")
notify.send(from_user, recipient=to_user, verb='commented', action_object=from_user)
notify.send(from_user, recipient=to_user, verb="commented", action_object=from_user)
notification = Notification.objects.get(recipient=to_user)
delta = timezone.now() - notification.timestamp
delta = datetime.now() - notification.timestamp
self.assertTrue(delta.seconds < 60)
def test_humanize_naturalday_timestamp(self):
from_user = User.objects.create(username="from2", password="pwd", email="example@example.com")
to_user = User.objects.create(username="to2", password="pwd", email="example@example.com")
notify.send(from_user, recipient=to_user, verb='commented', action_object=from_user)
notify.send(from_user, recipient=to_user, verb="commented", action_object=from_user)
notification = Notification.objects.get(recipient=to_user)
self.assertEqual(notification.naturalday(), 'today')
self.assertEqual(notification.naturalday(), "today")
def test_humanize_naturaltime_timestamp(self):
from_user = User.objects.create(username="from2", password="pwd", email="example@example.com")
to_user = User.objects.create(username="to2", password="pwd", email="example@example.com")
notify.send(from_user, recipient=to_user, verb='commented', action_object=from_user)
notify.send(from_user, recipient=to_user, verb="commented", action_object=from_user)
notification = Notification.objects.get(recipient=to_user)
self.assertEqual(notification.naturaltime(), 'now')
self.assertEqual(notification.naturaltime(), "now")
class NotificationManagersTest(TestCase):
''' Django notifications Manager automated tests '''
"""Django notifications Manager automated tests"""
def setUp(self):
self.message_count = 10
self.other_user = User.objects.create(username="other1", password="pwd", email="example@example.com")
@ -91,16 +97,16 @@ class NotificationManagersTest(TestCase):
self.to_group.user_set.add(self.other_user)
for _ in range(self.message_count):
notify.send(self.from_user, recipient=self.to_user, verb='commented', action_object=self.from_user)
notify.send(self.from_user, recipient=self.to_user, verb="commented", action_object=self.from_user)
# Send notification to group
notify.send(self.from_user, recipient=self.to_group, verb='commented', action_object=self.from_user)
notify.send(self.from_user, recipient=self.to_group, verb="commented", action_object=self.from_user)
self.message_count += self.to_group.user_set.count()
# Send notification to user list
notify.send(self.from_user, recipient=self.to_user_list, verb='commented', action_object=self.from_user)
notify.send(self.from_user, recipient=self.to_user_list, verb="commented", action_object=self.from_user)
self.message_count += len(self.to_user_list)
def test_notify_send_return_val(self):
results = notify.send(self.from_user, recipient=self.to_user, verb='commented', action_object=self.from_user)
results = notify.send(self.from_user, recipient=self.to_user, verb="commented", action_object=self.from_user)
for result in results:
if result[0] is notify_handler:
self.assertEqual(len(result[1]), 1)
@ -108,7 +114,7 @@ class NotificationManagersTest(TestCase):
self.assertEqual(type(result[1][0]), Notification)
def test_notify_send_return_val_group(self): # pylint: disable=invalid-name
results = notify.send(self.from_user, recipient=self.to_group, verb='commented', action_object=self.from_user)
results = notify.send(self.from_user, recipient=self.to_group, verb="commented", action_object=self.from_user)
for result in results:
if result[0] is notify_handler:
self.assertEqual(len(result[1]), self.to_group.user_set.count())
@ -120,7 +126,7 @@ class NotificationManagersTest(TestCase):
self.assertEqual(Notification.objects.unread().count(), self.message_count)
notification = Notification.objects.filter(recipient=self.to_user).first()
notification.mark_as_read()
self.assertEqual(Notification.objects.unread().count(), self.message_count-1)
self.assertEqual(Notification.objects.unread().count(), self.message_count - 1)
for notification in Notification.objects.unread():
self.assertTrue(notification.unread)
@ -137,18 +143,16 @@ class NotificationManagersTest(TestCase):
Notification.objects.filter(recipient=self.to_user).mark_all_as_read()
self.assertEqual(self.to_user.notifications.unread().count(), 0)
@override_settings(DJANGO_NOTIFICATIONS_CONFIG={
'SOFT_DELETE': True
}) # pylint: disable=invalid-name
@override_settings(DJANGO_NOTIFICATIONS_CONFIG={"SOFT_DELETE": True}) # pylint: disable=invalid-name
def test_mark_all_as_read_manager_with_soft_delete(self):
# even soft-deleted notifications should be marked as read
# refer: https://github.com/django-notifications/django-notifications/issues/126
to_delete = Notification.objects.filter(recipient=self.to_user).order_by('id')[0]
to_delete = Notification.objects.filter(recipient=self.to_user).order_by("id")[0]
to_delete.deleted = True
to_delete.save()
self.assertTrue(Notification.objects.filter(recipient=self.to_user).order_by('id')[0].unread)
self.assertTrue(Notification.objects.filter(recipient=self.to_user).order_by("id")[0].unread)
Notification.objects.filter(recipient=self.to_user).mark_all_as_read()
self.assertFalse(Notification.objects.filter(recipient=self.to_user).order_by('id')[0].unread)
self.assertFalse(Notification.objects.filter(recipient=self.to_user).order_by("id")[0].unread)
def test_mark_all_as_unread_manager(self):
self.assertEqual(Notification.objects.unread().count(), self.message_count)
@ -163,14 +167,12 @@ class NotificationManagersTest(TestCase):
self.assertRaises(ImproperlyConfigured, Notification.objects.mark_all_as_deleted)
self.assertRaises(ImproperlyConfigured, Notification.objects.mark_all_as_active)
@override_settings(DJANGO_NOTIFICATIONS_CONFIG={
'SOFT_DELETE': True
})
@override_settings(DJANGO_NOTIFICATIONS_CONFIG={"SOFT_DELETE": True})
def test_mark_all_deleted_manager(self):
notification = Notification.objects.filter(recipient=self.to_user).first()
notification.mark_as_read()
self.assertEqual(Notification.objects.read().count(), 1)
self.assertEqual(Notification.objects.unread().count(), self.message_count-1)
self.assertEqual(Notification.objects.unread().count(), self.message_count - 1)
self.assertEqual(Notification.objects.active().count(), self.message_count)
self.assertEqual(Notification.objects.deleted().count(), 0)
@ -182,13 +184,14 @@ class NotificationManagersTest(TestCase):
Notification.objects.mark_all_as_active()
self.assertEqual(Notification.objects.read().count(), 1)
self.assertEqual(Notification.objects.unread().count(), self.message_count-1)
self.assertEqual(Notification.objects.unread().count(), self.message_count - 1)
self.assertEqual(Notification.objects.active().count(), self.message_count)
self.assertEqual(Notification.objects.deleted().count(), 0)
class NotificationTestPages(TestCase):
''' Django notifications automated page tests '''
"""Django notifications automated page tests"""
def setUp(self):
self.message_count = 10
self.from_user = User.objects.create_user(username="from", password="pwd", email="example@example.com")
@ -196,284 +199,295 @@ class NotificationTestPages(TestCase):
self.to_user.is_staff = True
self.to_user.save()
for _ in range(self.message_count):
notify.send(self.from_user, recipient=self.to_user, verb='commented', action_object=self.from_user)
notify.send(self.from_user, recipient=self.to_user, verb="commented", action_object=self.from_user)
def logout(self):
self.client.post(reverse('admin:logout')+'?next=/', {})
self.client.post(reverse("admin:logout") + "?next=/", {})
def login(self, username, password):
self.logout()
response = self.client.post(reverse('login'), {'username': username, 'password': password})
response = self.client.post(reverse("login"), {"username": username, "password": password})
self.assertEqual(response.status_code, 302)
return response
def test_all_messages_page(self):
self.login('to', 'pwd')
response = self.client.get(reverse('notifications:all'))
self.login("to", "pwd")
response = self.client.get(reverse("notifications:all"))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.context['notifications']), len(self.to_user.notifications.all()))
self.assertEqual(len(response.context["notifications"]), len(self.to_user.notifications.all()))
def test_unread_messages_pages(self):
self.login('to', 'pwd')
response = self.client.get(reverse('notifications:unread'))
self.login("to", "pwd")
response = self.client.get(reverse("notifications:unread"))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.context['notifications']), len(self.to_user.notifications.unread()))
self.assertEqual(len(response.context['notifications']), self.message_count)
self.assertEqual(len(response.context["notifications"]), len(self.to_user.notifications.unread()))
self.assertEqual(len(response.context["notifications"]), self.message_count)
for index, notification in enumerate(self.to_user.notifications.all()):
if index % 3 == 0:
response = self.client.get(reverse('notifications:mark_as_read', args=[id2slug(notification.id)]))
response = self.client.get(reverse("notifications:mark_as_read", args=[id2slug(notification.id)]))
self.assertEqual(response.status_code, 302)
response = self.client.get(reverse('notifications:unread'))
response = self.client.get(reverse("notifications:unread"))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.context['notifications']), len(self.to_user.notifications.unread()))
self.assertTrue(len(response.context['notifications']) < self.message_count)
self.assertEqual(len(response.context["notifications"]), len(self.to_user.notifications.unread()))
self.assertTrue(len(response.context["notifications"]) < self.message_count)
response = self.client.get(reverse('notifications:mark_all_as_read'))
self.assertRedirects(response, reverse('notifications:unread'))
response = self.client.get(reverse('notifications:unread'))
self.assertEqual(len(response.context['notifications']), len(self.to_user.notifications.unread()))
self.assertEqual(len(response.context['notifications']), 0)
response = self.client.get(reverse("notifications:mark_all_as_read"))
self.assertRedirects(response, reverse("notifications:unread"))
response = self.client.get(reverse("notifications:unread"))
self.assertEqual(len(response.context["notifications"]), len(self.to_user.notifications.unread()))
self.assertEqual(len(response.context["notifications"]), 0)
def test_next_pages(self):
self.login('to', 'pwd')
query_parameters = '?var1=hello&var2=world'
self.login("to", "pwd")
query_parameters = "?var1=hello&var2=world"
response = self.client.get(reverse('notifications:mark_all_as_read'),data={
"next": reverse('notifications:unread') + query_parameters,
})
self.assertRedirects(response, reverse('notifications:unread') + query_parameters)
response = self.client.get(
reverse("notifications:mark_all_as_read"),
data={
"next": reverse("notifications:unread") + query_parameters,
},
)
self.assertRedirects(response, reverse("notifications:unread") + query_parameters)
slug = id2slug(self.to_user.notifications.first().id)
response = self.client.get(reverse('notifications:mark_as_read', args=[slug]), data={
"next": reverse('notifications:unread') + query_parameters,
})
self.assertRedirects(response, reverse('notifications:unread') + query_parameters)
response = self.client.get(
reverse("notifications:mark_as_read", args=[slug]),
data={
"next": reverse("notifications:unread") + query_parameters,
},
)
self.assertRedirects(response, reverse("notifications:unread") + query_parameters)
slug = id2slug(self.to_user.notifications.first().id)
response = self.client.get(reverse('notifications:mark_as_unread', args=[slug]), {
"next": reverse('notifications:unread') + query_parameters,
})
self.assertRedirects(response, reverse('notifications:unread') + query_parameters)
response = self.client.get(
reverse("notifications:mark_as_unread", args=[slug]),
{
"next": reverse("notifications:unread") + query_parameters,
},
)
self.assertRedirects(response, reverse("notifications:unread") + query_parameters)
@override_settings(ALLOWED_HOSTS=["www.notifications.com"])
def test_malicious_next_pages(self):
self.client.force_login(self.to_user)
query_parameters = '?var1=hello&var2=world'
query_parameters = "?var1=hello&var2=world"
for next_url in MALICIOUS_NEXT_URLS:
response = self.client.get(reverse('notifications:mark_all_as_read'),data={
"next": next_url + query_parameters,
})
self.assertRedirects(response, reverse('notifications:unread'))
response = self.client.get(
reverse("notifications:mark_all_as_read"),
data={
"next": next_url + query_parameters,
},
)
self.assertRedirects(response, reverse("notifications:unread"))
def test_delete_messages_pages(self):
self.login('to', 'pwd')
self.login("to", "pwd")
slug = id2slug(self.to_user.notifications.first().id)
response = self.client.get(reverse('notifications:delete', args=[slug]))
self.assertRedirects(response, reverse('notifications:all'))
response = self.client.get(reverse("notifications:delete", args=[slug]))
self.assertRedirects(response, reverse("notifications:all"))
response = self.client.get(reverse('notifications:all'))
response = self.client.get(reverse("notifications:all"))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.context['notifications']), len(self.to_user.notifications.all()))
self.assertEqual(len(response.context['notifications']), self.message_count-1)
self.assertEqual(len(response.context["notifications"]), len(self.to_user.notifications.all()))
self.assertEqual(len(response.context["notifications"]), self.message_count - 1)
response = self.client.get(reverse('notifications:unread'))
response = self.client.get(reverse("notifications:unread"))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.context['notifications']), len(self.to_user.notifications.unread()))
self.assertEqual(len(response.context['notifications']), self.message_count-1)
self.assertEqual(len(response.context["notifications"]), len(self.to_user.notifications.unread()))
self.assertEqual(len(response.context["notifications"]), self.message_count - 1)
@override_settings(DJANGO_NOTIFICATIONS_CONFIG={
'SOFT_DELETE': True
}) # pylint: disable=invalid-name
@override_settings(DJANGO_NOTIFICATIONS_CONFIG={"SOFT_DELETE": True}) # pylint: disable=invalid-name
def test_soft_delete_messages_manager(self):
self.login('to', 'pwd')
self.login("to", "pwd")
slug = id2slug(self.to_user.notifications.first().id)
response = self.client.get(reverse('notifications:delete', args=[slug]))
self.assertRedirects(response, reverse('notifications:all'))
response = self.client.get(reverse("notifications:delete", args=[slug]))
self.assertRedirects(response, reverse("notifications:all"))
response = self.client.get(reverse('notifications:all'))
response = self.client.get(reverse("notifications:all"))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.context['notifications']), len(self.to_user.notifications.active()))
self.assertEqual(len(response.context['notifications']), self.message_count-1)
self.assertEqual(len(response.context["notifications"]), len(self.to_user.notifications.active()))
self.assertEqual(len(response.context["notifications"]), self.message_count - 1)
response = self.client.get(reverse('notifications:unread'))
response = self.client.get(reverse("notifications:unread"))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.context['notifications']), len(self.to_user.notifications.unread()))
self.assertEqual(len(response.context['notifications']), self.message_count-1)
self.assertEqual(len(response.context["notifications"]), len(self.to_user.notifications.unread()))
self.assertEqual(len(response.context["notifications"]), self.message_count - 1)
def test_unread_count_api(self):
self.login('to', 'pwd')
self.login("to", "pwd")
response = self.client.get(reverse('notifications:live_unread_notification_count'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(list(data.keys()), ['unread_count'])
self.assertEqual(data['unread_count'], self.message_count)
response = self.client.get(reverse("notifications:live_unread_notification_count"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(list(data.keys()), ["unread_count"])
self.assertEqual(data["unread_count"], self.message_count)
Notification.objects.filter(recipient=self.to_user).mark_all_as_read()
response = self.client.get(reverse('notifications:live_unread_notification_count'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(list(data.keys()), ['unread_count'])
self.assertEqual(data['unread_count'], 0)
response = self.client.get(reverse("notifications:live_unread_notification_count"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(list(data.keys()), ["unread_count"])
self.assertEqual(data["unread_count"], 0)
notify.send(self.from_user, recipient=self.to_user, verb='commented', action_object=self.from_user)
response = self.client.get(reverse('notifications:live_unread_notification_count'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(list(data.keys()), ['unread_count'])
self.assertEqual(data['unread_count'], 1)
notify.send(self.from_user, recipient=self.to_user, verb="commented", action_object=self.from_user)
response = self.client.get(reverse("notifications:live_unread_notification_count"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(list(data.keys()), ["unread_count"])
self.assertEqual(data["unread_count"], 1)
def test_all_count_api(self):
self.login('to', 'pwd')
self.login("to", "pwd")
response = self.client.get(reverse('notifications:live_all_notification_count'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(list(data.keys()), ['all_count'])
self.assertEqual(data['all_count'], self.message_count)
response = self.client.get(reverse("notifications:live_all_notification_count"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(list(data.keys()), ["all_count"])
self.assertEqual(data["all_count"], self.message_count)
Notification.objects.filter(recipient=self.to_user).mark_all_as_read()
response = self.client.get(reverse('notifications:live_all_notification_count'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(list(data.keys()), ['all_count'])
self.assertEqual(data['all_count'], self.message_count)
response = self.client.get(reverse("notifications:live_all_notification_count"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(list(data.keys()), ["all_count"])
self.assertEqual(data["all_count"], self.message_count)
notify.send(self.from_user, recipient=self.to_user, verb='commented', action_object=self.from_user)
response = self.client.get(reverse('notifications:live_all_notification_count'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(list(data.keys()), ['all_count'])
self.assertEqual(data['all_count'], self.message_count + 1)
notify.send(self.from_user, recipient=self.to_user, verb="commented", action_object=self.from_user)
response = self.client.get(reverse("notifications:live_all_notification_count"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(list(data.keys()), ["all_count"])
self.assertEqual(data["all_count"], self.message_count + 1)
def test_unread_list_api(self):
self.login('to', 'pwd')
self.login("to", "pwd")
response = self.client.get(reverse('notifications:live_unread_notification_list'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(sorted(list(data.keys())), ['unread_count', 'unread_list'])
self.assertEqual(data['unread_count'], self.message_count)
self.assertEqual(len(data['unread_list']), self.message_count)
response = self.client.get(reverse("notifications:live_unread_notification_list"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(sorted(list(data.keys())), ["unread_count", "unread_list"])
self.assertEqual(data["unread_count"], self.message_count)
self.assertEqual(len(data["unread_list"]), self.message_count)
response = self.client.get(reverse('notifications:live_unread_notification_list'), data={"max": 5})
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(sorted(list(data.keys())), ['unread_count', 'unread_list'])
self.assertEqual(data['unread_count'], self.message_count)
self.assertEqual(len(data['unread_list']), 5)
response = self.client.get(reverse("notifications:live_unread_notification_list"), data={"max": 5})
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(sorted(list(data.keys())), ["unread_count", "unread_list"])
self.assertEqual(data["unread_count"], self.message_count)
self.assertEqual(len(data["unread_list"]), 5)
# Test with a bad 'max' value
response = self.client.get(reverse('notifications:live_unread_notification_list'), data={
"max": "this_is_wrong",
})
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(sorted(list(data.keys())), ['unread_count', 'unread_list'])
self.assertEqual(data['unread_count'], self.message_count)
self.assertEqual(len(data['unread_list']), self.message_count)
response = self.client.get(
reverse("notifications:live_unread_notification_list"),
data={
"max": "this_is_wrong",
},
)
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(sorted(list(data.keys())), ["unread_count", "unread_list"])
self.assertEqual(data["unread_count"], self.message_count)
self.assertEqual(len(data["unread_list"]), self.message_count)
Notification.objects.filter(recipient=self.to_user).mark_all_as_read()
response = self.client.get(reverse('notifications:live_unread_notification_list'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(sorted(list(data.keys())), ['unread_count', 'unread_list'])
self.assertEqual(data['unread_count'], 0)
self.assertEqual(len(data['unread_list']), 0)
response = self.client.get(reverse("notifications:live_unread_notification_list"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(sorted(list(data.keys())), ["unread_count", "unread_list"])
self.assertEqual(data["unread_count"], 0)
self.assertEqual(len(data["unread_list"]), 0)
notify.send(self.from_user, recipient=self.to_user, verb='commented', action_object=self.from_user)
response = self.client.get(reverse('notifications:live_unread_notification_list'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(sorted(list(data.keys())), ['unread_count', 'unread_list'])
self.assertEqual(data['unread_count'], 1)
self.assertEqual(len(data['unread_list']), 1)
self.assertEqual(data['unread_list'][0]['verb'], 'commented')
self.assertEqual(data['unread_list'][0]['slug'], id2slug(data['unread_list'][0]['id']))
notify.send(self.from_user, recipient=self.to_user, verb="commented", action_object=self.from_user)
response = self.client.get(reverse("notifications:live_unread_notification_list"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(sorted(list(data.keys())), ["unread_count", "unread_list"])
self.assertEqual(data["unread_count"], 1)
self.assertEqual(len(data["unread_list"]), 1)
self.assertEqual(data["unread_list"][0]["verb"], "commented")
self.assertEqual(data["unread_list"][0]["slug"], id2slug(data["unread_list"][0]["id"]))
def test_all_list_api(self):
self.login('to', 'pwd')
self.login("to", "pwd")
response = self.client.get(reverse('notifications:live_all_notification_list'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(sorted(list(data.keys())), ['all_count', 'all_list'])
self.assertEqual(data['all_count'], self.message_count)
self.assertEqual(len(data['all_list']), self.message_count)
response = self.client.get(reverse("notifications:live_all_notification_list"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(sorted(list(data.keys())), ["all_count", "all_list"])
self.assertEqual(data["all_count"], self.message_count)
self.assertEqual(len(data["all_list"]), self.message_count)
response = self.client.get(reverse('notifications:live_all_notification_list'), data={"max": 5})
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(sorted(list(data.keys())), ['all_count', 'all_list'])
self.assertEqual(data['all_count'], self.message_count)
self.assertEqual(len(data['all_list']), 5)
response = self.client.get(reverse("notifications:live_all_notification_list"), data={"max": 5})
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(sorted(list(data.keys())), ["all_count", "all_list"])
self.assertEqual(data["all_count"], self.message_count)
self.assertEqual(len(data["all_list"]), 5)
# Test with a bad 'max' value
response = self.client.get(reverse('notifications:live_all_notification_list'), data={
"max": "this_is_wrong",
})
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(sorted(list(data.keys())), ['all_count', 'all_list'])
self.assertEqual(data['all_count'], self.message_count)
self.assertEqual(len(data['all_list']), self.message_count)
response = self.client.get(
reverse("notifications:live_all_notification_list"),
data={
"max": "this_is_wrong",
},
)
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(sorted(list(data.keys())), ["all_count", "all_list"])
self.assertEqual(data["all_count"], self.message_count)
self.assertEqual(len(data["all_list"]), self.message_count)
Notification.objects.filter(recipient=self.to_user).mark_all_as_read()
response = self.client.get(reverse('notifications:live_all_notification_list'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(sorted(list(data.keys())), ['all_count', 'all_list'])
self.assertEqual(data['all_count'], self.message_count)
self.assertEqual(len(data['all_list']), self.message_count)
response = self.client.get(reverse("notifications:live_all_notification_list"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(sorted(list(data.keys())), ["all_count", "all_list"])
self.assertEqual(data["all_count"], self.message_count)
self.assertEqual(len(data["all_list"]), self.message_count)
notify.send(self.from_user, recipient=self.to_user, verb='commented', action_object=self.from_user)
response = self.client.get(reverse('notifications:live_all_notification_list'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(sorted(list(data.keys())), ['all_count', 'all_list'])
self.assertEqual(data['all_count'], self.message_count + 1)
self.assertEqual(len(data['all_list']), self.message_count)
self.assertEqual(data['all_list'][0]['verb'], 'commented')
self.assertEqual(data['all_list'][0]['slug'], id2slug(data['all_list'][0]['id']))
notify.send(self.from_user, recipient=self.to_user, verb="commented", action_object=self.from_user)
response = self.client.get(reverse("notifications:live_all_notification_list"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(sorted(list(data.keys())), ["all_count", "all_list"])
self.assertEqual(data["all_count"], self.message_count + 1)
self.assertEqual(len(data["all_list"]), self.message_count)
self.assertEqual(data["all_list"][0]["verb"], "commented")
self.assertEqual(data["all_list"][0]["slug"], id2slug(data["all_list"][0]["id"]))
def test_unread_list_api_mark_as_read(self): # pylint: disable=invalid-name
self.login('to', 'pwd')
self.login("to", "pwd")
num_requested = 3
response = self.client.get(
reverse('notifications:live_unread_notification_list'),
data={"max": num_requested, "mark_as_read": 1}
reverse("notifications:live_unread_notification_list"), data={"max": num_requested, "mark_as_read": 1}
)
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(data['unread_count'],
self.message_count - num_requested)
self.assertEqual(len(data['unread_list']), num_requested)
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(data["unread_count"], self.message_count - num_requested)
self.assertEqual(len(data["unread_list"]), num_requested)
response = self.client.get(
reverse('notifications:live_unread_notification_list'),
data={"max": num_requested, "mark_as_read": 1}
reverse("notifications:live_unread_notification_list"), data={"max": num_requested, "mark_as_read": 1}
)
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(data['unread_count'],
self.message_count - 2*num_requested)
self.assertEqual(len(data['unread_list']), num_requested)
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(data["unread_count"], self.message_count - 2 * num_requested)
self.assertEqual(len(data["unread_list"]), num_requested)
def test_live_update_tags(self):
from django.shortcuts import render
self.login('to', 'pwd')
self.login("to", "pwd")
factory = RequestFactory()
request = factory.get('/notification/live_updater')
request = factory.get("/notification/live_updater")
request.user = self.to_user
render(request, 'notifications/test_tags.html', {'request': request, 'nonce': 'nonce-T5esDNXMnDe5lKMQ6ZzTUw=='})
render(request, "notifications/test_tags.html", {"request": request, "nonce": "nonce-T5esDNXMnDe5lKMQ6ZzTUw=="})
# TODO: Add more tests to check what is being output.
def test_anon_user_gets_nothing(self):
response = self.client.post(reverse('notifications:live_unread_notification_count'))
response = self.client.post(reverse("notifications:live_unread_notification_count"))
self.assertEqual(response.status_code, 200)
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(data['unread_count'], 0)
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(data["unread_count"], 0)
response = self.client.post(reverse('notifications:live_unread_notification_list'))
response = self.client.post(reverse("notifications:live_unread_notification_list"))
self.assertEqual(response.status_code, 200)
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(data['unread_count'], 0)
self.assertEqual(data['unread_list'], [])
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(data["unread_count"], 0)
self.assertEqual(data["unread_list"], [])
class NotificationTestExtraData(TestCase):
''' Django notifications automated extra data tests '''
"""Django notifications automated extra data tests"""
def setUp(self):
self.message_count = 1
self.from_user = User.objects.create_user(username="from", password="pwd", email="example@example.com")
@ -484,31 +498,32 @@ class NotificationTestExtraData(TestCase):
notify.send(
self.from_user,
recipient=self.to_user,
verb='commented',
verb="commented",
action_object=self.from_user,
url="/learn/ask-a-pro/q/test-question-9/299/",
other_content="Hello my 'world'"
other_content="Hello my 'world'",
)
def logout(self):
self.client.post(reverse('admin:logout')+'?next=/', {})
self.client.post(reverse("admin:logout") + "?next=/", {})
def login(self, username, password):
self.logout()
response = self.client.post(reverse('login'), {'username': username, 'password': password})
response = self.client.post(reverse("login"), {"username": username, "password": password})
self.assertEqual(response.status_code, 302)
return response
def test_extra_data(self):
self.login('to', 'pwd')
response = self.client.post(reverse('notifications:live_unread_notification_list'))
data = json.loads(response.content.decode('utf-8'))
self.assertEqual(data['unread_list'][0]['data']['url'], "/learn/ask-a-pro/q/test-question-9/299/")
self.assertEqual(data['unread_list'][0]['data']['other_content'], "Hello my 'world'")
self.login("to", "pwd")
response = self.client.post(reverse("notifications:live_unread_notification_list"))
data = json.loads(response.content.decode("utf-8"))
self.assertEqual(data["unread_list"][0]["data"]["url"], "/learn/ask-a-pro/q/test-question-9/299/")
self.assertEqual(data["unread_list"][0]["data"]["other_content"], "Hello my 'world'")
class TagTest(TestCase):
''' Django notifications automated tags tests '''
"""Django notifications automated tags tests"""
def setUp(self):
self.message_count = 1
self.from_user = User.objects.create_user(username="from", password="pwd", email="example@example.com")
@ -519,26 +534,27 @@ class TagTest(TestCase):
notify.send(
self.from_user,
recipient=self.to_user,
verb='commented',
verb="commented",
action_object=self.from_user,
url="/learn/ask-a-pro/q/test-question-9/299/",
other_content="Hello my 'world'"
other_content="Hello my 'world'",
)
def tag_test(self, template, context, output):
t = Template('{% load notifications_tags %}'+template)
c = Context(context)
self.assertEqual(t.render(c), output)
template = Template("{% load notifications_tags %}" + template)
context = Context(context)
self.assertEqual(template.render(context), output)
def test_has_notification(self):
template = "{{ user|has_notification }}"
context = {"user":self.to_user}
output = u"True"
context = {"user": self.to_user}
output = "True"
self.tag_test(template, context, output)
class AdminTest(TestCase):
app_name = "notifications"
def setUp(self):
self.message_count = 10
self.from_user = User.objects.create_user(username="from", password="pwd", email="example@example.com")
@ -550,15 +566,15 @@ class AdminTest(TestCase):
notify.send(
self.from_user,
recipient=self.to_user,
verb='commented',
verb="commented",
action_object=self.from_user,
)
def test_list(self):
self.client.login(username='to', password='pwd')
self.client.login(username="to", password="pwd")
with CaptureQueriesContext(connection=connection) as context:
response = self.client.get(reverse('admin:{0}_notification_changelist'.format(self.app_name)))
response = self.client.get(reverse(f"admin:{self.app_name}_notification_changelist"))
self.assertLessEqual(len(context), 6)
self.assertEqual(response.status_code, 200, response.content)

16
notifications/tests/urls.py
View file

@ -1,15 +1,15 @@
''' Django notification urls for tests '''
""" Django notification urls for tests """
# -*- coding: utf-8 -*-
from django.contrib import admin
from notifications.tests.views import live_tester, make_notification
from django.urls import include, path
from django.contrib.auth.views import LoginView
from django.urls import include, path
from notifications.tests.views import live_tester, make_notification
urlpatterns = [
path('test_make/', make_notification),
path('test/', live_tester),
path('login/', LoginView.as_view(), name='login'), # reverse for django login is not working
path('admin/', admin.site.urls),
path('', include('notifications.urls', namespace='notifications')),
path("test_make/", make_notification),
path("test/", live_tester),
path("login/", LoginView.as_view(), name="login"), # reverse for django login is not working
path("admin/", admin.site.urls),
path("", include("notifications.urls", namespace="notifications")),
]

39
notifications/tests/views.py
View file

@ -1,34 +1,41 @@
''' Django notifications views for tests '''
""" Django notifications views for tests """
# -*- coding: utf-8 -*-
import random
from django.contrib.auth.decorators import login_required
from django.http import HttpResponse
from django.shortcuts import render
from notifications.signals import notify
@login_required
def live_tester(request):
notify.send(sender=request.user, recipient=request.user, verb='you loaded the page')
notify.send(sender=request.user, recipient=request.user, verb="you loaded the page")
return render(request, 'test_live.html', {
'unread_count': request.user.notifications.unread().count(),
'notifications': request.user.notifications.all()
})
return render(
request,
"test_live.html",
{
"unread_count": request.user.notifications.unread().count(),
"notifications": request.user.notifications.all(),
},
)
def make_notification(request):
the_notification = random.choice(
[
"reticulating splines",
"cleaning the car",
"jumping the shark",
"testing the app",
"attaching the plumbus",
]
)
the_notification = random.choice([
'reticulating splines',
'cleaning the car',
'jumping the shark',
'testing the app',
'attaching the plumbus',
])
notify.send(sender=request.user, recipient=request.user,
verb='you asked for a notification - you are ' + the_notification)
notify.send(
sender=request.user, recipient=request.user, verb="you asked for a notification - you are " + the_notification
)
return HttpResponse()

25
notifications/urls.py
View file

@ -1,20 +1,19 @@
''' Django notification urls file '''
""" Django notification urls file """
# -*- coding: utf-8 -*-
from django.urls import path
from . import views
app_name = 'notifications'
app_name = "notifications"
urlpatterns = [
path('', views.AllNotificationsList.as_view(), name='all'),
path('unread/', views.UnreadNotificationsList.as_view(), name='unread'),
path('mark-all-as-read/', views.mark_all_as_read, name='mark_all_as_read'),
path('mark-as-read/<int:slug>)/', views.mark_as_read, name='mark_as_read'),
path('mark-as-unread/<int:slug>/', views.mark_as_unread, name='mark_as_unread'),
path('delete/<int:slug>/', views.delete, name='delete'),
path('api/unread_count/', views.live_unread_notification_count, name='live_unread_notification_count'),
path('api/all_count/', views.live_all_notification_count, name='live_all_notification_count'),
path('api/unread_list/', views.live_unread_notification_list, name='live_unread_notification_list'),
path('api/all_list/', views.live_all_notification_list, name='live_all_notification_list'),
path("", views.AllNotificationsList.as_view(), name="all"),
path("unread/", views.UnreadNotificationsList.as_view(), name="unread"),
path("mark-all-as-read/", views.mark_all_as_read, name="mark_all_as_read"),
path("mark-as-read/<int:slug>)/", views.mark_as_read, name="mark_as_read"),
path("mark-as-unread/<int:slug>/", views.mark_as_unread, name="mark_as_unread"),
path("delete/<int:slug>/", views.delete, name="delete"),
path("api/unread_count/", views.live_unread_notification_count, name="live_unread_notification_count"),
path("api/all_count/", views.live_all_notification_count, name="live_all_notification_count"),
path("api/unread_list/", views.live_unread_notification_list, name="live_unread_notification_list"),
path("api/all_list/", views.live_all_notification_list, name="live_all_notification_list"),
]

3
notifications/utils.py
View file

@ -1,6 +1,5 @@
'''' Django notifications utils file '''
"""' Django notifications utils file """
# -*- coding: utf-8 -*-
import sys
def slug2id(slug):

81
notifications/views.py
View file

@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
''' Django Notifications example views '''
""" Django Notifications example views """
from django.conf import settings
from django.contrib.auth.decorators import login_required
from django.http import JsonResponse
@ -15,18 +15,17 @@ from notifications import settings as notification_settings
from notifications.helpers import get_notification_list
from notifications.utils import slug2id
Notification = load_model('notifications', 'Notification')
Notification = load_model("notifications", "Notification")
class NotificationViewList(ListView):
template_name = 'notifications/list.html'
context_object_name = 'notifications'
paginate_by = notification_settings.get_config()['PAGINATE_BY']
template_name = "notifications/list.html"
context_object_name = "notifications"
paginate_by = notification_settings.get_config()["PAGINATE_BY"]
@method_decorator(login_required)
def dispatch(self, request, *args, **kwargs):
return super(NotificationViewList, self).dispatch(
request, *args, **kwargs)
return super().dispatch(request, *args, **kwargs)
class AllNotificationsList(NotificationViewList):
@ -35,7 +34,7 @@ class AllNotificationsList(NotificationViewList):
"""
def get_queryset(self):
if notification_settings.get_config()['SOFT_DELETE']:
if notification_settings.get_config()["SOFT_DELETE"]:
qset = self.request.user.notifications.active()
else:
qset = self.request.user.notifications.all()
@ -43,7 +42,6 @@ class AllNotificationsList(NotificationViewList):
class UnreadNotificationsList(NotificationViewList):
def get_queryset(self):
return self.request.user.notifications.unread()
@ -52,124 +50,105 @@ class UnreadNotificationsList(NotificationViewList):
def mark_all_as_read(request):
request.user.notifications.mark_all_as_read()
_next = request.GET.get('next')
_next = request.GET.get("next")
if _next and url_has_allowed_host_and_scheme(_next, settings.ALLOWED_HOSTS):
return redirect(iri_to_uri(_next))
return redirect('notifications:unread')
return redirect("notifications:unread")
@login_required
def mark_as_read(request, slug=None):
notification_id = slug2id(slug)
notification = get_object_or_404(
Notification, recipient=request.user, id=notification_id)
notification = get_object_or_404(Notification, recipient=request.user, id=notification_id)
notification.mark_as_read()
_next = request.GET.get('next')
_next = request.GET.get("next")
if _next and url_has_allowed_host_and_scheme(_next, settings.ALLOWED_HOSTS):
return redirect(iri_to_uri(_next))
return redirect('notifications:unread')
return redirect("notifications:unread")
@login_required
def mark_as_unread(request, slug=None):
notification_id = slug2id(slug)
notification = get_object_or_404(
Notification, recipient=request.user, id=notification_id)
notification = get_object_or_404(Notification, recipient=request.user, id=notification_id)
notification.mark_as_unread()
_next = request.GET.get('next')
_next = request.GET.get("next")
if _next and url_has_allowed_host_and_scheme(_next, settings.ALLOWED_HOSTS):
return redirect(iri_to_uri(_next))
return redirect('notifications:unread')
return redirect("notifications:unread")
@login_required
def delete(request, slug=None):
notification_id = slug2id(slug)
notification = get_object_or_404(
Notification, recipient=request.user, id=notification_id)
notification = get_object_or_404(Notification, recipient=request.user, id=notification_id)
if notification_settings.get_config()['SOFT_DELETE']:
if notification_settings.get_config()["SOFT_DELETE"]:
notification.deleted = True
notification.save()
else:
notification.delete()
_next = request.GET.get('next')
_next = request.GET.get("next")
if _next and url_has_allowed_host_and_scheme(_next, settings.ALLOWED_HOSTS):
return redirect(iri_to_uri(_next))
return redirect('notifications:all')
return redirect("notifications:all")
@never_cache
def live_unread_notification_count(request):
if not request.user.is_authenticated:
data = {
'unread_count': 0
}
data = {"unread_count": 0}
else:
data = {
'unread_count': request.user.notifications.unread().count(),
"unread_count": request.user.notifications.unread().count(),
}
return JsonResponse(data)
@never_cache
def live_unread_notification_list(request):
''' Return a json with a unread notification list '''
"""Return a json with a unread notification list"""
if not request.user.is_authenticated:
data = {
'unread_count': 0,
'unread_list': []
}
data = {"unread_count": 0, "unread_list": []}
return JsonResponse(data)
unread_list = get_notification_list(request, 'unread')
unread_list = get_notification_list(request, "unread")
data = {
'unread_count': request.user.notifications.unread().count(),
'unread_list': unread_list
}
data = {"unread_count": request.user.notifications.unread().count(), "unread_list": unread_list}
return JsonResponse(data)
@never_cache
def live_all_notification_list(request):
''' Return a json with a unread notification list '''
"""Return a json with a unread notification list"""
if not request.user.is_authenticated:
data = {
'all_count': 0,
'all_list': []
}
data = {"all_count": 0, "all_list": []}
return JsonResponse(data)
all_list = get_notification_list(request)
data = {
'all_count': request.user.notifications.count(),
'all_list': all_list
}
data = {"all_count": request.user.notifications.count(), "all_list": all_list}
return JsonResponse(data)
def live_all_notification_count(request):
if not request.user.is_authenticated:
data = {
'all_count': 0
}
data = {"all_count": 0}
else:
data = {
'all_count': request.user.notifications.count(),
"all_count": request.user.notifications.count(),
}
return JsonResponse(data)

5
pyproject.toml
View file

@ -115,7 +115,9 @@ load-plugins = [
"pylint_django",
"pylint.extensions.mccabe",
]
ignore = "migrations/"
ignore = [
"migrations",
]
jobs = 0
django-settings-module = "notifications.settings"
@ -129,7 +131,6 @@ disable = [
"missing-function-docstring",
"missing-class-docstring",
"missing-module-docstring",
# "invalid-name",
"fixme",
]