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

feat: added parser to update existing notification on get (#32450)

Browse Source
This commit is contained in:
Muhammad Adeel Tajamul
2023-06-20 14:10:21 +05:00
committed by GitHub
parent 4c36156c8c
commit aab3624960
6 changed files with 417 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,7 +1,11 @@
"""
Base setup for Notification Apps and Types.
"""
from typing import Dict
from .utils import (
find_app_in_normalized_apps,
find_pref_in_normalized_prefs,
)
COURSE_NOTIFICATION_TYPES = {
'new_comment_on_response': {
@@ -65,11 +69,111 @@ COURSE_NOTIFICATION_APPS = {
}
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'),
'info': preference.get('info'),
}
return denormalized_preferences
@staticmethod
def update_preferences(preferences):
"""
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()
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']:
preference[channel] = pref[channel]
return NotificationPreferenceSyncManager.denormalize_preferences(new_prefs)
class NotificationTypeManager:
"""
Manager for notification types
"""
notification_types: Dict = {}
def __init__(self):
self.notification_types = COURSE_NOTIFICATION_TYPES
@@ -143,10 +247,6 @@ class NotificationAppManager:
"""
Notification app manager
"""
notification_apps: Dict = {}
def __init__(self):
self.notification_apps = COURSE_NOTIFICATION_APPS
def add_core_notification_preference(self, notification_app_attrs, notification_types):
"""
@@ -175,9 +275,7 @@ class NotificationAppManager:
notification_app_preferences['notification_types'] = notification_types
notification_app_preferences['non_editable'] = non_editable_channels
course_notification_preference_config[notification_app_key] = notification_app_preferences
return course_notification_preference_config
return None
return course_notification_preference_config
def get_notification_content(notification_type, context):

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

@@ -3,13 +3,16 @@ Handlers for notifications
"""
import logging
from django.core.exceptions import ObjectDoesNotExist
from django.db import IntegrityError
from django.db.models.signals import post_save
from django.dispatch import receiver
from openedx_events.learning.signals import COURSE_UNENROLLMENT_COMPLETED
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference
log = logging.getLogger(__name__)
@@ -25,3 +28,17 @@ def course_enrollment_post_save(sender, instance, created, **kwargs):
except IntegrityError:
log.info(f'CourseNotificationPreference already exists for user {instance.user} '
f'and course {instance.course_id}')
@receiver(COURSE_UNENROLLMENT_COMPLETED)
def on_user_course_unenrollment(enrollment, **kwargs):
"""
Removes user notification preference when user un-enrolls from the course
"""
try:
user_id = enrollment.user.id
course_key = enrollment.course.course_key
preference = CourseNotificationPreference.objects.get(user__id=user_id, course_id=course_key)
preference.delete()
except ObjectDoesNotExist:
log.info(f'Notification Preference doesnot exist for {enrollment.user.pii.username} in {course_key}')

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

@@ -1,14 +1,22 @@
"""
Models for notifications
"""
import logging
from django.contrib.auth import get_user_model
from django.db import models
from model_utils.models import TimeStampedModel
from opaque_keys.edx.django.models import CourseKeyField
from openedx.core.djangoapps.notifications.base_notification import NotificationAppManager, get_notification_content
from openedx.core.djangoapps.notifications.base_notification import (
NotificationAppManager,
NotificationPreferenceSyncManager,
get_notification_content,
)
User = get_user_model()
log = logging.getLogger(__name__)
NOTIFICATION_CHANNELS = ['web', 'push', 'email']
@@ -120,3 +128,26 @@ class CourseNotificationPreference(TimeStampedModel):
def __str__(self):
return f'{self.user.username} - {self.course_id} - {self.notification_preference_config}'
@staticmethod
def get_updated_user_course_preferences(user, course_id):
"""
Returns updated courses preferences for a user
"""
preferences, _ = CourseNotificationPreference.objects.get_or_create(
user=user,
course_id=course_id,
is_active=True,
)
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)
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 for {user.username} to new config. {e}')
return preferences

235
openedx/core/djangoapps/notifications/tests/test_base_notification.py Normal file
View File

@@ -0,0 +1,235 @@
"""
Tests for base_notification
"""
from common.djangoapps.student.tests.factories import UserFactory
from openedx.core.djangoapps.notifications import base_notification
from openedx.core.djangoapps.notifications import models
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
get_course_notification_preference_config_version,
)
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,
}
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,
'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_non_editable_addition_and_removal(self):
"""
Tests if non-editable updates on existing preferences
"""
current_config_version = get_course_notification_preference_config_version()
base_notification.COURSE_NOTIFICATION_TYPES[self.default_type_name]['non-editable'] = ['web']
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_non_editable = preferences[self.default_app_name]['non_editable'][self.default_type_name]
assert 'web' in preference_non_editable
base_notification.COURSE_NOTIFICATION_TYPES[self.default_type_name]['non-editable'] = []
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
preference_non_editable = preferences[self.default_app_name]['non_editable'].get(self.default_type_name, [])
assert preference_non_editable == []
def test_notification_type_info_updates(self):
"""
Preference info updates when default info is update
"""
current_config_version = get_course_notification_preference_config_version()
new_info = "NEW INFO"
base_notification.COURSE_NOTIFICATION_TYPES[self.default_type_name]['info'] = new_info
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
notification_type = preferences[self.default_app_name]['notification_types'][self.default_type_name]
assert notification_type['info'] == new_info
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

23
openedx/core/djangoapps/notifications/utils.py Normal file
View File

@@ -0,0 +1,23 @@
"""
Utils function for notifications app
"""
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

10
openedx/core/djangoapps/notifications/views.py
View File

@@ -17,7 +17,7 @@ from rest_framework.views import APIView
from common.djangoapps.student.models import CourseEnrollment
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
get_course_notification_preference_config_version
get_course_notification_preference_config_version,
)
from .base_notification import COURSE_NOTIFICATION_APPS
@@ -147,12 +147,8 @@ class UserNotificationPreferenceView(APIView):
}
"""
course_id = CourseKey.from_string(course_key_string)
user_notification_preference, _ = CourseNotificationPreference.objects.get_or_create(
user=request.user,
course_id=course_id,
is_active=True,
)
serializer = UserCourseNotificationPreferenceSerializer(user_notification_preference)
user_preference = CourseNotificationPreference.get_updated_user_course_preferences(request.user, course_id)
serializer = UserCourseNotificationPreferenceSerializer(user_preference)
return Response(serializer.data)
def patch(self, request, course_key_string):