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

refactor: removed all references to CourseNotificationPreference (#37768)

Browse Source
This commit is contained in:
Ahtisham Shahid
2025-12-29 15:11:56 +05:00
committed by GitHub
parent 360a97fdd3
commit 9d2bbb1797
20 changed files with 81 additions and 953 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
cms/djangoapps/contentstore/tests/test_utils.py
View File

@@ -25,7 +25,7 @@ from cms.djangoapps.contentstore.utils import send_course_update_notification
from common.djangoapps.student.models import CourseEnrollment
from common.djangoapps.student.tests.factories import GlobalStaffFactory, InstructorFactory, UserFactory
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, Notification
from openedx.core.djangoapps.notifications.models import Notification
from openedx.core.djangoapps.site_configuration.tests.test_util import with_site_configuration_context
from xmodule.modulestore import ModuleStoreEnum # lint-amnesty, pylint: disable=wrong-import-order
from xmodule.modulestore.django import modulestore # lint-amnesty, pylint: disable=wrong-import-order
@@ -954,7 +954,6 @@ class CourseUpdateNotificationTests(OpenEdxEventsTestMixin, ModuleStoreTestCase)
super().setUp()
self.user = UserFactory()
self.course = CourseFactory.create(org='testorg', number='testcourse', run='testrun')
CourseNotificationPreference.objects.create(user_id=self.user.id, course_id=self.course.id)
def test_course_update_notification_sent(self):
"""

39
openedx/core/djangoapps/content/course_overviews/tests/test_serializers.py
View File

@@ -1,39 +0,0 @@
"""
CourseOverviewSerializer tests
"""
from django.test import TestCase
from openedx.core.djangoapps.content.course_overviews.serializers import CourseOverviewBaseSerializer
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from ..models import CourseOverview
class TestCourseOverviewSerializer(TestCase):
"""
TestCourseOverviewSerializer tests.
"""
def setUp(self):
super().setUp()
CourseOverviewFactory.create()
def test_get_course_overview_serializer(self):
"""
CourseOverviewBaseSerializer should add additional fields in the
to_representation method that is overridden.
"""
overview = CourseOverview.objects.first()
data = CourseOverviewBaseSerializer(overview).data
fields = [
'display_name_with_default',
'has_started',
'has_ended',
'pacing',
]
for field in fields:
assert field in data
assert isinstance(data['has_started'], bool)
assert isinstance(data['has_ended'], bool)

51
openedx/core/djangoapps/enrollments/tests/test_views.py
View File

@@ -40,7 +40,6 @@ from openedx.core.djangoapps.enrollments import api, data
from openedx.core.djangoapps.enrollments.errors import CourseEnrollmentError
from openedx.core.djangoapps.enrollments.views import EnrollmentUserThrottle
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference
from openedx.core.djangoapps.oauth_dispatch.jwt import create_jwt_for_user
from openedx.core.djangoapps.user_api.models import RetirementState, UserOrgTag, UserRetirementStatus
from openedx.core.djangolib.testing.utils import skip_unless_lms
@@ -56,20 +55,20 @@ class EnrollmentTestMixin:
API_KEY = "i am a key"
def assert_enrollment_status(
self,
course_id=None,
username=None,
expected_status=status.HTTP_200_OK,
email_opt_in=None,
as_server=False,
mode=CourseMode.DEFAULT_MODE_SLUG,
is_active=None,
enrollment_attributes=None,
min_mongo_calls=0,
max_mongo_calls=0,
linked_enterprise_customer=None,
cohort=None,
force_enrollment=False,
self,
course_id=None,
username=None,
expected_status=status.HTTP_200_OK,
email_opt_in=None,
as_server=False,
mode=CourseMode.DEFAULT_MODE_SLUG,
is_active=None,
enrollment_attributes=None,
min_mongo_calls=0,
max_mongo_calls=0,
linked_enterprise_customer=None,
cohort=None,
force_enrollment=False,
):
"""
Enroll in the course and verify the response's status code. If the expected status is 200, also validates
@@ -199,10 +198,6 @@ class EnrollmentTest(EnrollmentTestMixin, ModuleStoreTestCase, APITestCase, Ente
password=self.PASSWORD,
)
self.client.login(username=self.USERNAME, password=self.PASSWORD)
CourseNotificationPreference.objects.create(
user=self.user,
course_id=self.course.id,
)
@ddt.data(
# Default (no course modes in the database)
@@ -982,11 +977,11 @@ class EnrollmentTest(EnrollmentTestMixin, ModuleStoreTestCase, APITestCase, Ente
assert course_mode == CourseMode.DEFAULT_MODE_SLUG
@ddt.data(
((CourseMode.DEFAULT_MODE_SLUG, ), CourseMode.DEFAULT_MODE_SLUG),
((CourseMode.DEFAULT_MODE_SLUG,), CourseMode.DEFAULT_MODE_SLUG),
((CourseMode.DEFAULT_MODE_SLUG, CourseMode.VERIFIED), CourseMode.DEFAULT_MODE_SLUG),
((CourseMode.DEFAULT_MODE_SLUG, CourseMode.VERIFIED), CourseMode.VERIFIED),
((CourseMode.PROFESSIONAL, ), CourseMode.PROFESSIONAL),
((CourseMode.NO_ID_PROFESSIONAL_MODE, ), CourseMode.NO_ID_PROFESSIONAL_MODE),
((CourseMode.PROFESSIONAL,), CourseMode.PROFESSIONAL),
((CourseMode.NO_ID_PROFESSIONAL_MODE,), CourseMode.NO_ID_PROFESSIONAL_MODE),
((CourseMode.VERIFIED, CourseMode.CREDIT_MODE), CourseMode.VERIFIED),
((CourseMode.VERIFIED, CourseMode.CREDIT_MODE), CourseMode.CREDIT_MODE),
)
@@ -1274,7 +1269,7 @@ class EnrollmentTest(EnrollmentTestMixin, ModuleStoreTestCase, APITestCase, Ente
username='enterprise_worker',
linked_enterprise_customer='this-is-a-real-uuid',
)
assert httpretty.last_request().path == '/consent/api/v1/data_sharing_consent' # pylint: disable=no-member
assert httpretty.last_request().path == '/consent/api/v1/data_sharing_consent' # pylint: disable=no-member
assert httpretty.last_request().method == httpretty.POST
def test_enrollment_attributes_always_written(self):
@@ -1907,15 +1902,15 @@ class CourseEnrollmentsApiListTest(APITestCase, ModuleStoreTestCase):
@ddt.data(
# Non-existent user
({'username': 'nobody'}, ),
({'username': 'nobody', 'course_id': 'e/d/X'}, ),
({'username': 'nobody'},),
({'username': 'nobody', 'course_id': 'e/d/X'},),
# Non-existent course
({'course_id': 'a/b/c'}, ),
({'course_id': 'a/b/c', 'username': 'student1'}, ),
({'course_id': 'a/b/c'},),
({'course_id': 'a/b/c', 'username': 'student1'},),
# Non-existent course and user
({'course_id': 'a/b/c', 'username': 'dummy'}, )
({'course_id': 'a/b/c', 'username': 'dummy'},)
)
@ddt.unpack
def test_non_existent_course_user(self, query_params):

63
openedx/core/djangoapps/notifications/admin.py
View File

@@ -6,7 +6,7 @@ from django.contrib import admin
from django.utils.translation import gettext_lazy as _
from .base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES
from .models import CourseNotificationPreference, Notification
from .models import Notification
class NotificationAppNameListFilter(admin.SimpleListFilter):
@@ -19,13 +19,13 @@ class NotificationAppNameListFilter(admin.SimpleListFilter):
def lookups(self, request, model_admin):
lookup_list = [
(app_name, app_name)
for app_name in COURSE_NOTIFICATION_APPS.keys()
for app_name in COURSE_NOTIFICATION_APPS
]
return lookup_list
def queryset(self, request, queryset):
app_name = self.value()
if app_name not in COURSE_NOTIFICATION_APPS.keys():
if app_name not in COURSE_NOTIFICATION_APPS:
return queryset
return queryset.filter(app_name=app_name)
@@ -40,13 +40,13 @@ class NotificationTypeListFilter(admin.SimpleListFilter):
def lookups(self, request, model_admin):
lookup_list = [
(notification_type, notification_type)
for notification_type in COURSE_NOTIFICATION_TYPES.keys()
for notification_type in COURSE_NOTIFICATION_TYPES
]
return lookup_list
def queryset(self, request, queryset):
notification_type = self.value()
if notification_type not in COURSE_NOTIFICATION_TYPES.keys():
if notification_type not in COURSE_NOTIFICATION_TYPES:
return queryset
return queryset.filter(notification_type=notification_type)
@@ -60,57 +60,4 @@ class NotificationAdmin(admin.ModelAdmin):
list_filter = (NotificationAppNameListFilter, NotificationTypeListFilter)
class CourseNotificationPreferenceAdmin(admin.ModelAdmin):
"""
Admin for Course Notification Preferences
"""
model = CourseNotificationPreference
raw_id_fields = ('user',)
list_display = ('get_username', 'course_id')
search_fields = ('course_id', 'user__username')
search_help_text = _('Search by username, course_id. '
'Specify fields with username: or course_id: prefixes. '
'If no prefix is specified, search will be done on username. \n'
'Examples: \n'
' - testuser (default username search) \n'
' - username:testuser (username keyword search) \n'
' - course_id:course-v1:edX+DemoX+Demo_Course (course_id keyword search) \n'
' - username:testuser, course_id:course-v1:edX+DemoX+Demo_Course (combined keyword search) \n'
)
@admin.display(description='Username', ordering='user__username')
def get_username(self, obj):
return obj.user.username
def get_queryset(self, request):
queryset = super().get_queryset(request)
return queryset.select_related("user").only("id", "user__username", "course_id")
def get_search_results(self, request, queryset, search_term):
"""
Custom search for CourseNotificationPreference model
"""
if search_term:
criteria = search_term.split(',')
for criterion in criteria:
criterion = criterion.strip()
if criterion.startswith('username:'):
queryset = queryset.filter(user__username=criterion.split(':')[1])
elif criterion.startswith('course_id:'):
criteria = criterion.split(':')
course_id = ':'.join(criteria[1:]).strip()
queryset = queryset.filter(course_id=course_id)
else:
queryset = queryset.filter(user__username=search_term)
else:
queryset = queryset.all()
return queryset, True
admin.site.register(Notification, NotificationAdmin)
admin.site.register(CourseNotificationPreference, CourseNotificationPreferenceAdmin)

123
openedx/core/djangoapps/notifications/base_notification.py
View File

@@ -7,8 +7,9 @@ from django.utils.translation import gettext_lazy as _
from .email_notifications import EmailCadence
from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole
from .settings_override import get_notification_types_config, get_notification_apps_config
from .utils import find_app_in_normalized_apps, find_pref_in_normalized_prefs
from ..django_comment_common.models import FORUM_ROLE_ADMINISTRATOR, FORUM_ROLE_MODERATOR, FORUM_ROLE_COMMUNITY_TA
from .notification_content import get_notification_type_context_function
@@ -375,107 +376,6 @@ COURSE_NOTIFICATION_TYPES = get_notification_types_config()
COURSE_NOTIFICATION_APPS = get_notification_apps_config()
class NotificationPreferenceSyncManager:
"""
Sync Manager for Notification Preferences
"""
@staticmethod
def normalize_preferences(preferences):
"""
Normalizes preferences to reduce depth of structure.
This simplifies matching of preferences reducing effort to get difference.
"""
apps = []
prefs = []
non_editable = {}
core_notifications = {}
for app, app_pref in preferences.items():
apps.append({
'name': app,
'enabled': app_pref.get('enabled')
})
for pref_name, pref_values in app_pref.get('notification_types', {}).items():
prefs.append({
'name': pref_name,
'app_name': app,
**pref_values
})
non_editable[app] = app_pref.get('non_editable', {})
core_notifications[app] = app_pref.get('core_notification_types', [])
normalized_preferences = {
'apps': apps,
'preferences': prefs,
'non_editable': non_editable,
'core_notifications': core_notifications,
}
return normalized_preferences
@staticmethod
def denormalize_preferences(normalized_preferences):
"""
Denormalizes preference from simplified to normal structure for saving it in database
"""
denormalized_preferences = {}
for app in normalized_preferences.get('apps', []):
app_name = app.get('name')
app_toggle = app.get('enabled')
denormalized_preferences[app_name] = {
'enabled': app_toggle,
'core_notification_types': normalized_preferences.get('core_notifications', {}).get(app_name, []),
'notification_types': {},
'non_editable': normalized_preferences.get('non_editable', {}).get(app_name, {}),
}
for preference in normalized_preferences.get('preferences', []):
pref_name = preference.get('name')
app_name = preference.get('app_name')
denormalized_preferences[app_name]['notification_types'][pref_name] = {
'web': preference.get('web'),
'push': preference.get('push'),
'email': preference.get('email'),
'email_cadence': preference.get('email_cadence'),
}
return denormalized_preferences
@staticmethod
def update_preferences(preferences, email_opt_out=False):
"""
Creates a new preference version from old preferences.
New preference is created instead of updating old preference
Steps to update existing user preference
1) Normalize existing user preference
2) Normalize default preferences
3) Iterate over all the apps in default preference, if app_name exists in
existing preference, update new preference app enabled value as
existing enabled value
4) Iterate over all preferences, if preference_name exists in existing
preference, update new preference values of web, email and push as
existing web, email and push respectively
5) Denormalize new preference
"""
old_preferences = NotificationPreferenceSyncManager.normalize_preferences(preferences)
default_prefs = NotificationAppManager().get_notification_app_preferences(email_opt_out)
new_prefs = NotificationPreferenceSyncManager.normalize_preferences(default_prefs)
for app in new_prefs.get('apps'):
app_pref = find_app_in_normalized_apps(app.get('name'), old_preferences.get('apps'))
if app_pref:
app['enabled'] = app_pref['enabled']
for preference in new_prefs.get('preferences'):
pref_name = preference.get('name')
app_name = preference.get('app_name')
pref = find_pref_in_normalized_prefs(pref_name, app_name, old_preferences.get('preferences'))
if pref:
for channel in ['web', 'email', 'push', 'email_cadence']:
preference[channel] = pref.get(channel, preference.get(channel))
return NotificationPreferenceSyncManager.denormalize_preferences(new_prefs)
class NotificationTypeManager:
"""
Manager for notification types
@@ -511,18 +411,6 @@ class NotificationTypeManager:
non_core_notification_types.append(notification_type)
return core_notification_types, non_core_notification_types
@staticmethod
def get_non_editable_notification_channels(notification_types):
"""
Returns non_editable notification channels for the given notification types.
"""
non_editable_notification_channels = {}
for notification_type in notification_types:
if notification_type.get('non_editable', None):
non_editable_notification_channels[notification_type.get('name')] = \
notification_type.get('non_editable')
return non_editable_notification_channels
@staticmethod
def get_non_core_notification_type_preferences(non_core_notification_types, email_opt_out=False):
"""
@@ -568,13 +456,6 @@ class NotificationAppManager:
'email_cadence': notification_app_attrs.get('core_email_cadence', 'Daily'),
}
def add_core_notification_non_editable(self, notification_app_attrs, non_editable_channels):
"""
Adds non_editable for core notification.
"""
if notification_app_attrs.get('non_editable', None):
non_editable_channels['core'] = notification_app_attrs.get('non_editable')
def get_notification_app_preferences(self, email_opt_out=False):
"""
Returns notification app preferences for the given name.

2
openedx/core/djangoapps/notifications/email/events.py
View File

@@ -16,7 +16,7 @@ def send_user_email_digest_sent_event(user, cadence_type, notifications, message
"""
Sends tracker and segment email for user email digest
"""
notification_breakdown = {key: 0 for key in COURSE_NOTIFICATION_APPS.keys()}
notification_breakdown = {key: 0 for key in COURSE_NOTIFICATION_APPS}
for notification in notifications:
notification_breakdown[notification.app_name] += 1

66
openedx/core/djangoapps/notifications/email/tests/test_tasks.py
View File

@@ -20,7 +20,7 @@ from openedx.core.djangoapps.notifications.email.tasks import (
send_digest_email_to_user
)
from openedx.core.djangoapps.notifications.email.utils import get_start_end_date
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, NotificationPreference
from openedx.core.djangoapps.notifications.models import NotificationPreference
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
@@ -285,7 +285,7 @@ class TestEmailDigestAudience(ModuleStoreTestCase):
def test_audience_query_count(self):
with self.assertNumQueries(1):
audience = get_audience_for_cadence_email(EmailCadence.DAILY)
list(audience) # evaluating queryset
list(audience) # evaluating queryset
@ddt.data(True, False)
@patch('edx_ace.ace.send')
@@ -301,72 +301,12 @@ class TestEmailDigestAudience(ModuleStoreTestCase):
assert mock_func.called is email_value
@ddt.ddt
class TestPreferences(ModuleStoreTestCase):
"""
Tests preferences
"""
def setUp(self):
"""
Setup
"""
super().setUp()
self.user = UserFactory()
self.course = CourseFactory.create(display_name='test course', run="Testing_course")
self.preference = CourseNotificationPreference.objects.create(user=self.user, course_id=self.course.id)
created_date = datetime.datetime.now() - datetime.timedelta(hours=23)
create_notification(self.user, self.course.id, notification_type='new_discussion_post', created=created_date)
@patch('edx_ace.ace.send')
def test_email_send_for_digest_preference(self, mock_func):
"""
Tests email is send for digest notification preference
"""
start_date, end_date = get_start_end_date(EmailCadence.DAILY)
config = self.preference.notification_preference_config
types = config['discussion']['notification_types']
types['new_discussion_post']['email_cadence'] = EmailCadence.DAILY
types['new_discussion_post']['email'] = True
self.preference.save()
with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
send_digest_email_to_user(self.user, EmailCadence.DAILY, start_date, end_date)
assert not mock_func.called
@patch('edx_ace.ace.send')
def test_email_send_for_email_preference_value(self, mock_func):
"""
Tests email is sent iff preference value is True
"""
start_date, end_date = get_start_end_date(EmailCadence.DAILY)
config = self.preference.notification_preference_config
types = config['discussion']['notification_types']
types['new_discussion_post']['email_cadence'] = EmailCadence.DAILY
types['new_discussion_post']['email'] = True
self.preference.save()
with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
send_digest_email_to_user(self.user, EmailCadence.DAILY, start_date, end_date)
assert not mock_func.called
@patch('edx_ace.ace.send')
def test_email_not_send_if_different_digest_preference(self, mock_func):
"""
Tests email is not send if digest notification preference doesnot match
"""
start_date, end_date = get_start_end_date(EmailCadence.DAILY)
config = self.preference.notification_preference_config
types = config['discussion']['notification_types']
types['new_discussion_post']['email_cadence'] = EmailCadence.WEEKLY
self.preference.save()
with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
send_digest_email_to_user(self.user, EmailCadence.DAILY, start_date, end_date)
assert not mock_func.called
@ddt.ddt
class TestAccountPreferences(ModuleStoreTestCase):
"""
Tests preferences
"""
def setUp(self):
"""
Setup

29
openedx/core/djangoapps/notifications/email/tests/test_utils.py
View File

@@ -13,13 +13,10 @@ from pytz import utc
from waffle import get_waffle_flag_model # pylint: disable=invalid-django-waffle-import
from common.djangoapps.student.tests.factories import UserFactory
from openedx.core.djangoapps.notifications.base_notification import (
COURSE_NOTIFICATION_APPS,
COURSE_NOTIFICATION_TYPES,
)
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_EMAIL_NOTIFICATIONS
from openedx.core.djangoapps.notifications.email import ONE_CLICK_EMAIL_UNSUB_KEY
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, Notification
from openedx.core.djangoapps.notifications.models import Notification
from openedx.core.djangoapps.notifications.email.utils import (
add_additional_attributes_to_notifications,
create_app_notifications_dict,
@@ -275,6 +272,7 @@ class TestEncryption(ModuleStoreTestCase):
class TestUpdatePreferenceFromPatch(ModuleStoreTestCase):
"""
Tests if preferences are update according to patch data
this needs to be reimplemented as tests were removed in
"""
def setUp(self):
@@ -285,27 +283,6 @@ class TestUpdatePreferenceFromPatch(ModuleStoreTestCase):
self.user = UserFactory()
self.course_1 = CourseFactory.create(display_name='test course 1', run="Testing_course_1")
self.course_2 = CourseFactory.create(display_name='test course 2', run="Testing_course_2")
self.preference_1 = CourseNotificationPreference(course_id=self.course_1.id, user=self.user)
self.preference_2 = CourseNotificationPreference(course_id=self.course_2.id, user=self.user)
self.preference_1.save()
self.preference_2.save()
self.default_json = self.preference_1.notification_preference_config
def is_channel_editable(self, app_name, notification_type, channel):
"""
Returns if channel is editable
"""
if notification_type == 'core':
return channel not in COURSE_NOTIFICATION_APPS[app_name]['non_editable']
return channel not in COURSE_NOTIFICATION_TYPES[notification_type]['non_editable']
def get_default_cadence_value(self, app_name, notification_type):
"""
Returns default email cadence value
"""
if notification_type == 'core':
return COURSE_NOTIFICATION_APPS[app_name]['core_email_cadence']
return COURSE_NOTIFICATION_TYPES[notification_type]['email_cadence']
def test_preference_not_updated_if_invalid_username(self):
"""

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

@@ -45,11 +45,11 @@ def create_user_account_preferences(sender, instance, created, **kwargs): # pyl
if created:
try:
with transaction.atomic():
for name in COURSE_NOTIFICATION_TYPES.keys():
for name in COURSE_NOTIFICATION_TYPES:
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}')
log.info(f'Account-level Notification Preference 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.

2
openedx/core/djangoapps/notifications/management/tests/test_delete_expired_notifications.py
View File

@@ -9,7 +9,7 @@ from django.core.management import call_command
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
class GenerateCourseNotificationPreferencesTests(ModuleStoreTestCase):
class TestDeleteExpiredNotification(ModuleStoreTestCase):
"""
Tests for delete_expired_notifications management command.
"""

6
openedx/core/djangoapps/notifications/migrations/0004_auto_20230607_0757.py
View File

@@ -2,11 +2,9 @@
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
import django.utils.timezone
import model_utils.fields
import opaque_keys.edx.django.models
import openedx.core.djangoapps.notifications.models
class Migration(migrations.Migration):
@@ -24,8 +22,8 @@ class Migration(migrations.Migration):
('created', model_utils.fields.AutoCreatedField(default=django.utils.timezone.now, editable=False, verbose_name='created')),
('modified', model_utils.fields.AutoLastModifiedField(default=django.utils.timezone.now, editable=False, verbose_name='modified')),
('course_id', opaque_keys.edx.django.models.CourseKeyField(max_length=255)),
('notification_preference_config', models.JSONField(default=openedx.core.djangoapps.notifications.models.get_course_notification_preference_config)),
('config_version', models.IntegerField(default=openedx.core.djangoapps.notifications.models.get_course_notification_preference_config_version)),
('notification_preference_config', models.JSONField(default={})),
('config_version', models.IntegerField(default=0)),
('is_active', models.BooleanField(default=True)),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='notification_preferences', to=settings.AUTH_USER_MODEL)),
],

16
openedx/core/djangoapps/notifications/migrations/0010_delete_coursenotificationpreference.py Normal file
View File

@@ -0,0 +1,16 @@
# Generated by Django 5.2.7 on 2025-12-16 09:01
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('notifications', '0009_notification_push'),
]
operations = [
migrations.DeleteModel(
name='CourseNotificationPreference',
),
]

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

@@ -2,7 +2,6 @@
Models for notifications
"""
import logging
from typing import Dict
from django.contrib.auth import get_user_model
from django.db import models
@@ -10,15 +9,11 @@ from model_utils.models import TimeStampedModel
from opaque_keys.edx.django.models import CourseKeyField
from openedx.core.djangoapps.notifications.base_notification import (
NotificationAppManager,
NotificationPreferenceSyncManager,
get_notification_content,
COURSE_NOTIFICATION_APPS,
COURSE_NOTIFICATION_TYPES
)
from openedx.core.djangoapps.notifications.email import ONE_CLICK_EMAIL_UNSUB_KEY
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
from openedx.core.djangoapps.user_api.models import UserPreference
User = get_user_model()
log = logging.getLogger(__name__)
@@ -27,62 +22,6 @@ NOTIFICATION_CHANNELS = ['web', 'push', 'email']
ADDITIONAL_NOTIFICATION_CHANNEL_SETTINGS = ['email_cadence']
# Update this version when there is a change to any course specific notification type or app.
COURSE_NOTIFICATION_CONFIG_VERSION = 16
def get_course_notification_preference_config():
"""
Returns the course specific notification preference config.
Sample Response:
{
'discussion': {
'enabled': True,
'not_editable': {
'new_comment_on_post': ['push'],
'new_response_on_post': ['web'],
'new_response_on_comment': ['web', 'push']
},
'notification_types': {
'new_comment_on_post': {
'email': True,
'push': True,
'web': True,
'info': 'Comment on post'
},
'new_response_on_comment': {
'email': True,
'push': True,
'web': True,
'info': 'Response on comment'
},
'new_response_on_post': {
'email': True,
'push': True,
'web': True,
'info': 'New Response on Post'
},
'core': {
'email': True,
'push': True,
'web': True,
'info': 'comment on post and response on comment'
}
},
'core_notification_types': []
}
}
"""
return NotificationAppManager().get_notification_app_preferences()
def get_course_notification_preference_config_version():
"""
Returns the notification preference config version.
"""
return COURSE_NOTIFICATION_CONFIG_VERSION
def get_notification_channels():
"""
@@ -187,7 +126,7 @@ class NotificationPreference(TimeStampedModel):
is_active = models.BooleanField(default=True)
def __str__(self):
return f"{self.user_id} {self.type} (Web:{self.web}) (Push:{self.push})"\
return f"{self.user_id} {self.type} (Web:{self.web}) (Push:{self.push})" \
f"(Email:{self.email}, {self.email_cadence})"
@classmethod
@@ -236,200 +175,3 @@ class NotificationPreference(TimeStampedModel):
Returns the email cadence for the notification type.
"""
return self.email_cadence
class CourseNotificationPreference(TimeStampedModel):
"""
Model to store notification preferences for users
.. no_pii:
"""
user = models.ForeignKey(User, related_name="notification_preferences", on_delete=models.CASCADE)
course_id = CourseKeyField(max_length=255, null=False, blank=False)
notification_preference_config = models.JSONField(default=get_course_notification_preference_config)
# This version indicates the current version of this notification preference.
config_version = models.IntegerField(default=get_course_notification_preference_config_version)
is_active = models.BooleanField(default=True)
class Meta:
unique_together = ('user', 'course_id')
def __str__(self):
return f'{self.user.username} - {self.course_id}'
@staticmethod
def get_user_course_preference(user_id, course_id):
"""
Returns updated courses preferences for a user
"""
email_opt_out = False
try:
preferences = CourseNotificationPreference.objects.get(user_id=user_id, course_id=course_id, is_active=True)
except CourseNotificationPreference.DoesNotExist:
email_opt_out = UserPreference.objects.filter(user_id=user_id, key=ONE_CLICK_EMAIL_UNSUB_KEY).exists()
preferences = CourseNotificationPreference.objects.create(
user_id=user_id,
course_id=course_id,
is_active=True,
notification_preference_config=NotificationAppManager().get_notification_app_preferences(email_opt_out)
)
current_config_version = get_course_notification_preference_config_version()
if current_config_version != preferences.config_version:
try:
current_prefs = preferences.notification_preference_config
new_prefs = NotificationPreferenceSyncManager.update_preferences(current_prefs, email_opt_out)
preferences.config_version = current_config_version
preferences.notification_preference_config = new_prefs
preferences.save()
# pylint: disable-next=broad-except
except Exception as e:
log.error(f'Unable to update notification preference to new config. {e}')
return preferences
@staticmethod
def get_user_notification_preferences(user):
"""
Checks if all user preferences have updated versions and returns the user preferences.
Updates any preferences that need to be updated to the latest config version.
"""
preferences = CourseNotificationPreference.objects.filter(user=user, is_active=True)
email_opt_out = UserPreference.objects.filter(user_id=user.id, key=ONE_CLICK_EMAIL_UNSUB_KEY).exists()
current_config_version = get_course_notification_preference_config_version()
preferences_to_update = []
try:
for preference in preferences:
if preference.config_version != current_config_version:
current_prefs = preference.notification_preference_config
new_prefs = NotificationPreferenceSyncManager.update_preferences(current_prefs, email_opt_out)
preference.config_version = current_config_version
preference.notification_preference_config = new_prefs
preferences_to_update.append(preference)
if preferences_to_update:
CourseNotificationPreference.objects.bulk_update(
preferences_to_update,
['config_version', 'notification_preference_config']
)
except Exception as e: # pylint: disable=broad-exception-caught
log.error(f'Unable to update notification preference to new config: {str(e)}')
return preferences
@staticmethod
def get_updated_user_course_preferences(user, course_id):
return CourseNotificationPreference.get_user_course_preference(user.id, course_id)
def get_app_config(self, app_name) -> Dict:
"""
Returns the app config for the given app name.
"""
return self.notification_preference_config.get(app_name, {})
def get_notification_types(self, app_name) -> Dict:
"""
Returns the notification types for the given app name.
Sample Response:
{
'new_comment_on_post': {
'email': True,
'push': True,
'web': True,
'info': 'Comment on post'
},
'new_response_on_comment': {
'email': True,
'push': True,
'web': True,
'info': 'Response on comment'
},
"""
return self.get_app_config(app_name).get('notification_types', {})
def get_notification_type_config(self, app_name, notification_type) -> Dict:
"""
Returns the notification type config for the given app name and notification type.
Sample Response:
{
'email': True,
'push': True,
'web': True,
'info': 'Comment on post'
}
"""
return self.get_notification_types(app_name).get(notification_type, {})
def get_web_config(self, app_name, notification_type) -> bool:
"""
Returns the web config for the given app name and notification type.
"""
if self.is_core(app_name, notification_type):
return self.get_core_config(app_name).get('web', False)
return self.get_notification_type_config(app_name, notification_type).get('web', False)
def is_enabled_for_any_channel(self, app_name, notification_type) -> bool:
"""
Returns True if the notification type is enabled for any channel.
"""
if self.is_core(app_name, notification_type):
return any(self.get_core_config(app_name).get(channel, False) for channel in NOTIFICATION_CHANNELS)
return any(self.get_notification_type_config(app_name, notification_type).get(channel, False) for channel in
NOTIFICATION_CHANNELS)
def get_channels_for_notification_type(self, app_name, notification_type) -> list:
"""
Returns the channels for the given app name and notification type.
if notification is core then return according to core settings
Sample Response:
['web', 'push']
"""
if self.is_core(app_name, notification_type):
notification_channels = [channel for channel in NOTIFICATION_CHANNELS if
self.get_core_config(app_name).get(channel, False)]
additional_channel_settings = [channel for channel in ADDITIONAL_NOTIFICATION_CHANNEL_SETTINGS if
self.get_core_config(app_name).get(channel, False)]
else:
notification_channels = [channel for channel in NOTIFICATION_CHANNELS if
self.get_notification_type_config(app_name, notification_type).get(channel, False)]
additional_channel_settings = [channel for channel in ADDITIONAL_NOTIFICATION_CHANNEL_SETTINGS if
self.get_notification_type_config(app_name, notification_type).get(channel,
False)]
return notification_channels + additional_channel_settings
def is_core(self, app_name, notification_type) -> bool:
"""
Returns True if the given notification type is a core notification type.
"""
return notification_type in self.get_app_config(app_name).get('core_notification_types', [])
def get_core_config(self, app_name) -> Dict:
"""
Returns the core config for the given app name.
Sample Response:
{
'email': True,
'push': True,
'web': True,
'info': 'comment on post and response on comment'
}
"""
return self.get_notification_types(app_name).get('core', {})
def is_email_enabled_for_notification_type(self, app_name, notification_type) -> bool:
"""
Returns True if the email is enabled for the given app name and notification type.
"""
if self.is_core(app_name, notification_type):
return self.get_core_config(app_name).get('email', False)
return self.get_notification_type_config(app_name, notification_type).get('email', False)
def get_email_cadence_for_notification_type(self, app_name, notification_type) -> str:
"""
Returns the email cadence for the given app name and notification type.
"""
if self.is_core(app_name, notification_type):
return self.get_core_config(app_name).get('email_cadence', EmailCadence.NEVER)
return self.get_notification_type_config(app_name, notification_type).get('email_cadence', EmailCadence.NEVER)

23
openedx/core/djangoapps/notifications/policies.py
View File

@@ -2,9 +2,8 @@
from edx_ace.channel import ChannelType
from edx_ace.policy import Policy, PolicyResult
from opaque_keys.edx.keys import CourseKey
from .models import CourseNotificationPreference
from .models import NotificationPreference
class CoursePushNotificationOptout(Policy):
@@ -18,24 +17,16 @@ class CoursePushNotificationOptout(Policy):
:param message:
:return: PolicyResult
"""
course_ids = message.context.get('course_ids', [])
app_label = message.context.get('app_label')
if not (app_label or message.context.get('push_notification_extra_context', {})):
return PolicyResult(deny={ChannelType.PUSH})
course_keys = [CourseKey.from_string(course_id) for course_id in course_ids]
for course_key in course_keys:
course_notification_preference = CourseNotificationPreference.get_user_course_preference(
message.recipient.lms_user_id,
course_key
)
push_notification_preference = course_notification_preference.get_notification_type_config(
app_label,
notification_type='push',
).get('push', False)
if not push_notification_preference:
return PolicyResult(deny={ChannelType.PUSH})
notification_preference = NotificationPreference.objects.get_or_create(
user_id=message.recipient.lms_user_id,
app=app_label
)
if not notification_preference.push:
return PolicyResult(deny={ChannelType.PUSH})
return PolicyResult(deny=frozenset())

11
openedx/core/djangoapps/notifications/serializers.py
View File

@@ -5,7 +5,6 @@ Serializers for the notifications API.
from django.core.exceptions import ValidationError
from rest_framework import serializers
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.djangoapps.notifications.models import (
Notification,
get_additional_notification_channel_settings,
@@ -71,16 +70,6 @@ def add_info_to_notification_config(config_obj):
return config_obj
class CourseOverviewSerializer(serializers.ModelSerializer):
"""
Serializer for CourseOverview model.
"""
class Meta:
model = CourseOverview
fields = ('id', 'display_name')
class NotificationSerializer(serializers.ModelSerializer):
"""
Serializer for the Notification model.

208
openedx/core/djangoapps/notifications/tests/test_base_notification.py
View File

@@ -1,208 +1,8 @@
"""
Tests for base_notification
"""
from common.djangoapps.student.tests.factories import UserFactory
from openedx.core.djangoapps.notifications import base_notification, models
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
get_course_notification_preference_config_version
)
from openedx.core.djangoapps.notifications import base_notification
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
class NotificationPreferenceSyncManagerTest(ModuleStoreTestCase):
"""
Tests NotificationPreferenceSyncManager
"""
@classmethod
def setUpClass(cls):
"""
Overriding this method to save current config
"""
super(NotificationPreferenceSyncManagerTest, cls).setUpClass()
cls.current_apps = base_notification.COURSE_NOTIFICATION_APPS
cls.current_types = base_notification.COURSE_NOTIFICATION_TYPES
cls.current_config_version = models.COURSE_NOTIFICATION_CONFIG_VERSION
@classmethod
def tearDownClass(cls):
"""
Overriding this method to restore saved config
"""
super(NotificationPreferenceSyncManagerTest, cls).tearDownClass()
base_notification.COURSE_NOTIFICATION_APPS = cls.current_apps
base_notification.COURSE_NOTIFICATION_TYPES = cls.current_types
models.COURSE_NOTIFICATION_CONFIG_VERSION = cls.current_config_version
def setUp(self):
"""
Setup test cases
"""
super().setUp()
self.user = UserFactory()
self.course = CourseFactory.create(
org='testorg',
number='testcourse',
run='testrun'
)
self.default_app_name = "default_app"
self.default_app_value = self._create_notification_app()
self.default_type_name = "default_type"
self.default_type_value = self._create_notification_type(self.default_type_name)
self._set_course_notification_apps({self.default_app_name: self.default_app_value})
self._set_course_notification_types({self.default_type_name: self.default_type_value})
self._set_notification_config_version(1)
self.preference = CourseNotificationPreference(
user=self.user,
course_id=self.course.id,
)
def _set_course_notification_apps(self, apps):
"""
Set COURSE_NOTIFICATION_APPS
"""
base_notification.COURSE_NOTIFICATION_APPS = apps
def _set_course_notification_types(self, notifications_types):
"""
Set COURSE_NOTIFICATION_TYPES
"""
base_notification.COURSE_NOTIFICATION_TYPES = notifications_types
def _set_notification_config_version(self, config_version):
"""
Set COURSE_NOTIFICATION_CONFIG_VERSION
"""
models.COURSE_NOTIFICATION_CONFIG_VERSION = config_version
def _create_notification_app(self, overrides=None):
"""
Create a notification app
"""
notification_app = {
'enabled': True,
'core_info': '',
'core_web': True,
'core_email': True,
'core_push': True,
'core_email_cadence': 'Daily',
}
if overrides is not None:
notification_app.update(overrides)
return notification_app
def _create_notification_type(self, name, overrides=None):
"""
Creates a new notification type
"""
notification_type = {
'notification_app': self.default_app_name,
'name': name,
'is_core': False,
'web': True,
'email': True,
'push': True,
'email_cadence': 'Daily',
'info': '',
'non_editable': [],
'content_template': '',
'content_context': {},
'email_template': '',
}
if overrides is not None:
notification_type.update(overrides)
return notification_type
def test_app_addition_and_removal(self):
"""
Tests if new app is added/removed in existing preference
"""
current_config_version = get_course_notification_preference_config_version()
app_name = 'discussion'
new_app_value = self._create_notification_app()
self._set_notification_config_version(current_config_version + 1)
self._set_course_notification_apps({app_name: new_app_value})
new_config = CourseNotificationPreference.get_updated_user_course_preferences(self.user, self.course.id)
assert self.default_app_name not in new_config.notification_preference_config
assert app_name in new_config.notification_preference_config
def test_app_toggle_value_persist(self):
"""
Tests if app toggle value persists even if default is changed
"""
enabled_value = self.default_app_value['enabled']
config = CourseNotificationPreference.get_updated_user_course_preferences(self.user, self.course.id)
assert config.notification_preference_config[self.default_app_name]['enabled'] == enabled_value
base_notification.COURSE_NOTIFICATION_APPS[self.default_app_name]['enabled'] = not enabled_value
current_config_version = get_course_notification_preference_config_version()
self._set_notification_config_version(current_config_version + 1)
new_config = CourseNotificationPreference.get_updated_user_course_preferences(self.user, self.course.id)
assert new_config.config_version == current_config_version + 1
assert new_config.notification_preference_config[self.default_app_name]['enabled'] == enabled_value
def test_notification_type_addition_and_removal(self):
"""
Test if new notification type is added/removed in existing preferences
"""
current_config_version = get_course_notification_preference_config_version()
type_name = 'new_type'
new_type_value = self._create_notification_type(type_name)
self._set_notification_config_version(current_config_version + 1)
self._set_course_notification_types({
type_name: new_type_value
})
preferences = CourseNotificationPreference.get_updated_user_course_preferences(self.user, self.course.id)
new_config = preferences.notification_preference_config
assert type_name in new_config[self.default_app_name]['notification_types']
assert self.default_type_name not in new_config[self.default_app_name]['notification_types']
def test_notification_type_toggle_value_persist(self):
"""
Tests if notification type value persists if default is changed
"""
config = CourseNotificationPreference.get_updated_user_course_preferences(self.user, self.course.id)
preferences = config.notification_preference_config
preference_type = preferences[self.default_app_name]['notification_types'][self.default_type_name]
web_value = preference_type['web']
email_value = preference_type['email']
push_value = preference_type['push']
base_notification.COURSE_NOTIFICATION_TYPES[self.default_type_name].update({
'web': not web_value,
'email': not email_value,
'push': not push_value,
})
current_config_version = get_course_notification_preference_config_version()
self._set_notification_config_version(current_config_version + 1)
new_config = CourseNotificationPreference.get_updated_user_course_preferences(self.user, self.course.id)
preferences = new_config.notification_preference_config
preference_type = preferences[self.default_app_name]['notification_types'][self.default_type_name]
assert new_config.config_version == current_config_version + 1
assert preference_type['web'] == web_value
assert preference_type['email'] == email_value
assert preference_type['push'] == push_value
def test_notification_type_in_core(self):
"""
Tests addition/removal of core in notification type
"""
current_config_version = get_course_notification_preference_config_version()
base_notification.COURSE_NOTIFICATION_TYPES[self.default_type_name]['is_core'] = True
self._set_notification_config_version(current_config_version + 1)
new_config = CourseNotificationPreference.get_updated_user_course_preferences(self.user, self.course.id)
preferences = new_config.notification_preference_config
core_notifications = preferences[self.default_app_name]['core_notification_types']
assert self.default_type_name in core_notifications
base_notification.COURSE_NOTIFICATION_TYPES[self.default_type_name]['is_core'] = False
self._set_notification_config_version(current_config_version + 2)
new_config = CourseNotificationPreference.get_updated_user_course_preferences(self.user, self.course.id)
preferences = new_config.notification_preference_config
core_notifications = preferences[self.default_app_name]['core_notification_types']
assert self.default_type_name not in core_notifications
class NotificationPreferenceValidationTest(ModuleStoreTestCase):
@@ -217,7 +17,7 @@ class NotificationPreferenceValidationTest(ModuleStoreTestCase):
"""
bool_keys = ['enabled', 'core_web', 'core_push', 'core_email']
notification_apps = base_notification.COURSE_NOTIFICATION_APPS
assert "" not in notification_apps.keys()
assert "" not in notification_apps
for app_data in notification_apps.values():
assert 'core_info' in app_data.keys()
assert isinstance(app_data['non_editable'], list)
@@ -232,7 +32,7 @@ class NotificationPreferenceValidationTest(ModuleStoreTestCase):
"""
str_keys = ['notification_app', 'name', 'email_template']
notification_types = base_notification.COURSE_NOTIFICATION_TYPES
assert "" not in notification_types.keys()
assert "" not in notification_types
for notification_type in notification_types.values():
if not notification_type['is_core']:
continue
@@ -250,7 +50,7 @@ class NotificationPreferenceValidationTest(ModuleStoreTestCase):
str_keys = ['notification_app', 'name', 'info', 'email_template']
bool_keys = ['is_core', 'web', 'email', 'push']
notification_types = base_notification.COURSE_NOTIFICATION_TYPES
assert "" not in notification_types.keys()
assert "" not in notification_types
for notification_type in notification_types.values():
if notification_type['is_core']:
continue

44
openedx/core/djangoapps/notifications/tests/test_models.py
View File

@@ -1,44 +0,0 @@
"""
Test the notification app models
"""
import unittest
from unittest import mock
import pytest
from common.djangoapps.student.tests.factories import UserFactory
from openedx.core.djangoapps.notifications.base_notification import NotificationAppManager
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, \
COURSE_NOTIFICATION_CONFIG_VERSION
@pytest.mark.django_db
class TestPreferenceModel(unittest.TestCase):
"""
Test the CourseNotificationPreference model.
"""
def test_get_user_notification_preferences_method(self):
"""
Test the get_user_notification_preferences method. and check if version is updated properly.
"""
# Create a mock user and notification preference
user = UserFactory()
CourseNotificationPreference.objects.create(
user_id=user.id,
course_id='course-v1:edX+DemoX+Demo_Course',
is_active=True,
notification_preference_config=NotificationAppManager().get_notification_app_preferences(True)
)
# Check if the notification preference is created
preference = CourseNotificationPreference.objects.get(user_id=user.id)
self.assertIsNotNone(preference)
self.assertTrue(preference.is_active)
with mock.patch(
'openedx.core.djangoapps.notifications.models.COURSE_NOTIFICATION_CONFIG_VERSION',
COURSE_NOTIFICATION_CONFIG_VERSION + 1
):
updated_preferences = preference.get_user_notification_preferences(user)
for updated_preference in updated_preferences:
assert updated_preference.config_version == COURSE_NOTIFICATION_CONFIG_VERSION + 1

60
openedx/core/djangoapps/notifications/tests/test_tasks_with_account_level_pref.py
View File

@@ -16,7 +16,7 @@ from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
from ..config.waffle import ENABLE_NOTIFICATIONS
from ..models import CourseNotificationPreference, Notification, NotificationPreference
from ..models import Notification, NotificationPreference
from ..tasks import (
delete_notifications,
send_notifications,
@@ -43,12 +43,6 @@ class SendNotificationsTest(ModuleStoreTestCase):
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
@@ -114,12 +108,6 @@ class SendNotificationsTest(ModuleStoreTestCase):
}
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,
@@ -138,11 +126,6 @@ class SendNotificationsTest(ModuleStoreTestCase):
"""
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,
@@ -239,16 +222,6 @@ class SendBatchNotificationsTest(ModuleStoreTestCase):
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,
@@ -292,34 +265,22 @@ class SendBatchNotificationsTest(ModuleStoreTestCase):
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),
("new_response", 2),
("new_response", 2),
("new_response", 2),
("new_discussion_post", 0),
("new_discussion_post", 0),
("new_discussion_post", 0),
)
@ddt.unpack
def test_preference_enabled_in_batch_audience(self, notification_type,
user_1_pref_exists, user_2_pref_exists, generated_count):
def test_preference_enabled_in_batch_audience(self, notification_type, 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 = {
@@ -417,11 +378,6 @@ class NotificationCreationOnChannelsTests(ModuleStoreTestCase):
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',

4
openedx/core/djangoapps/notifications/tests/test_views.py
View File

@@ -28,7 +28,8 @@ from openedx.core.djangoapps.django_comment_common.models import (
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx.core.djangoapps.notifications.email.utils import encrypt_string
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference, Notification, NotificationPreference
Notification,
NotificationPreference
)
from openedx.core.djangoapps.notifications.serializers import (
add_info_to_notification_config,
@@ -491,7 +492,6 @@ class UpdatePreferenceFromEncryptedDataView(ModuleStoreTestCase):
self.user = UserFactory(password=password)
self.client.login(username=self.user.username, password=password)
self.course = CourseFactory.create(display_name='test course 1', run="Testing_course_1")
CourseNotificationPreference(course_id=self.course.id, user=self.user).save()
@override_settings(LMS_BASE="example.com", ONE_CLICK_UNSUBSCRIBE_RATE_LIMIT='1/d')
def test_rate_limit_on_unsub(self):

20
openedx/core/djangoapps/notifications/utils.py
View File

@@ -9,26 +9,6 @@ from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICAT
from openedx.core.lib.cache_utils import request_cached
def find_app_in_normalized_apps(app_name, apps_list):
"""
Returns app preference based on app_name
"""
for app in apps_list:
if app.get('name') == app_name:
return app
return None
def find_pref_in_normalized_prefs(pref_name, app_name, prefs_list):
"""
Returns preference based on preference_name and app_name
"""
for pref in prefs_list:
if pref.get('name') == pref_name and pref.get('app_name') == app_name:
return pref
return None
def get_show_notifications_tray():
"""
Returns whether notifications tray is enabled via waffle flag