tests: add and adapt notification test cases

tests/conftest.py

@@ -3,6 +3,7 @@
 # Copyright (C) 2022 TU Wien.
 # Copyright (C) 2022 European Union.
 # Copyright (C) 2022 CERN.
+# Copyright (C) 2023 Graz University of Technology.
 #
 # Invenio-Users-Resources is free software; you can redistribute it and/or
 # modify it under the terms of the MIT License; see LICENSE file for more
@@ -19,16 +20,27 @@
 from invenio_access.permissions import any_user as any_user_need
 from invenio_accounts.proxies import current_datastore
 from invenio_app.factory import create_api
+from marshmallow import fields
 
 from invenio_users_resources.proxies import (
     current_groups_service,
     current_users_service,
 )
 from invenio_users_resources.records import GroupAggregate, UserAggregate
+from invenio_users_resources.services.schemas import (
+    NotificationPreferences,
+    UserPreferencesSchema,
+)
 
 pytest_plugins = ("celery.contrib.pytest",)
 
 
+class UserPreferencesNotificationsSchema(UserPreferencesSchema):
+    """Schema extending preferences with notification preferences."""
+
+    notifications = fields.Nested(NotificationPreferences)
+
+
 #
 # Application
 #
@@ -43,6 +55,8 @@ def app_config(app_config):
     ] = "invenio_jsonschemas.proxies.current_refresolver_store"
     # Variable not used. We set it to silent warnings
     app_config["JSONSCHEMAS_HOST"] = "not-used"
+    # setting preferences schema to test notifications
+    app_config["ACCOUNTS_USER_PREFERENCES_SCHEMA"] = UserPreferencesNotificationsSchema
 
     return app_config
 
@@ -146,6 +160,36 @@ def users_data():
             },
             "active": False,
         },
+        {
+            "username": "notification_enabled",
+            "email": "[email protected]",
+            "profile": {
+                "full_name": "Mr. Worldwide",
+                "affiliations": "World",
+            },
+            "preferences": {
+                "visibility": "restricted",
+                "email_visibility": "public",
+                "notifications": {
+                    "enabled": True,
+                },
+            },
+        },
+        {
+            "username": "notification_disabled",
+            "email": "[email protected]",
+            "profile": {
+                "full_name": "Loner",
+                "affiliations": "Home",
+            },
+            "preferences": {
+                "visibility": "restricted",
+                "email_visibility": "public",
+                "notifications": {
+                    "enabled": False,
+                },
+            },
+        },
     ]
 
 
@@ -226,3 +270,15 @@ def user_inactive(users):
 def user_unconfirmed(users):
     """Unconfirmed user."""
     return users["unconfirmed"]
+
+
+@pytest.fixture(scope="module")
+def user_notification_enabled(users):
+    """User with notfications enabled."""
+    return users["notification_enabled"]
+
+
+@pytest.fixture(scope="module")
+def user_notification_disabled(users):
+    """User with notfications disabled."""
+    return users["notification_disabled"]

tests/test_notifications.py

@@ -8,35 +8,75 @@
 
 """Notification related tests."""
 
-from copy import deepcopy
-
 from invenio_notifications.models import Notification, Recipient
 
-from invenio_users_resources.notifications import UserPreferencesRecipientFilter
+from invenio_users_resources.notifications import (
+    UserPreferencesRecipientFilter,
+    UserRecipient,
+)
 from invenio_users_resources.records.api import UserAggregate
 
 
-def test_user_recipient_filter(user_pub):
-    """Test user recipient filter for notifications."""
-    preferences_filter = UserPreferencesRecipientFilter()
+def test_user_recipient_generator(user_notification_disabled, user_notification_enabled):
+    generator_disabled = UserRecipient(key="disabled")
+    generator_enabled = UserRecipient(key="enabled")
+
+    user_notifications_disabled = UserAggregate.from_user(user_notification_disabled.user).dumps()
+    user_notifications_enabled = UserAggregate.from_user(
+        user_notification_enabled.user
+    ).dumps()
+
+    n = Notification(
+        type="",
+        context={
+            "disabled": user_notifications_disabled,
+            "enabled": user_notifications_enabled,
+        },
+    )
+
+    recipients = {}
 
-    u = UserAggregate.from_user(user_pub.user).dumps()
+    generator_disabled(n, recipients=recipients)
+    assert 1 == len(recipients)
+    assert user_notifications_disabled["id"] in recipients.keys()
 
-    user_notifications_enabled = deepcopy(u)
-    user_notifications_enabled["preferences"]["notifications"] = {"enabled": True}
-    user_notifications_disabled = deepcopy(u)
-    user_notifications_disabled["preferences"]["notifications"] = {"enabled": False}
+    generator_enabled(n, recipients=recipients)
+    assert 2 == len(recipients)
+    assert [
+        user_notifications_disabled["id"],
+        user_notifications_enabled["id"],
+    ] == list(recipients)
+
+    # checking that user does not get added twice
+    generator_disabled(n, recipients=recipients)
+    assert 2 == len(recipients)
+    assert [
+        user_notifications_disabled["id"],
+        user_notifications_enabled["id"],
+    ] == list(recipients)
+
+
+def test_user_recipient_filter(user_notification_disabled, user_notification_enabled):
+    """Test user recipient filter for notifications."""
+
+    user_notifications_disabled = UserAggregate.from_user(user_notification_disabled.user).dumps()
+    user_notifications_enabled = UserAggregate.from_user(
+        user_notification_enabled.user
+    ).dumps()
 
     n = Notification(type="", context={})
-    recipient_enabled = Recipient(data=user_notifications_enabled)
-    recipient_disabled = Recipient(data=user_notifications_disabled)
+    recipient_enabled = Recipient(data=user_notifications_disabled)
+    recipient_disabled = Recipient(data=user_notifications_enabled)
+
+    recipients = {
+        user_notifications_disabled["id"]: recipient_disabled,
+        user_notifications_enabled["id"]: recipient_enabled,
+    }
 
+    preferences_filter = UserPreferencesRecipientFilter()
     filtered_recipients = preferences_filter(
         notification=n,
-        recipients={
-            user_notifications_disabled["id"]: recipient_disabled,
-            user_notifications_enabled["id"]: recipient_enabled,
-        },
+        recipients=recipients,
     )
 
     assert 1 == len(filtered_recipients)