mirror of
https://github.com/Hopiu/django-notifications.git
synced 2026-03-16 21:30:24 +00:00
Refactoring the NotificationQuerySet and its tests
This commit is contained in:
Alvaro Leonel
parent
78faf1abaa
commit
888c59eff3
3 changed files with 133 additions and 293 deletions
|
|
@ -1,5 +1,7 @@
|
|||
from typing import NewType
|
||||
from typing import NewType, Type, Union
|
||||
|
||||
from django.contrib.auth.base_user import AbstractBaseUser
|
||||
|
||||
AbstractUser = NewType("AbstractUser", AbstractBaseUser) # type: ignore[valid-newtype]
|
||||
|
||||
OptionalAbstractUser = Union[None, Type[AbstractUser]]
|
||||
|
|
|
|||
|
|
@ -1,122 +1,91 @@
|
|||
from typing import Type, Union
|
||||
|
||||
from django.core.exceptions import ImproperlyConfigured
|
||||
from django.db import models
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
from notifications.notification_types import AbstractUser
|
||||
from notifications.helpers import assert_soft_delete
|
||||
from notifications.notification_types import OptionalAbstractUser
|
||||
from notifications.settings import notification_settings
|
||||
|
||||
|
||||
def assert_soft_delete() -> None:
|
||||
if not notification_settings.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"
|
||||
raise ImproperlyConfigured(msg)
|
||||
|
||||
|
||||
class NotificationQuerySet(models.QuerySet):
|
||||
"""Notification QuerySet"""
|
||||
|
||||
def unread(self, include_deleted: bool = False) -> "NotificationQuerySet":
|
||||
"""Return only unread items in the current queryset"""
|
||||
def _filter_by(self, include_deleted: bool = False, **kwargs) -> "NotificationQuerySet":
|
||||
if notification_settings.SOFT_DELETE and not include_deleted:
|
||||
return self.filter(unread=True, deleted=False)
|
||||
return self.filter(deleted=False, **kwargs)
|
||||
|
||||
# When SOFT_DELETE=False, developers are supposed NOT to touch 'deleted' field
|
||||
# in this case, to improve query performance, don't filter by 'deleted' field
|
||||
return self.filter(unread=True)
|
||||
|
||||
def read(self, include_deleted: bool = False) -> "NotificationQuerySet":
|
||||
"""Return only read items in the current queryset"""
|
||||
if notification_settings.SOFT_DELETE and not include_deleted:
|
||||
return self.filter(unread=False, deleted=False)
|
||||
|
||||
# When SOFT_DELETE=False, developers are supposed NOT to touch 'deleted' field
|
||||
# in this case, to improve query performance, don't filter by 'deleted' field
|
||||
return self.filter(unread=False)
|
||||
|
||||
def mark_all_as_read(self, recipient: Union[None, Type[AbstractUser]] = None) -> int:
|
||||
"""Mark as read any unread messages in the current queryset.
|
||||
|
||||
Optionally, filter these by recipient first.
|
||||
"""
|
||||
# We want to filter out read ones, as later we will store
|
||||
# the time they were marked as read.
|
||||
qset = self.unread(True)
|
||||
if recipient:
|
||||
qset = qset.filter(recipient=recipient)
|
||||
|
||||
return qset.update(unread=False)
|
||||
|
||||
def mark_all_as_unread(self, recipient: Union[None, Type[AbstractUser]] = None) -> int:
|
||||
"""Mark as unread any read messages in the current queryset.
|
||||
|
||||
Optionally, filter these by recipient first.
|
||||
"""
|
||||
qset = self.read(True)
|
||||
|
||||
if recipient:
|
||||
qset = qset.filter(recipient=recipient)
|
||||
|
||||
return qset.update(unread=True)
|
||||
|
||||
def deleted(self) -> "NotificationQuerySet":
|
||||
"""Return only deleted items in the current queryset"""
|
||||
assert_soft_delete()
|
||||
return self.filter(deleted=True)
|
||||
return self.filter(**kwargs)
|
||||
|
||||
def active(self) -> "NotificationQuerySet":
|
||||
"""Return only active(un-deleted) items in the current queryset"""
|
||||
"""If SOFT_DELETE is active, return only active items in the current queryset"""
|
||||
assert_soft_delete()
|
||||
return self.filter(deleted=False)
|
||||
|
||||
def mark_all_as_deleted(self, recipient: Union[None, Type[AbstractUser]] = None) -> int:
|
||||
"""Mark current queryset as deleted.
|
||||
Optionally, filter by recipient first.
|
||||
"""
|
||||
def deleted(self) -> "NotificationQuerySet":
|
||||
"""If SOFT_DELETE is active, return only deleted items in the current queryset"""
|
||||
assert_soft_delete()
|
||||
qset = self.active()
|
||||
if recipient:
|
||||
qset = qset.filter(recipient=recipient)
|
||||
|
||||
return qset.update(deleted=True)
|
||||
|
||||
def mark_all_as_active(self, recipient: Union[None, Type[AbstractUser]] = None) -> int:
|
||||
"""Mark current queryset as active(un-deleted).
|
||||
Optionally, filter by recipient first.
|
||||
"""
|
||||
assert_soft_delete()
|
||||
qset = self.deleted()
|
||||
if recipient:
|
||||
qset = qset.filter(recipient=recipient)
|
||||
|
||||
return qset.update(deleted=False)
|
||||
|
||||
def unsent(self, include_deleted: bool = False) -> "NotificationQuerySet":
|
||||
"""Return only unsent items in the current queryset"""
|
||||
if notification_settings.SOFT_DELETE and not include_deleted:
|
||||
return self.filter(emailed=False, deleted=False)
|
||||
|
||||
return self.filter(emailed=False)
|
||||
return self.filter(deleted=True)
|
||||
|
||||
def sent(self, include_deleted: bool = False) -> "NotificationQuerySet":
|
||||
"""Return only sent items in the current queryset"""
|
||||
if notification_settings.SOFT_DELETE and not include_deleted:
|
||||
return self.filter(emailed=True, deleted=False)
|
||||
return self._filter_by(include_deleted=include_deleted, emailed=True)
|
||||
|
||||
return self.filter(emailed=True)
|
||||
def unsent(self, include_deleted: bool = False) -> "NotificationQuerySet":
|
||||
"""Return only unsent items in the current queryset"""
|
||||
return self._filter_by(include_deleted=include_deleted, emailed=False)
|
||||
|
||||
def mark_as_unsent(self, recipient: Union[None, Type[AbstractUser]] = None) -> int:
|
||||
qset = self.sent()
|
||||
def public(self, include_deleted: bool = False) -> "NotificationQuerySet":
|
||||
"""Return only public items in the current queryset"""
|
||||
return self._filter_by(include_deleted=include_deleted, public=True)
|
||||
|
||||
def private(self, include_deleted: bool = False) -> "NotificationQuerySet":
|
||||
"""Return only private items in the current queryset"""
|
||||
return self._filter_by(include_deleted=include_deleted, public=False)
|
||||
|
||||
def read(self, include_deleted: bool = False) -> "NotificationQuerySet":
|
||||
"""Return only read items in the current queryset"""
|
||||
return self._filter_by(include_deleted=include_deleted, unread=False)
|
||||
|
||||
def unread(self, include_deleted: bool = False) -> "NotificationQuerySet":
|
||||
"""Return only unread items in the current queryset"""
|
||||
return self._filter_by(include_deleted=include_deleted, unread=True)
|
||||
|
||||
def _mark_all_as(self, recipient: OptionalAbstractUser = None, **kwargs) -> int:
|
||||
if recipient:
|
||||
qset = qset.filter(recipient=recipient)
|
||||
return qset.update(emailed=False)
|
||||
return self.filter(recipient=recipient).update(**kwargs)
|
||||
return self.update(**kwargs)
|
||||
|
||||
def mark_as_sent(self, recipient: Union[None, Type[AbstractUser]] = None) -> int:
|
||||
qset = self.unsent()
|
||||
if recipient:
|
||||
qset = qset.filter(recipient=recipient)
|
||||
return qset.update(emailed=True)
|
||||
def mark_all_as_active(self, recipient: OptionalAbstractUser = None) -> int:
|
||||
"""If SOFT_DELETE is activated, mark all deleted notifications as active."""
|
||||
assert_soft_delete()
|
||||
return self.deleted()._mark_all_as(recipient=recipient, deleted=False) # pylint: disable=protected-access
|
||||
|
||||
def mark_all_as_deleted(self, recipient: OptionalAbstractUser = None) -> int:
|
||||
"""If SOFT_DELETE is activated, mark all active notifications as deleted."""
|
||||
assert_soft_delete()
|
||||
return self.active()._mark_all_as(recipient=recipient, deleted=True) # pylint: disable=protected-access
|
||||
|
||||
def mark_all_as_sent(self, recipient: OptionalAbstractUser = None) -> int:
|
||||
"""Mark all unsent notifications as sent."""
|
||||
return self.unsent()._mark_all_as(recipient=recipient, emailed=True) # pylint: disable=protected-access
|
||||
|
||||
def mark_all_as_unsent(self, recipient: OptionalAbstractUser = None) -> int:
|
||||
"""Mark all sent notifications as unsent."""
|
||||
return self.sent()._mark_all_as(recipient=recipient, emailed=False) # pylint: disable=protected-access
|
||||
|
||||
def mark_all_as_public(self, recipient: OptionalAbstractUser = None) -> int:
|
||||
"""Mark all private notifications as public."""
|
||||
return self.private()._mark_all_as(recipient=recipient, public=True) # pylint: disable=protected-access
|
||||
|
||||
def mark_all_as_private(self, recipient: OptionalAbstractUser = None) -> int:
|
||||
"""Mark all public notifications as private."""
|
||||
return self.public()._mark_all_as(recipient=recipient, public=False) # pylint: disable=protected-access
|
||||
|
||||
def mark_all_as_read(self, recipient: OptionalAbstractUser = None) -> int:
|
||||
"""Mark all unread notifications as read."""
|
||||
return self.unread()._mark_all_as(recipient=recipient, unread=False) # pylint: disable=protected-access
|
||||
|
||||
def mark_all_as_unread(self, recipient: OptionalAbstractUser = None) -> int:
|
||||
"""Mark all read notifications as unread."""
|
||||
return self.read()._mark_all_as(recipient=recipient, unread=True) # pylint: disable=protected-access
|
||||
|
|
|
|||
|
|
@ -12,232 +12,101 @@ User = get_user_model()
|
|||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"emailed,method",
|
||||
"method,field,initial_status,expected,expect_deleted",
|
||||
(
|
||||
(True, Notification.objects.sent),
|
||||
(False, Notification.objects.unsent),
|
||||
("sent", "emailed", True, 4, 2),
|
||||
("sent", "emailed", False, 0, 0),
|
||||
("unsent", "emailed", True, 0, 0),
|
||||
("unsent", "emailed", False, 4, 2),
|
||||
("public", "public", True, 4, 2),
|
||||
("private", "public", True, 0, 0),
|
||||
("read", "unread", False, 4, 2),
|
||||
("read", "unread", True, 0, 0),
|
||||
("unread", "unread", False, 0, 0),
|
||||
("unread", "unread", True, 4, 2),
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_sent_unsent_methods(emailed, method):
|
||||
NotificationFullFactory.create_batch(3, emailed=emailed)
|
||||
assert method().count() == 3
|
||||
def test_filters(method, field, initial_status, expected, expect_deleted):
|
||||
NotificationFullFactory.create_batch(2, **{field: initial_status})
|
||||
NotificationFullFactory.create_batch(2, deleted=True, **{field: initial_status})
|
||||
func = getattr(Notification.objects, method)
|
||||
assert func().count() == expected
|
||||
|
||||
first_notification = Notification.objects.first()
|
||||
first_notification.emailed = not emailed
|
||||
first_notification.save()
|
||||
assert method().count() == 2
|
||||
|
||||
Notification.objects.all().update(emailed=not emailed)
|
||||
assert method().count() == 0
|
||||
|
||||
first_notification.emailed = emailed
|
||||
first_notification.save()
|
||||
assert method().count() == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"read,method",
|
||||
(
|
||||
(False, Notification.objects.read),
|
||||
(True, Notification.objects.unread),
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_read_unread_methods(read, method):
|
||||
NotificationFullFactory.create_batch(3, unread=read)
|
||||
assert method().count() == 3
|
||||
|
||||
first_notification = Notification.objects.first()
|
||||
first_notification.unread = not read
|
||||
first_notification.save()
|
||||
assert method().count() == 2
|
||||
|
||||
Notification.objects.all().update(unread=not read)
|
||||
assert method().count() == 0
|
||||
|
||||
first_notification.unread = read
|
||||
first_notification.save()
|
||||
assert method().count() == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"read,method",
|
||||
(
|
||||
(False, Notification.objects.read),
|
||||
(True, Notification.objects.unread),
|
||||
),
|
||||
)
|
||||
@override_settings(DJANGO_NOTIFICATIONS_CONFIG={"SOFT_DELETE": True})
|
||||
@pytest.mark.django_db
|
||||
def test_read_unread_with_deleted_notifications(read, method):
|
||||
NotificationFullFactory.create_batch(3, unread=read)
|
||||
assert method().count() == 3
|
||||
|
||||
first_notification = Notification.objects.first()
|
||||
first_notification.deleted = True
|
||||
first_notification.save()
|
||||
|
||||
assert method().count() == 2
|
||||
assert method(include_deleted=True).count() == 3
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"status,method,check_method",
|
||||
(
|
||||
(True, Notification.objects.mark_all_as_read, Notification.objects.read),
|
||||
(False, Notification.objects.mark_all_as_unread, Notification.objects.unread),
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_mark_all_as_read_unread(status, method, check_method):
|
||||
NotificationFullFactory.create_batch(3, unread=status)
|
||||
assert check_method().count() == 0
|
||||
|
||||
method()
|
||||
assert check_method().count() == 3
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"status,method,check_method",
|
||||
(
|
||||
(True, Notification.objects.mark_all_as_read, Notification.objects.read),
|
||||
(False, Notification.objects.mark_all_as_unread, Notification.objects.unread),
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_mark_all_as_read_unread_with_recipient(status, method, check_method):
|
||||
recipient = RecipientFactory()
|
||||
NotificationFullFactory.create_batch(2, unread=status, recipient=recipient)
|
||||
NotificationFullFactory.create_batch(1, unread=status)
|
||||
assert Notification.objects.count() == 3
|
||||
assert check_method().count() == 0
|
||||
|
||||
method(recipient=recipient)
|
||||
assert check_method().count() == 2
|
||||
|
||||
|
||||
@override_settings(DJANGO_NOTIFICATIONS_CONFIG={"SOFT_DELETE": True})
|
||||
@pytest.mark.parametrize(
|
||||
"deleted,method",
|
||||
(
|
||||
(True, Notification.objects.deleted),
|
||||
(False, Notification.objects.active),
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_deleted_active_methods(deleted, method):
|
||||
NotificationFullFactory.create_batch(3, deleted=deleted)
|
||||
assert method().count() == 3
|
||||
|
||||
first_notification = Notification.objects.first()
|
||||
first_notification.deleted = not deleted
|
||||
first_notification.save()
|
||||
assert method().count() == 2
|
||||
|
||||
Notification.objects.all().update(deleted=not deleted)
|
||||
assert method().count() == 0
|
||||
|
||||
first_notification.deleted = deleted
|
||||
first_notification.save()
|
||||
assert method().count() == 1
|
||||
with override_settings(DJANGO_NOTIFICATIONS_CONFIG={"SOFT_DELETE": True}):
|
||||
assert func().count() == expect_deleted
|
||||
assert func(include_deleted=True).count() == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"method",
|
||||
(
|
||||
Notification.objects.deleted,
|
||||
Notification.objects.active,
|
||||
"active",
|
||||
"deleted",
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_deleted_active_methods_without_soft_delete(method):
|
||||
def test_filter_active_deleted(method):
|
||||
NotificationFullFactory.create_batch(2, deleted=False)
|
||||
NotificationFullFactory.create_batch(2, deleted=True)
|
||||
|
||||
func = getattr(Notification.objects, method)
|
||||
with pytest.raises(ImproperlyConfigured):
|
||||
assert method()
|
||||
func()
|
||||
|
||||
with override_settings(DJANGO_NOTIFICATIONS_CONFIG={"SOFT_DELETE": True}):
|
||||
assert func().count() == 2
|
||||
|
||||
|
||||
@override_settings(DJANGO_NOTIFICATIONS_CONFIG={"SOFT_DELETE": True})
|
||||
@pytest.mark.parametrize(
|
||||
"status,method,check_method",
|
||||
"method,field,initial_status,expected,final_status",
|
||||
(
|
||||
(False, Notification.objects.mark_all_as_deleted, Notification.objects.deleted),
|
||||
(True, Notification.objects.mark_all_as_active, Notification.objects.active),
|
||||
("mark_all_as_sent", "emailed", False, 2, True),
|
||||
("mark_all_as_sent", "emailed", True, 0, True),
|
||||
("mark_all_as_unsent", "emailed", True, 2, False),
|
||||
("mark_all_as_unsent", "emailed", False, 0, False),
|
||||
("mark_all_as_public", "public", True, 0, True),
|
||||
("mark_all_as_public", "public", False, 2, True),
|
||||
("mark_all_as_private", "public", False, 0, False),
|
||||
("mark_all_as_private", "public", True, 2, False),
|
||||
("mark_all_as_read", "unread", False, 0, False),
|
||||
("mark_all_as_read", "unread", True, 2, False),
|
||||
("mark_all_as_unread", "unread", False, 2, True),
|
||||
("mark_all_as_unread", "unread", True, 0, True),
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_mark_all_as_deleted_active(status, method, check_method):
|
||||
NotificationFullFactory.create_batch(3, deleted=status)
|
||||
assert Notification.objects.count() == 3
|
||||
assert check_method().count() == 0
|
||||
|
||||
method()
|
||||
assert check_method().count() == 3
|
||||
|
||||
|
||||
@override_settings(DJANGO_NOTIFICATIONS_CONFIG={"SOFT_DELETE": True})
|
||||
@pytest.mark.parametrize(
|
||||
"status,method,check_method",
|
||||
(
|
||||
(False, Notification.objects.mark_all_as_deleted, Notification.objects.deleted),
|
||||
(True, Notification.objects.mark_all_as_active, Notification.objects.active),
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_mark_all_as_deleted_active_with_recipient(status, method, check_method):
|
||||
def test_mark_all(method, field, initial_status, expected, final_status):
|
||||
recipient = RecipientFactory()
|
||||
NotificationFullFactory.create_batch(2, deleted=status, recipient=recipient)
|
||||
NotificationFullFactory.create_batch(1, deleted=status)
|
||||
assert Notification.objects.count() == 3
|
||||
assert check_method().count() == 0
|
||||
NotificationFullFactory.create_batch(2, **{field: initial_status})
|
||||
NotificationFullFactory.create_batch(2, recipient=recipient, **{field: initial_status})
|
||||
func = getattr(Notification.objects, method)
|
||||
|
||||
method(recipient=recipient)
|
||||
assert check_method().count() == 2
|
||||
assert func(recipient=recipient) == expected
|
||||
assert func() == expected
|
||||
|
||||
assert Notification.objects.filter(**{field: final_status}).count() == 4
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"method",
|
||||
"method,initial_status",
|
||||
(
|
||||
Notification.objects.mark_all_as_deleted,
|
||||
Notification.objects.mark_all_as_active,
|
||||
("mark_all_as_active", True),
|
||||
("mark_all_as_deleted", False),
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_mark_all_as_deleted_active_without_soft_delete(method):
|
||||
def test_mark_all_active_deleted(method, initial_status):
|
||||
recipient = RecipientFactory()
|
||||
NotificationFullFactory.create_batch(2, recipient=recipient, deleted=initial_status)
|
||||
NotificationFullFactory.create_batch(2, deleted=initial_status)
|
||||
|
||||
func = getattr(Notification.objects, method)
|
||||
with pytest.raises(ImproperlyConfigured):
|
||||
method()
|
||||
func()
|
||||
|
||||
with override_settings(DJANGO_NOTIFICATIONS_CONFIG={"SOFT_DELETE": True}):
|
||||
assert func(recipient=recipient) == 2
|
||||
assert func() == 2
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"status,method,check_method",
|
||||
(
|
||||
(False, Notification.objects.mark_as_sent, Notification.objects.sent),
|
||||
(True, Notification.objects.mark_as_unsent, Notification.objects.unsent),
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_mark_as_sent_unsent_method(status, method, check_method):
|
||||
NotificationFullFactory.create_batch(3, emailed=status)
|
||||
assert Notification.objects.count() == 3
|
||||
assert check_method().count() == 0
|
||||
|
||||
method()
|
||||
assert check_method().count() == 3
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"status,method,check_method",
|
||||
(
|
||||
(False, Notification.objects.mark_as_sent, Notification.objects.sent),
|
||||
(True, Notification.objects.mark_as_unsent, Notification.objects.unsent),
|
||||
),
|
||||
)
|
||||
@pytest.mark.django_db
|
||||
def test_mark_as_sent_unsent_with_recipient(status, method, check_method):
|
||||
recipient = RecipientFactory()
|
||||
NotificationFullFactory.create_batch(2, emailed=status, recipient=recipient)
|
||||
NotificationFullFactory.create_batch(1, emailed=status)
|
||||
assert Notification.objects.count() == 3
|
||||
assert check_method().count() == 0
|
||||
|
||||
method(recipient=recipient)
|
||||
assert check_method().count() == 2
|
||||
assert Notification.objects.filter(deleted=not initial_status).count() == 4
|
||||
|
|
|
|||
Loading…
Reference in a new issue