From 888c59eff30f0414f8d223252f492b896a9584de Mon Sep 17 00:00:00 2001 From: Alvaro Leonel Date: Sun, 28 Apr 2024 18:42:14 -0300 Subject: [PATCH] Refactoring the NotificationQuerySet and its tests --- notifications/notification_types.py | 4 +- notifications/querysets.py | 161 +++++++--------- notifications/tests/test_querysets.py | 261 +++++++------------------- 3 files changed, 133 insertions(+), 293 deletions(-) diff --git a/notifications/notification_types.py b/notifications/notification_types.py index dc1804e..ecc8abf 100644 --- a/notifications/notification_types.py +++ b/notifications/notification_types.py @@ -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]] diff --git a/notifications/querysets.py b/notifications/querysets.py index 21204b3..fd69428 100644 --- a/notifications/querysets.py +++ b/notifications/querysets.py @@ -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 diff --git a/notifications/tests/test_querysets.py b/notifications/tests/test_querysets.py index a7d7669..4491dc7 100644 --- a/notifications/tests/test_querysets.py +++ b/notifications/tests/test_querysets.py @@ -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