Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat: added account level preference model in notification task (#36858)

Browse Source 
feat: added account level pref model in notification task

feat: added handler for preference creation on user signup

feat: added unit tests for account level preferences

* fix: removed app level check in notifications

* fix: updated flaky test
This commit is contained in:
Ahtisham Shahid
2025-07-01 14:09:02 +05:00
committed by GitHub
parent b27c41cc8c
commit 498dd56fec
7 changed files with 786 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
openedx/core/djangoapps/notifications/config/waffle.py
View File

@@ -69,3 +69,14 @@ ENABLE_NOTIFY_ALL_LEARNERS = CourseWaffleFlag(f'{WAFFLE_NAMESPACE}.enable_post_n
# .. toggle_target_removal_date: 2026-05-27
# .. toggle_warning: When the flag is ON, Notifications will go through ace push channels.
ENABLE_PUSH_NOTIFICATIONS = CourseWaffleFlag(f'{WAFFLE_NAMESPACE}.enable_push_notifications', __name__)
# .. toggle_name: notifications.enable_account_level_preferences
# .. toggle_implementation: CourseWaffleFlag
# .. toggle_default: False
# .. toggle_description: Waffle flag to enable account level preferences for notifications
# .. toggle_use_cases: temporary, open_edx
# .. toggle_creation_date: 2025-04-29
# .. toggle_target_removal_date: 2025-07-29
# .. toggle_warning: When the flag is ON, account level preferences for notifications are enabled.
# .. toggle_tickets: INF-1472
ENABLE_ACCOUNT_LEVEL_PREFERENCES = WaffleFlag(f'{WAFFLE_NAMESPACE}.enable_account_level_preferences', __name__)

31
openedx/core/djangoapps/notifications/handlers.py
View File

@@ -3,8 +3,10 @@ Handlers for notifications
"""
import logging
from django.contrib.auth import get_user_model
from django.core.exceptions import ObjectDoesNotExist
from django.db import IntegrityError, transaction
from django.db import IntegrityError, transaction, ProgrammingError
from django.db.models.signals import post_save
from django.dispatch import receiver
from openedx_events.learning.signals import (
COURSE_ENROLLMENT_CREATED,
@@ -21,12 +23,14 @@ from openedx.core.djangoapps.notifications.audience_filters import (
ForumRoleAudienceFilter,
TeamAudienceFilter
)
from openedx.core.djangoapps.notifications.base_notification import NotificationAppManager
from openedx.core.djangoapps.notifications.base_notification import NotificationAppManager, COURSE_NOTIFICATION_TYPES
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS, ENABLE_ORA_GRADE_NOTIFICATION
from openedx.core.djangoapps.notifications.email import ONE_CLICK_EMAIL_UNSUB_KEY
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, NotificationPreference
from openedx.core.djangoapps.notifications.tasks import create_notification_preference
from openedx.core.djangoapps.user_api.models import UserPreference
User = get_user_model()
log = logging.getLogger(__name__)
AUDIENCE_FILTER_CLASSES = {
@@ -63,6 +67,27 @@ def course_enrollment_post_save(signal, sender, enrollment, metadata, **kwargs):
f'and course {enrollment.course.course_key}')
@receiver(post_save, sender=User)
def create_user_account_preferences(sender, instance, created, **kwargs): # pylint: disable=unused-argument
"""
Initialize user notification preferences when new user is created.
"""
preferences = []
if created:
try:
with transaction.atomic():
for name in COURSE_NOTIFICATION_TYPES.keys():
preferences.append(create_notification_preference(instance.id, name))
NotificationPreference.objects.bulk_create(preferences, ignore_conflicts=True)
except IntegrityError:
log.info(f'Account-level CourseNotificationPreference already exists for user {instance.id}')
except ProgrammingError as e:
# This is here because there is a dependency issue in the migrations where
# this signal handler tries to run before the NotificationPreference model is created.
# In reality, this should never be hit because migrations will have already run.
log.error(f'ProgrammingError encountered while creating user preferences: {e}')
@receiver(COURSE_UNENROLLMENT_COMPLETED)
def on_user_course_unenrollment(enrollment, **kwargs):
"""

11
openedx/core/djangoapps/notifications/management/tests/test_migrate_to_account_level_model.py
View File

@@ -5,15 +5,14 @@ Test for account level migration command
from unittest.mock import Mock, patch
from django.contrib.auth import get_user_model
from django.test import TestCase
from django.core.management import call_command
from django.db.models.signals import post_save
from django.test import TestCase
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
NotificationPreference
)
from openedx.core.djangoapps.notifications.handlers import create_user_account_preferences
from openedx.core.djangoapps.notifications.management.commands.migrate_preferences_to_account_level_model import Command
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, NotificationPreference
User = get_user_model()
COMMAND_MODULE = 'openedx.core.djangoapps.notifications.management.commands.migrate_preferences_to_account_level_model'
@@ -24,6 +23,8 @@ class MigrateNotificationPreferencesTestCase(TestCase):
def setUp(self):
"""Set up test data."""
# Disconnect before creating users
post_save.disconnect(create_user_account_preferences, sender=User)
self.user1 = User.objects.create_user(username='user1', email='user1@example.com')
self.user2 = User.objects.create_user(username='user2', email='user2@example.com')
self.user3 = User.objects.create_user(username='user3', email='user3@example.com')

30
openedx/core/djangoapps/notifications/models.py
View File

@@ -130,6 +130,7 @@ class NotificationPreference(TimeStampedModel):
"""
Model to store notification preferences for users at account level
"""
class EmailCadenceChoices(models.TextChoices):
DAILY = 'Daily'
WEEKLY = 'Weekly'
@@ -148,6 +149,33 @@ class NotificationPreference(TimeStampedModel):
email_cadence = models.CharField(max_length=64, choices=EmailCadenceChoices.choices, null=False, blank=False)
is_active = models.BooleanField(default=True)
def is_enabled_for_any_channel(self, *args, **kwargs) -> bool:
"""
Returns True if the notification preference is enabled for any channel.
"""
return self.web or self.push or self.email
def get_channels_for_notification_type(self, *args, **kwargs) -> list:
"""
Returns the channels for the given app name and notification type.
Sample Response:
['web', 'push']
"""
channels = []
if self.web:
channels.append('web')
if self.push:
channels.append('push')
if self.email:
channels.append('email')
return channels
def get_email_cadence_for_notification_type(self, *args, **kwargs) -> str:
"""
Returns the email cadence for the notification type.
"""
return self.email_cadence
class CourseNotificationPreference(TimeStampedModel):
"""
@@ -192,7 +220,7 @@ class CourseNotificationPreference(TimeStampedModel):
preferences.config_version = current_config_version
preferences.notification_preference_config = new_prefs
preferences.save()
# pylint: disable-next=broad-except
# pylint: disable-next=broad-except
except Exception as e:
log.error(f'Unable to update notification preference to new config. {e}')
return preferences

143
openedx/core/djangoapps/notifications/tasks.py
View File

@@ -16,16 +16,19 @@ from pytz import UTC
from common.djangoapps.student.models import CourseEnrollment
from openedx.core.djangoapps.notifications.audience_filters import NotificationFilter
from openedx.core.djangoapps.notifications.base_notification import (
COURSE_NOTIFICATION_APPS,
COURSE_NOTIFICATION_TYPES,
get_default_values_of_preference,
get_notification_content
)
from openedx.core.djangoapps.notifications.email.tasks import send_immediate_cadence_email
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
from openedx.core.djangoapps.notifications.config.waffle import (
ENABLE_ACCOUNT_LEVEL_PREFERENCES,
ENABLE_NOTIFICATION_GROUPING,
ENABLE_NOTIFICATIONS,
ENABLE_PUSH_NOTIFICATIONS
)
from openedx.core.djangoapps.notifications.email.tasks import send_immediate_cadence_email
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
from openedx.core.djangoapps.notifications.events import notification_generated_event
from openedx.core.djangoapps.notifications.grouping_notifications import (
NotificationRegistry,
@@ -35,12 +38,12 @@ from openedx.core.djangoapps.notifications.grouping_notifications import (
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
Notification,
NotificationPreference,
get_course_notification_preference_config_version
)
from openedx.core.djangoapps.notifications.push.tasks import send_ace_msg_to_push_channel
from openedx.core.djangoapps.notifications.utils import clean_arguments, get_list_in_batches
logger = get_task_logger(__name__)
@@ -137,6 +140,8 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
if not is_notification_valid(notification_type, context):
raise ValidationError(f"Notification is not valid {app_name} {notification_type} {context}")
account_level_pref_enabled = ENABLE_ACCOUNT_LEVEL_PREFERENCES.is_enabled()
user_ids = list(set(user_ids))
batch_size = settings.NOTIFICATION_CREATION_BATCH_SIZE
group_by_id = context.pop('group_by_id', '')
@@ -164,18 +169,31 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
logger.debug(f'After applying filters, sending notifications to {len(batch_user_ids)} users in {course_key}')
existing_notifications = (
get_user_existing_notifications(batch_user_ids, notification_type, group_by_id, course_key))\
get_user_existing_notifications(batch_user_ids, notification_type, group_by_id, course_key)) \
if grouping_enabled else {}
# check if what is preferences of user and make decision to send notification or not
preferences = CourseNotificationPreference.objects.filter(
user_id__in=batch_user_ids,
course_id=course_key,
)
preferences = list(preferences)
if account_level_pref_enabled:
preferences = NotificationPreference.objects.filter(
user_id__in=batch_user_ids,
app=app_name,
type=notification_type
)
else:
preferences = CourseNotificationPreference.objects.filter(
user_id__in=batch_user_ids,
course_id=course_key,
)
preferences = list(preferences)
if default_web_config:
preferences = create_notification_pref_if_not_exists(batch_user_ids, preferences, course_key)
if account_level_pref_enabled:
preferences = create_account_notification_pref_if_not_exists(
batch_user_ids, preferences, notification_type
)
else:
preferences = create_notification_pref_if_not_exists(batch_user_ids, preferences, course_key)
if not preferences:
continue
@@ -183,15 +201,15 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
notifications = []
for preference in preferences:
user_id = preference.user_id
preference = update_user_preference(preference, user_id, course_key)
if not account_level_pref_enabled:
preference = update_user_preference(preference, user_id, course_key)
if (
preference and
preference.is_enabled_for_any_channel(app_name, notification_type) and
preference.get_app_config(app_name).get('enabled', False)
preference.is_enabled_for_any_channel(app_name, notification_type)
):
notification_preferences = preference.get_channels_for_notification_type(app_name, notification_type)
email_enabled = 'email' in preference.get_channels_for_notification_type(app_name, notification_type)
email_enabled = 'email' in notification_preferences
email_cadence = preference.get_email_cadence_for_notification_type(app_name, notification_type)
push_notification = is_push_notification_enabled and 'push' in notification_preferences
new_notification = Notification(
@@ -257,6 +275,82 @@ def update_user_preference(preference: CourseNotificationPreference, user_id, co
return preference
def update_account_user_preference(user_id: int) -> None:
"""
Update account level user preferences to ensure all notification types are present.
"""
notification_types = set(COURSE_NOTIFICATION_TYPES.keys())
# Get existing notification types for the user
existing_types = set(
NotificationPreference.objects
.filter(user_id=user_id, type__in=notification_types)
.values_list('type', flat=True)
)
# Find missing notification types
missing_types = notification_types - existing_types
if not missing_types:
return
# Create new preferences for missing types
new_preferences = [
create_notification_preference(user_id, notification_type)
for notification_type in missing_types
]
# Bulk create all new preferences
NotificationPreference.objects.bulk_create(new_preferences)
return
def create_notification_preference(user_id: int, notification_type: str) -> NotificationPreference:
"""
Create a single notification preference with appropriate defaults.
Args:
user_id: ID of the user
notification_type: Type of notification
Returns:
NotificationPreference instance
"""
notification_config = COURSE_NOTIFICATION_TYPES.get(notification_type, {})
is_core = notification_config.get('is_core', False)
app = COURSE_NOTIFICATION_TYPES[notification_type]['notification_app']
email_cadence = notification_config.get('email_cadence', EmailCadence.DAILY)
if is_core:
email_cadence = COURSE_NOTIFICATION_APPS[app]['core_email_cadence']
return NotificationPreference(
user_id=user_id,
type=notification_type,
app=app,
web=_get_channel_default(is_core, notification_type, 'web'),
push=_get_channel_default(is_core, notification_type, 'push'),
email=_get_channel_default(is_core, notification_type, 'email'),
email_cadence=email_cadence,
)
def _get_channel_default(is_core: bool, notification_type: str, channel: str) -> bool:
"""
Get the default value for a notification channel.
Args:
is_core: Whether this is a core notification
notification_type: Type of notification
channel: Channel name (web, push, email)
Returns:
Default boolean value for the channel
"""
if is_core:
notification_app = COURSE_NOTIFICATION_TYPES[notification_type]['notification_app']
return COURSE_NOTIFICATION_APPS[notification_app][f'core_{channel}']
return COURSE_NOTIFICATION_TYPES[notification_type][channel]
def create_notification_pref_if_not_exists(user_ids: List, preferences: List, course_id: CourseKey):
"""
Create notification preference if not exist.
@@ -275,3 +369,24 @@ def create_notification_pref_if_not_exists(user_ids: List, preferences: List, co
CourseNotificationPreference.objects.bulk_create(new_preferences, ignore_conflicts=True)
preferences = preferences + new_preferences
return preferences
def create_account_notification_pref_if_not_exists(user_ids: List, preferences: List, notification_type: str):
"""
Create account level notification preference if not exist.
"""
new_preferences = []
for user_id in user_ids:
if not any(preference.user_id == int(user_id) for preference in preferences):
new_preferences.append(create_notification_preference(
user_id=int(user_id),
notification_type=notification_type,
))
if new_preferences:
# ignoring conflicts because it is possible that preference is already created by another process
# conflicts may arise because of constraint on user_id and course_id fields in model
NotificationPreference.objects.bulk_create(new_preferences, ignore_conflicts=True)
preferences = preferences + new_preferences
return preferences

36
openedx/core/djangoapps/notifications/tests/test_tasks.py
View File

@@ -230,32 +230,6 @@ class SendNotificationsTest(ModuleStoreTestCase):
self.assertEqual(Notification.objects.filter(user_id=self.user.id).count(), 1)
user_notifications_mock.assert_called_once()
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.data(
('discussion', 'new_comment_on_response'), # core notification
('discussion', 'new_response'), # non core notification
)
@ddt.unpack
def test_send_with_app_disabled_notifications(self, app_name, notification_type):
"""
Test send_notifications does not create a new notification if the app is disabled.
"""
self.preference_v1.notification_preference_config['discussion']['enabled'] = False
self.preference_v1.save()
context = {
'post_title': 'Post title',
'replier_name': 'replier name',
}
content_url = 'https://example.com/'
# Call the `send_notifications` function.
send_notifications([self.user.id], str(self.course_1.id), app_name, notification_type, context, content_url)
# Assert that `Notification` objects are not created for the users.
notification = Notification.objects.filter(user_id=self.user.id).first()
self.assertIsNone(notification)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
def test_notification_not_created_when_context_is_incomplete(self):
try:
@@ -295,9 +269,9 @@ class SendBatchNotificationsTest(ModuleStoreTestCase):
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.data(
(settings.NOTIFICATION_CREATION_BATCH_SIZE, 13, 6),
(settings.NOTIFICATION_CREATION_BATCH_SIZE + 10, 15, 9),
(settings.NOTIFICATION_CREATION_BATCH_SIZE - 10, 13, 5),
(settings.NOTIFICATION_CREATION_BATCH_SIZE, 14, 7),
(settings.NOTIFICATION_CREATION_BATCH_SIZE + 10, 16, 10),
(settings.NOTIFICATION_CREATION_BATCH_SIZE - 10, 14, 6),
)
@ddt.unpack
def test_notification_is_send_in_batch(self, creation_size, prefs_query_count, notifications_query_count):
@@ -348,7 +322,7 @@ class SendBatchNotificationsTest(ModuleStoreTestCase):
"username": "Test Author"
}
with override_waffle_flag(ENABLE_NOTIFICATIONS, active=True):
with self.assertNumQueries(13):
with self.assertNumQueries(14):
send_notifications(user_ids, str(self.course.id), notification_app, notification_type,
context, "http://test.url")
@@ -368,7 +342,7 @@ class SendBatchNotificationsTest(ModuleStoreTestCase):
}
with override_waffle_flag(ENABLE_NOTIFICATIONS, active=True):
with override_waffle_flag(ENABLE_PUSH_NOTIFICATIONS, active=True):
with self.assertNumQueries(15):
with self.assertNumQueries(16):
send_notifications(user_ids, str(self.course.id), notification_app, notification_type,
context, "http://test.url")

578
openedx/core/djangoapps/notifications/tests/test_tasks_with_account_level_pref.py Normal file
View File

@@ -0,0 +1,578 @@
"""
Tests for notifications tasks.
"""
import datetime
from unittest.mock import patch
import ddt
from django.conf import settings
from django.core.exceptions import ValidationError
from edx_toggles.toggles.testutils import override_waffle_flag
from common.djangoapps.student.models import CourseEnrollment
from common.djangoapps.student.tests.factories import UserFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
from ..config.waffle import ENABLE_ACCOUNT_LEVEL_PREFERENCES, ENABLE_NOTIFICATION_GROUPING, ENABLE_NOTIFICATIONS
from ..models import CourseNotificationPreference, Notification, NotificationPreference
from ..tasks import (
create_notification_pref_if_not_exists,
delete_notifications,
send_notifications,
update_user_preference
)
from .utils import create_notification
@patch('openedx.core.djangoapps.notifications.models.COURSE_NOTIFICATION_CONFIG_VERSION', 1)
@override_waffle_flag(ENABLE_ACCOUNT_LEVEL_PREFERENCES, active=True)
class TestNotificationsTasks(ModuleStoreTestCase):
"""
Tests for notifications tasks.
"""
def setUp(self):
"""
Create a course and users for the course.
"""
super().setUp()
self.user = UserFactory()
self.user_1 = UserFactory()
self.user_2 = UserFactory()
self.course_1 = CourseFactory.create(
org='testorg',
number='testcourse',
run='testrun'
)
self.course_2 = CourseFactory.create(
org='testorg',
number='testcourse_2',
run='testrun'
)
self.preference_v1 = CourseNotificationPreference.objects.create(
user_id=self.user.id,
course_id=self.course_1.id,
config_version=0,
)
self.preference_v2 = CourseNotificationPreference.objects.create(
user_id=self.user.id,
course_id=self.course_2.id,
config_version=1,
)
def test_update_user_preference(self):
"""
Test whether update_user_preference updates the preference with the latest config version.
"""
# Test whether update_user_preference updates the preference with a different config version
updated_preference = update_user_preference(self.preference_v1, self.user, self.course_1.id)
self.assertEqual(updated_preference.config_version, 1)
# Test whether update_user_preference does not update the preference if the config version is the same
updated_preference = update_user_preference(self.preference_v2, self.user, self.course_2.id)
self.assertEqual(updated_preference.config_version, 1)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
def test_create_notification_pref_if_not_exists(self):
"""
Test whether create_notification_pref_if_not_exists creates a new preference if it doesn't exist.
"""
# Test whether create_notification_pref_if_not_exists creates a new preference if it doesn't exist
user_ids = [self.user.id, self.user_1.id, self.user_2.id]
preferences = [self.preference_v2]
updated_preferences = create_notification_pref_if_not_exists(user_ids, preferences, self.course_2.id)
self.assertEqual(len(updated_preferences), 3) # Should have created two new preferences
# Test whether create_notification_pref_if_not_exists doesn't create a new preference if it already exists
updated_preferences = create_notification_pref_if_not_exists(user_ids, preferences, self.course_2.id)
self.assertEqual(len(updated_preferences), 3) # No new preferences should be created this time
@ddt.ddt
@override_waffle_flag(ENABLE_ACCOUNT_LEVEL_PREFERENCES, active=True)
class SendNotificationsTest(ModuleStoreTestCase):
"""
Tests for send_notifications.
"""
def setUp(self):
"""
Create a course and users for the course.
"""
super().setUp()
self.user = UserFactory()
self.course_1 = CourseFactory.create(
org='testorg',
number='testcourse',
run='testrun'
)
self.preference_v1 = CourseNotificationPreference.objects.create(
user_id=self.user.id,
course_id=self.course_1.id,
config_version=0,
)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.data(
('discussion', 'new_comment_on_response'), # core notification
('discussion', 'new_response'), # non core notification
)
@ddt.unpack
def test_send_notifications(self, app_name, notification_type):
"""
Test whether send_notifications creates a new notification.
"""
context = {
'post_title': 'Post title',
'replier_name': 'replier name',
}
content_url = 'https://example.com/'
# Call the `send_notifications` function.
with patch('openedx.core.djangoapps.notifications.tasks.notification_generated_event') as event_mock:
send_notifications([self.user.id], str(self.course_1.id), app_name, notification_type, context, content_url)
assert event_mock.called
assert event_mock.call_args[0][0] == [self.user.id]
assert event_mock.call_args[0][1] == app_name
assert event_mock.call_args[0][2] == notification_type
# Assert that `Notification` objects have been created for the users.
notification = Notification.objects.filter(user_id=self.user.id).first()
# Assert that the `Notification` objects have the correct properties.
self.assertEqual(notification.user_id, self.user.id)
self.assertEqual(notification.app_name, app_name)
self.assertEqual(notification.notification_type, notification_type)
self.assertEqual(notification.content_context, context)
self.assertEqual(notification.content_url, content_url)
self.assertEqual(notification.course_id, self.course_1.id)
@ddt.data(True, False)
def test_enable_notification_flag(self, flag_value):
"""
Tests if notification is sent when flag is enabled and notification
is not sent when flag is disabled
"""
app_name = "discussion"
notification_type = "new_response"
context = {
'post_title': 'Post title',
'replier_name': 'replier name',
}
content_url = 'https://example.com/'
with override_waffle_flag(ENABLE_NOTIFICATIONS, active=flag_value):
send_notifications([self.user.id], str(self.course_1.id), app_name, notification_type, context, content_url)
created_notifications_count = 1 if flag_value else 0
self.assertEqual(len(Notification.objects.all()), created_notifications_count)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
def test_notification_not_send_with_preference_disabled(self):
"""
Tests notification not send if preference is disabled
"""
app_name = "discussion"
notification_type = "new_response"
context = {
'post_title': 'Post title',
'replier_name': 'replier name',
}
content_url = 'https://example.com/'
preference = CourseNotificationPreference.get_user_course_preference(self.user.id, self.course_1.id)
app_prefs = preference.notification_preference_config[app_name]
app_prefs['notification_types']['core']['web'] = False
app_prefs['notification_types']['core']['email'] = False
app_prefs['notification_types']['core']['push'] = False
preference.save()
account_preferences, __created = NotificationPreference.objects.get_or_create(
user_id=self.user.id,
app=app_name,
type=notification_type,
)
account_preferences.web = False
account_preferences.email = False
account_preferences.push = False
account_preferences.save()
send_notifications([self.user.id], str(self.course_1.id), app_name, notification_type, context, content_url)
self.assertEqual(len(Notification.objects.all()), 0)
@override_waffle_flag(ENABLE_NOTIFICATION_GROUPING, True)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
def test_send_notification_with_grouping_enabled(self):
"""
Test send_notifications with grouping enabled.
"""
(
self.preference_v1.notification_preference_config['discussion']
['notification_types']['new_discussion_post']['web']
) = True
self.preference_v1.save()
account_preferences, __created = NotificationPreference.objects.get_or_create(
user_id=self.user.id,
app='discussion',
type='new_discussion_post',
)
account_preferences.web = True
account_preferences.save()
with patch('openedx.core.djangoapps.notifications.tasks.group_user_notifications') as user_notifications_mock:
context = {
'post_title': 'Test Post',
'username': 'Test Author',
'group_by_id': 'group_by_id'
}
content_url = 'https://example.com/'
send_notifications(
[self.user.id],
str(self.course_1.id),
'discussion',
'new_discussion_post',
{**context},
content_url
)
send_notifications(
[self.user.id],
str(self.course_1.id),
'discussion',
'new_discussion_post',
{**context},
content_url
)
self.assertEqual(Notification.objects.filter(user_id=self.user.id).count(), 1)
user_notifications_mock.assert_called_once()
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
def test_notification_not_created_when_context_is_incomplete(self):
try:
send_notifications([self.user.id], str(self.course_1.id), "discussion", "new_comment", {}, "")
except Exception as exc: # pylint: disable=broad-except
assert isinstance(exc, ValidationError)
@ddt.ddt
@override_waffle_flag(ENABLE_ACCOUNT_LEVEL_PREFERENCES, active=True)
class SendBatchNotificationsTest(ModuleStoreTestCase):
"""
Test that notification and notification preferences are created in batches
"""
def setUp(self):
"""
Setups test case
"""
super().setUp()
self.course = CourseFactory.create(
org='test_org',
number='test_course',
run='test_run'
)
def _create_users(self, num_of_users):
"""
Create users and enroll them in course
"""
users = [
UserFactory.create(username=f'user{i}', email=f'user{i}@example.com')
for i in range(num_of_users)
]
for user in users:
CourseEnrollment.enroll(user=user, course_key=self.course.id)
return users
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.data(
(settings.NOTIFICATION_CREATION_BATCH_SIZE, 14, 5),
(settings.NOTIFICATION_CREATION_BATCH_SIZE + 10, 16, 7),
(settings.NOTIFICATION_CREATION_BATCH_SIZE - 10, 14, 5),
)
@ddt.unpack
def test_notification_is_send_in_batch(self, creation_size, prefs_query_count, notifications_query_count):
"""
Tests notifications and notification preferences are created in batches
"""
notification_app = "discussion"
notification_type = "new_discussion_post"
users = self._create_users(creation_size)
user_ids = [user.id for user in users]
context = {
"post_title": "Test Post",
"username": "Test Author"
}
# Creating preferences and asserting query count
with self.assertNumQueries(prefs_query_count):
send_notifications(user_ids, str(self.course.id), notification_app, notification_type,
context, "http://test.url")
# Updating preferences for notification creation
preferences = CourseNotificationPreference.objects.filter(
user_id__in=user_ids,
course_id=self.course.id
)
for preference in preferences:
discussion_config = preference.notification_preference_config['discussion']
discussion_config['notification_types'][notification_type]['web'] = True
preference.save()
# Creating notifications and asserting query count
with self.assertNumQueries(notifications_query_count):
send_notifications(user_ids, str(self.course.id), notification_app, notification_type,
context, "http://test.url")
def test_preference_not_created_for_default_off_preference(self):
"""
Tests if new preferences are NOT created when default preference for
notification type is False
"""
notification_app = "discussion"
notification_type = "new_discussion_post"
users = self._create_users(20)
user_ids = [user.id for user in users]
context = {
"post_title": "Test Post",
"username": "Test Author"
}
with override_waffle_flag(ENABLE_NOTIFICATIONS, active=True):
with self.assertNumQueries(14):
send_notifications(user_ids, str(self.course.id), notification_app, notification_type,
context, "http://test.url")
def test_preference_created_for_default_on_preference(self):
"""
Tests if new preferences are created when default preference for
notification type is True
"""
notification_app = "discussion"
notification_type = "new_comment"
users = self._create_users(20)
NotificationPreference.objects.all().delete()
user_ids = [user.id for user in users]
context = {
"post_title": "Test Post",
"author_name": "Test Author",
"replier_name": "Replier Name"
}
with override_waffle_flag(ENABLE_NOTIFICATIONS, active=True):
with self.assertNumQueries(16):
send_notifications(user_ids, str(self.course.id), notification_app, notification_type,
context, "http://test.url")
def _update_user_preference(self, user_id, pref_exists):
"""
Removes or creates user preference based on pref_exists
"""
if pref_exists:
CourseNotificationPreference.objects.get_or_create(user_id=user_id, course_id=self.course.id)
else:
CourseNotificationPreference.objects.filter(user_id=user_id, course_id=self.course.id).delete()
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.data(
("new_response", True, True, 2),
("new_response", False, False, 2),
("new_response", True, False, 2),
("new_discussion_post", True, True, 0),
("new_discussion_post", False, False, 0),
("new_discussion_post", True, False, 0),
)
@ddt.unpack
def test_preference_enabled_in_batch_audience(self, notification_type,
user_1_pref_exists, user_2_pref_exists, generated_count):
"""
Tests if users with preference enabled in batch gets notification
"""
users = self._create_users(2)
user_ids = [user.id for user in users]
self._update_user_preference(user_ids[0], user_1_pref_exists)
self._update_user_preference(user_ids[1], user_2_pref_exists)
app_name = "discussion"
context = {
'post_title': 'Post title',
'username': 'Username',
'replier_name': 'replier name',
'author_name': 'Authorname'
}
content_url = 'https://example.com/'
send_notifications(user_ids, str(self.course.id), app_name, notification_type, context, content_url)
self.assertEqual(len(Notification.objects.all()), generated_count)
@override_waffle_flag(ENABLE_ACCOUNT_LEVEL_PREFERENCES, active=True)
class TestDeleteNotificationTask(ModuleStoreTestCase):
"""
Tests delete_notification_function
"""
def setUp(self):
"""
Setup
"""
super().setUp()
self.user = UserFactory()
self.course_1 = CourseFactory.create(org='org', number='num', run='run_01')
self.course_2 = CourseFactory.create(org='org', number='num', run='run_02')
Notification.objects.all().delete()
def test_app_name_param(self):
"""
Tests if app_name parameter works as expected
"""
assert not Notification.objects.all()
create_notification(self.user, self.course_1.id, app_name='discussion', notification_type='new_comment')
create_notification(self.user, self.course_1.id, app_name='updates', notification_type='course_updates')
delete_notifications({'app_name': 'discussion'})
assert not Notification.objects.filter(app_name='discussion')
assert Notification.objects.filter(app_name='updates')
def test_notification_type_param(self):
"""
Tests if notification_type parameter works as expected
"""
assert not Notification.objects.all()
create_notification(self.user, self.course_1.id, app_name='discussion', notification_type='new_comment')
create_notification(self.user, self.course_1.id, app_name='discussion', notification_type='new_response')
delete_notifications({'notification_type': 'new_comment'})
assert not Notification.objects.filter(notification_type='new_comment')
assert Notification.objects.filter(notification_type='new_response')
def test_created_param(self):
"""
Tests if created parameter works as expected
"""
assert not Notification.objects.all()
create_notification(self.user, self.course_1.id, created=datetime.datetime(2024, 2, 10))
create_notification(self.user, self.course_2.id, created=datetime.datetime(2024, 3, 12, 5))
kwargs = {
'created': {
'created__gte': datetime.datetime(2024, 3, 12, 0, 0, 0),
'created__lte': datetime.datetime(2024, 3, 12, 23, 59, 59),
}
}
delete_notifications(kwargs)
self.assertEqual(Notification.objects.all().count(), 1)
def test_course_id_param(self):
"""
Tests if course_id parameter works as expected
"""
assert not Notification.objects.all()
create_notification(self.user, self.course_1.id)
create_notification(self.user, self.course_2.id)
delete_notifications({'course_id': self.course_1.id})
assert not Notification.objects.filter(course_id=self.course_1.id)
assert Notification.objects.filter(course_id=self.course_2.id)
@ddt.ddt
class NotificationCreationOnChannelsTests(ModuleStoreTestCase):
"""
Tests for notification creation and channels value.
"""
def setUp(self):
"""
Create a course and users for tests.
"""
super().setUp()
self.user = UserFactory()
self.course = CourseFactory.create(
org='testorg',
number='testcourse',
run='testrun'
)
self.preference = CourseNotificationPreference.objects.create(
user_id=self.user.id,
course_id=self.course.id,
config_version=0,
)
self.account_preference, __created = NotificationPreference.objects.get_or_create(
user_id=self.user.id,
app='discussion',
type='new_discussion_post',
web=False,
email=False,
)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.data(
(False, False, 0),
(False, True, 1),
(True, False, 1),
(True, True, 1),
)
@ddt.unpack
def test_notification_is_created_when_any_channel_is_enabled(self, web_value, email_value, generated_count):
"""
Tests if notification is created if any preference is enabled
"""
app_name = 'discussion'
notification_type = 'new_discussion_post'
app_prefs = self.preference.notification_preference_config[app_name]
app_prefs['notification_types'][notification_type]['web'] = web_value
app_prefs['notification_types'][notification_type]['email'] = email_value
kwargs = {
'user_ids': [self.user.id],
'course_key': str(self.course.id),
'app_name': app_name,
'notification_type': notification_type,
'content_url': 'https://example.com/',
'context': {
'post_title': 'Post title',
'username': 'user name',
},
}
self.preference.save()
with patch('openedx.core.djangoapps.notifications.tasks.notification_generated_event') as event_mock:
send_notifications(**kwargs)
notifications = Notification.objects.all()
assert len(notifications) == generated_count
if notifications:
notification = Notification.objects.all()[0]
assert notification.web == web_value
assert notification.email == email_value
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@override_waffle_flag(ENABLE_ACCOUNT_LEVEL_PREFERENCES, active=True)
@ddt.data(
(False, False, 0),
(False, True, 1),
(True, False, 1),
(True, True, 1),
)
@ddt.unpack
def test_notification_is_created_when_any_channel_is_account_level(self, web_value, email_value, generated_count):
"""
Tests if notification is created if any preference is enabled on account level preferences
"""
app_name = 'discussion'
notification_type = 'new_discussion_post'
self.account_preference.web = web_value
self.account_preference.email = email_value
self.account_preference.save()
kwargs = {
'user_ids': [self.user.id],
'course_key': str(self.course.id),
'app_name': app_name,
'notification_type': notification_type,
'content_url': 'https://example.com/',
'context': {
'post_title': 'Post title',
'username': 'user name',
},
}
with patch('openedx.core.djangoapps.notifications.tasks.notification_generated_event') as event_mock:
send_notifications(**kwargs)
notifications = Notification.objects.all()
assert len(notifications) == generated_count
if notifications:
notification = Notification.objects.all()[0]
assert notification.web == web_value
assert notification.email == email_value