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

chore: cleaned up course level preferences (#37106)

Browse Source 
* chore: cleaned up course level preferences

* fix: fixed import issue

* fix: fixed lint errors
This commit is contained in:
Eemaan Amir
2025-08-04 12:20:54 +05:00
committed by GitHub
parent d59f5b39fe
commit 0b4c75c21c
16 changed files with 15 additions and 3469 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -39,7 +39,7 @@ from openedx.core.djangoapps.embargo.test_utils import restrict_course
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.handlers import ENABLE_NOTIFICATIONS
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

25
openedx/core/djangoapps/notifications/email/tasks.py
View File

@@ -12,10 +12,8 @@ from edx_django_utils.monitoring import set_code_owner_attribute
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
Notification,
NotificationPreference,
get_course_notification_preference_config_version
)
from .events import send_immediate_email_digest_sent_event, send_user_email_digest_sent_event
from .message_type import EmailNotificationMessageType
@@ -53,29 +51,6 @@ def get_audience_for_cadence_email(cadence_type):
return users
def get_user_preferences_for_courses(course_ids, user):
"""
Returns updated user preference for course_ids
"""
# Create new preferences
new_preferences = []
preferences = CourseNotificationPreference.objects.filter(user=user, course_id__in=course_ids)
preferences = list(preferences)
for course_id in course_ids:
if not any(preference.course_id == course_id for preference in preferences):
pref = CourseNotificationPreference(user=user, course_id=course_id)
new_preferences.append(pref)
if new_preferences:
CourseNotificationPreference.objects.bulk_create(new_preferences, ignore_conflicts=True)
# Update preferences to latest config version
current_version = get_course_notification_preference_config_version()
for preference in preferences:
if preference.config_version != current_version:
preference = preference.get_user_course_preference(user.id, preference.course_id)
new_preferences.append(preference)
return new_preferences
def send_digest_email_to_user(user, cadence_type, start_date, end_date, user_language='en', courses_data=None):
"""
Send [cadence_type] email to user.

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

@@ -4,14 +4,11 @@ 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, ProgrammingError
from django.db.models.signals import post_save
from django.dispatch import receiver
from openedx_events.learning.signals import (
COURSE_ENROLLMENT_CREATED,
COURSE_NOTIFICATION_REQUESTED,
COURSE_UNENROLLMENT_COMPLETED,
USER_NOTIFICATION_REQUESTED
)
@@ -23,12 +20,9 @@ from openedx.core.djangoapps.notifications.audience_filters import (
ForumRoleAudienceFilter,
TeamAudienceFilter
)
from openedx.core.djangoapps.notifications.base_notification import NotificationAppManager, COURSE_NOTIFICATION_TYPES
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx.core.djangoapps.notifications.email import ONE_CLICK_EMAIL_UNSUB_KEY
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, NotificationPreference
from openedx.core.djangoapps.notifications.base_notification import COURSE_NOTIFICATION_TYPES
from openedx.core.djangoapps.notifications.models import 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__)
@@ -42,31 +36,6 @@ AUDIENCE_FILTER_CLASSES = {
}
@receiver(COURSE_ENROLLMENT_CREATED)
def course_enrollment_post_save(signal, sender, enrollment, metadata, **kwargs):
"""
Watches for post_save signal for creates on the CourseEnrollment table.
Generate a CourseNotificationPreference if new Enrollment is created
"""
if ENABLE_NOTIFICATIONS.is_enabled(enrollment.course.course_key):
try:
with transaction.atomic():
email_opt_out = UserPreference.objects.filter(
user_id=enrollment.user.id,
key=ONE_CLICK_EMAIL_UNSUB_KEY
).exists()
CourseNotificationPreference.objects.create(
user_id=enrollment.user.id,
course_id=enrollment.course.course_key,
notification_preference_config=NotificationAppManager().get_notification_app_preferences(
email_opt_out
)
)
except IntegrityError:
log.info(f'CourseNotificationPreference already exists for user {enrollment.user.id} '
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
"""
@@ -88,20 +57,6 @@ def create_user_account_preferences(sender, instance, created, **kwargs): # pyl
log.error(f'ProgrammingError encountered while creating user preferences: {e}')
@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 does not exist for {enrollment.user.pii.username} in {course_key}')
@receiver(USER_NOTIFICATION_REQUESTED)
def generate_user_notifications(signal, sender, notification_data, metadata, **kwargs):
"""

34
openedx/core/djangoapps/notifications/management/commands/generate_course_notification_preferences.py
View File

@@ -1,34 +0,0 @@
"""
Management command for creating Course Notification Preferences for users in course.
"""
import logging
from django.core.management.base import BaseCommand
from openedx.core.djangoapps.notifications.tasks import create_course_notification_preferences_for_courses
logger = logging.getLogger(__name__)
class Command(BaseCommand):
"""
Invoke with:
python manage.py [lms|cms] generate_course_notification_preferences [course_id] [course_id] ...
"""
help = (
"Back-fill missing course notification preferences. This will queue a celery task to "
"create course notification preferences for all active enrollments in the courses provided."
)
def add_arguments(self, parser):
parser.add_argument(
'course_ids',
help='course_ids seperated by space for which to create Course Notification Preferences.',
nargs='*'
)
def handle(self, *args, **options):
course_ids = options['course_ids']
create_course_notification_preferences_for_courses.delay(course_ids)

325
openedx/core/djangoapps/notifications/management/commands/migrate_preferences_to_account_level_model.py
View File

@@ -1,325 +0,0 @@
"""
Command to migrate course-level notification preferences to account-level preferences.
"""
import gc
import logging
from typing import Dict, List, Any, Iterator
from collections import defaultdict
from django.core.management.base import BaseCommand, CommandParser
from django.db import transaction
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, NotificationPreference
from openedx.core.djangoapps.notifications.utils import aggregate_notification_configs
from openedx.core.djangoapps.notifications.base_notification import NotificationTypeManager, COURSE_NOTIFICATION_APPS
logger = logging.getLogger(__name__)
DEFAULT_BATCH_SIZE = 1000
class Command(BaseCommand):
"""
Migrates course-level notification preferences to account-level notification preferences.
This command processes users in batches, aggregates their course-level preferences,
and creates new account-level preferences. It includes a dry-run mode.
Existing account-level preferences for a processed user will be deleted before
new ones are created to ensure idempotency.
"""
help = "Migrates course-level notification preferences to account-level preferences for all relevant users."
def add_arguments(self, parser: CommandParser):
parser.add_argument(
'--batch-size',
type=int,
default=DEFAULT_BATCH_SIZE,
help=f"The number of users to process in each batch. Default: {DEFAULT_BATCH_SIZE}"
)
parser.add_argument(
'--dry-run',
action='store_true',
help="Simulate the migration without making any database changes."
)
parser.add_argument(
'--use-default',
nargs='+',
choices=['web', 'push', 'email', 'email_cadence'],
help="Specify which notification channels should use default values. Can accept multiple values"
" (e.g., --use-default web push email)."
)
@staticmethod
def _run_garbage_collection():
"""
Run manual garbage collection
"""
try:
collected_objects = gc.collect()
logger.debug(f"Garbage collection freed {collected_objects} objects")
return collected_objects
except Exception as e: # pylint: disable=broad-except
logger.warning(f"Garbage collection failed: {e}")
return 0
@staticmethod
def _get_user_ids_to_process() -> Iterator[int]:
"""
Yields all distinct user IDs with course notification preferences.
"""
logger.info("Fetching all distinct user IDs with course notification preferences...")
user_id_queryset = (CourseNotificationPreference
.objects
.values_list('user_id', flat=True)
.distinct())
# The iterator with chunk_size is memory efficient for fetching the IDs themselves.
yield from user_id_queryset.iterator()
@staticmethod
def _create_preference_object(
user_id: int,
app_name: str,
notification_type: str,
values: Dict[str, Any],
use_default: List[str] = None
) -> NotificationPreference:
"""
Helper function to create a NotificationPreference instance.
Args:
user_id: The user ID for whom the preference is being created
app_name: The name of the notification app
notification_type: The type of notification (e.g., 'assignment', 'discussion')
values: A dictionary containing the preference values for web, email, push, etc.
use_default: List of channels that should use default values
"""
if use_default:
non_core_defaults, core_defaults = NotificationTypeManager().get_notification_app_preference(app_name)
if non_core_defaults and notification_type in non_core_defaults:
for default in use_default:
values[default] = non_core_defaults[notification_type][default]
elif core_defaults and notification_type in core_defaults:
for default in use_default:
values[default] = COURSE_NOTIFICATION_APPS[app_name][f'core_{default}']
return NotificationPreference(
user_id=user_id,
app=app_name,
type=notification_type,
web=values.get('web'),
email=values.get('email'),
push=values.get('push'),
email_cadence=values.get('email_cadence', EmailCadence.DAILY)
)
def _create_preferences_from_configs(
self,
user_id: int,
course_preferences_configs: List[Dict],
use_default: List[str] = None
) -> List[NotificationPreference]:
"""
Processes a list of preference configs for a single user.
Returns a list of NotificationPreference objects to be created.
Args:
user_id: The user ID to process preferences for
course_preferences_configs: List of preference configuration dictionaries
use_default: List of channels ('web', 'push', 'email') that should use default values
"""
new_account_preferences: List[NotificationPreference] = []
use_default = use_default or []
if not course_preferences_configs:
logger.debug(f"No course preferences found for user {user_id}. Skipping.")
return new_account_preferences
aggregated_data = aggregate_notification_configs(course_preferences_configs)
for app_name, app_config in aggregated_data.items():
if not isinstance(app_config, dict):
logger.warning(
f"Malformed app_config for app '{app_name}' for user {user_id}. "
f"Expected dict, got {type(app_config)}. Skipping app."
)
continue
notif_types = app_config.get('notification_types', {})
if not isinstance(notif_types, dict):
logger.warning(
f"Malformed 'notification_types' for app '{app_name}' for user {user_id}. Expected dict, "
f"got {type(notif_types)}. Skipping notification_types."
)
continue
# Handle regular notification types
for notification_type, values in notif_types.items():
if notification_type == 'core':
continue
if values is None or not isinstance(values, dict):
logger.warning(
f"Skipping malformed notification type data for '{notification_type}' "
f"in app '{app_name}' for user {user_id}."
)
continue
new_account_preferences.append(
self._create_preference_object(user_id, app_name, notification_type, values, use_default)
)
# Handle core notification types
core_types_list = app_config.get('core_notification_types', [])
if not isinstance(core_types_list, list):
logger.warning(
f"Malformed 'core_notification_types' for app '{app_name}' for user {user_id}. "
f"Expected list, got {type(core_types_list)}. Skipping core_notification_types."
)
continue
core_values = notif_types.get('core', {})
if not isinstance(core_values, dict):
logger.warning(
f"Malformed values for 'core' notification types in app '{app_name}' for user {user_id}. "
f"Expected dict, got {type(core_values)}. Using empty defaults."
)
core_values = {}
for core_type_name in core_types_list:
if core_type_name is None or not isinstance(core_type_name, str):
logger.warning(
f"Skipping malformed core_type_name: '{core_type_name}' in app '{app_name}' for user {user_id}."
)
continue
new_account_preferences.append(
self._create_preference_object(user_id, app_name, core_type_name, core_values, use_default)
)
return new_account_preferences
def _process_batch(self, user_ids: List[int], use_default: List[str] = None) -> List[NotificationPreference]:
"""
Fetches all preferences for a batch of users and processes them.
Args:
user_ids: List of user IDs to process
use_default: List of channels that should use default values
"""
all_new_preferences: List[NotificationPreference] = []
# 1. Fetch all preference data for the batch in a single query.
course_prefs = CourseNotificationPreference.objects.filter(
user_id__in=user_ids
).values('user_id', 'notification_preference_config')
# 2. Group the fetched data by user_id in memory.
prefs_by_user = defaultdict(list)
for pref in course_prefs:
prefs_by_user[pref['user_id']].append(pref['notification_preference_config'])
# 3. Process each user's grouped data.
for user_id, configs in prefs_by_user.items():
user_new_preferences = self._create_preferences_from_configs(user_id, configs, use_default)
if user_new_preferences:
all_new_preferences.extend(user_new_preferences)
logger.debug(f"User {user_id}: Aggregated {len(configs)} course preferences "
f"into {len(user_new_preferences)} account preferences.")
else:
logger.debug(f"User {user_id}: No account preferences generated from {len(configs)} "
f"course preferences.")
# Clear local references to help with garbage collection
del prefs_by_user
del course_prefs
return all_new_preferences
def handle(self, *args: Any, **options: Any): # pylint: disable=too-many-statements
dry_run = options['dry_run']
batch_size = options['batch_size']
use_default = options.get('use_default', [])
if dry_run:
logger.info(self.style.WARNING("Performing a DRY RUN. No changes will be made to the database."))
else:
# Clear all existing preferences once at the beginning.
# This is more efficient and safer than deleting per-user.
NotificationPreference.objects.all().delete()
logger.info('Cleared all existing account-level notification preferences.')
if use_default:
logger.info(f"Using default values for channels: {', '.join(use_default)}")
self._run_garbage_collection()
user_id_iterator = self._get_user_ids_to_process()
user_id_batch: List[int] = []
total_users_processed = 0
total_preferences_created = 0
for user_id in user_id_iterator:
user_id_batch.append(user_id)
if len(user_id_batch) >= batch_size:
try:
with transaction.atomic():
# Process the entire batch of users
preferences_to_create = self._process_batch(user_id_batch, use_default)
if preferences_to_create:
if not dry_run:
NotificationPreference.objects.bulk_create(preferences_to_create)
total_preferences_created += len(preferences_to_create)
logger.info(
self.style.SUCCESS(
f"Batch complete. {'Would create' if dry_run else 'Created'} "
f"{len(preferences_to_create)} preferences for {len(user_id_batch)} users."
)
)
else:
logger.info(f"Batch complete. No preferences to create for {len(user_id_batch)} users.")
total_users_processed += len(user_id_batch)
user_id_batch = [] # Reset the batch
user_id_batch.clear()
del preferences_to_create
except Exception as e: # pylint: disable=broad-except
logger.error(f"Failed to process batch containing users {user_id_batch}: {e}", exc_info=True)
# The transaction for the whole batch will be rolled back.
# Clear the batch to continue with the next set of users.
user_id_batch = []
if total_users_processed > 0 and total_users_processed % (batch_size * 5) == 0:
logger.info(f"PROGRESS: Total users processed so far: {total_users_processed}. "
f"Total preferences {'would be' if dry_run else ''} "
f"created: {total_preferences_created}")
# Process any remaining users in the last, smaller batch
if user_id_batch:
try:
with transaction.atomic():
preferences_to_create = self._process_batch(user_id_batch, use_default)
if preferences_to_create:
if not dry_run:
NotificationPreference.objects.bulk_create(preferences_to_create)
total_preferences_created += len(preferences_to_create)
logger.info(
self.style.SUCCESS(
f"Final batch complete. {'Would create' if dry_run else 'Created'} "
f"{len(preferences_to_create)} preferences for {len(user_id_batch)} users."
)
)
total_users_processed += len(user_id_batch)
del preferences_to_create
self._run_garbage_collection()
except Exception as e: # pylint: disable=broad-except
logger.error(f"Failed to process final batch of users {user_id_batch}: {e}", exc_info=True)
logger.info(
self.style.SUCCESS(
f"Migration complete. Processed {total_users_processed} users. "
f"{'Would have created' if dry_run else 'Created'} a total of {total_preferences_created} "
f"account-level preferences."
)
)
self._run_garbage_collection()
if dry_run:
logger.info(self.style.WARNING("DRY RUN finished. No actual changes were made."))

32
openedx/core/djangoapps/notifications/management/tests/test_generate_course_notification_preferences.py
View File

@@ -1,32 +0,0 @@
"""
Test for generate_notification_preferences management command.
"""
from unittest import mock
from django.core.management import call_command
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
class GenerateCourseNotificationPreferencesTests(ModuleStoreTestCase):
"""
Tests for generate_course_notification_preferences management command.
"""
def setUp(self):
super().setUp()
self.course_ids = ['course-v1:edX+DemoX.1+2T2017', 'course-v1:edX+DemoX.1+2T2018']
@mock.patch(
'openedx.core.djangoapps.notifications.tasks.create_course_notification_preferences_for_courses'
)
def test_generate_course_notification_preferences(self, mock_task):
"""
Test generate_course_notification_preferences command.
"""
call_command(
'generate_course_notification_preferences',
self.course_ids,
)
mock_task.delay.assert_called_once_with(self.course_ids)

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

@@ -1,465 +0,0 @@
# pylint: disable = W0212
"""
Test for account level migration command
"""
from unittest.mock import Mock, patch
from django.contrib.auth import get_user_model
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.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'
class MigrateNotificationPreferencesTestCase(TestCase):
"""Test cases for the migrate_preferences_to_account_level_model management command."""
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')
# Sample notification preference config
self.sample_config = {
"grading": {
"enabled": True,
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
},
"ora_grade_assigned": {
"web": True,
"push": False,
"email": True,
"email_cadence": "Daily"
}
},
"core_notification_types": ["grade_assigned", "grade_updated"]
},
"discussion": {
"enabled": True,
"notification_types": {
"core": {
"web": False,
"push": True,
"email": False,
"email_cadence": "Weekly"
},
"new_post": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Immediately"
}
},
"core_notification_types": ["new_response", "new_comment"]
}
}
def tearDown(self):
"""Clean up test data."""
CourseNotificationPreference.objects.all().delete()
NotificationPreference.objects.all().delete()
User.objects.all().delete()
def test_get_user_ids_to_process(self):
"""Test that _get_user_ids_to_process returns correct user IDs."""
# Create course preferences for users
CourseNotificationPreference.objects.create(
user=self.user1,
course_id='course-v1:Test+Course+1',
notification_preference_config=self.sample_config
)
CourseNotificationPreference.objects.create(
user=self.user2,
course_id='course-v1:Test+Course+2',
notification_preference_config=self.sample_config
)
command = Command()
user_ids = list(command._get_user_ids_to_process())
self.assertEqual(len(user_ids), 2)
self.assertIn(self.user1.id, user_ids)
self.assertIn(self.user2.id, user_ids)
def test_create_preference_object(self):
"""Test that _create_preference_object creates correct NotificationPreference instance."""
command = Command()
values = {
'web': True,
'push': False,
'email': True,
'email_cadence': 'Weekly'
}
preference = command._create_preference_object(
user_id=self.user1.id,
app_name='grading',
notification_type='ora_grade_assigned',
values=values
)
self.assertEqual(preference.user_id, self.user1.id)
self.assertEqual(preference.app, 'grading')
self.assertEqual(preference.type, 'ora_grade_assigned')
self.assertTrue(preference.web)
self.assertFalse(preference.push)
self.assertTrue(preference.email)
self.assertEqual(preference.email_cadence, 'Weekly')
def test_create_preference_object_with_defaults(self):
"""Test _create_preference_object with missing values uses defaults."""
command = Command()
values = {'web': True} # Missing other values
preference = command._create_preference_object(
user_id=self.user1.id,
app_name='grading',
notification_type='test_type',
values=values
)
self.assertTrue(preference.web)
self.assertIsNone(preference.push)
self.assertIsNone(preference.email)
self.assertEqual(preference.email_cadence, EmailCadence.DAILY)
@patch(f'{COMMAND_MODULE}.aggregate_notification_configs')
def test_process_user_preferences_success(self, mock_aggregate):
"""Test successful processing of user preferences."""
# Setup
CourseNotificationPreference.objects.create(
user=self.user1,
course_id='course-v1:Test+Course+1',
notification_preference_config=self.sample_config
)
mock_aggregate.return_value = {
'grading': {
'notification_types': {
'core': {'web': True, 'push': True, 'email': True, 'email_cadence': 'Daily'},
'grade_assigned': {'web': True, 'push': False, 'email': True, 'email_cadence': 'Daily'}
},
'core_notification_types': ['grade_updated']
}
}
command = Command()
preferences = command._process_batch([self.user1.id])
self.assertEqual(len(preferences), 2) # grade_assigned + grade_updated
# Check grade_assigned preference
grade_assigned_pref = next(p for p in preferences if p.type == 'grade_assigned')
self.assertEqual(grade_assigned_pref.app, 'grading')
self.assertTrue(grade_assigned_pref.web)
self.assertFalse(grade_assigned_pref.push)
self.assertTrue(grade_assigned_pref.email)
# Check core notification type
grade_updated_pref = next(p for p in preferences if p.type == 'grade_updated')
self.assertEqual(grade_updated_pref.app, 'grading')
self.assertTrue(grade_updated_pref.web)
self.assertTrue(grade_updated_pref.push)
self.assertTrue(grade_updated_pref.email)
@patch(f'{COMMAND_MODULE}.aggregate_notification_configs')
def test_process_user_preferences_no_course_preferences(self, mock_aggregate):
"""Test processing user with no course preferences."""
command = Command()
preferences = command._process_batch([self.user1.id])
self.assertEqual(len(preferences), 0)
mock_aggregate.assert_not_called()
@patch(f'{COMMAND_MODULE}.aggregate_notification_configs')
def test_process_user_preferences_malformed_data(self, mock_aggregate):
"""Test handling of malformed notification config data."""
CourseNotificationPreference.objects.create(
user=self.user1,
course_id='course-v1:Test+Course+1',
notification_preference_config=self.sample_config
)
# Mock malformed data
mock_aggregate.return_value = {
'grading': 'invalid_string', # Should be dict
'discussion': {
'notification_types': 'invalid_string', # Should be dict
'core_notification_types': []
},
'updates': {
'notification_types': {
'core': {'web': True, 'push': True, 'email': True},
'invalid_type': None # Invalid notification type data
},
'core_notification_types': 'invalid_string' # Should be list
}
}
command = Command()
with self.assertLogs(level='WARNING') as log:
preferences = command._process_batch([self.user1.id])
self.assertEqual(len(preferences), 0)
self.assertIn('Malformed app_config', log.output[0])
@patch(f'{COMMAND_MODULE}.logger')
def test_handle_dry_run_mode(self, mock_logger):
"""Test command execution in dry-run mode."""
CourseNotificationPreference.objects.create(
user=self.user1,
course_id='course-v1:Test+Course+1',
notification_preference_config=self.sample_config
)
with patch.object(Command, '_process_batch') as mock_process:
mock_process.return_value = [
NotificationPreference(
user_id=self.user1.id,
app='grading',
type='test_type',
web=True,
push=False,
email=True,
email_cadence='Daily'
)
]
call_command('migrate_preferences_to_account_level_model', '--dry-run', '--batch-size=1')
# Check that no actual database changes were made
self.assertEqual(NotificationPreference.objects.count(), 0)
# Verify dry-run logging
mock_logger.info.assert_any_call(
'Performing a DRY RUN. No changes will be made to the database.'
)
@patch(f'{COMMAND_MODULE}.logger')
def test_handle_use_default_mode(self, mock_logger):
"""Test command execution while using default mode."""
sample_config = {
"grading": {
"enabled": True,
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
},
"ora_grade_assigned": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
}
},
"core_notification_types": []
},
"discussion": {
"enabled": True,
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Weekly"
},
"new_discussion_post": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Immediately"
}
},
"core_notification_types": ["response_on_followed_post"]
}
}
CourseNotificationPreference.objects.create(
user=self.user1,
course_id='course-v1:Test+Course+1',
notification_preference_config=sample_config
)
call_command(
'migrate_preferences_to_account_level_model',
'--use-default',
'push'
)
# Check that no actual database changes were made
self.assertEqual(NotificationPreference.objects.count(), 3)
self.assertEqual(
NotificationPreference.objects.get(type='ora_grade_assigned').push,
False
)
self.assertEqual(
NotificationPreference.objects.get(type='new_discussion_post').push,
False
)
self.assertEqual(
NotificationPreference.objects.get(type='response_on_followed_post').push,
True
)
def test_handle_normal_execution(self):
"""Test normal command execution without dry-run."""
CourseNotificationPreference.objects.create(
user=self.user1,
course_id='course-v1:Test+Course+1',
notification_preference_config=self.sample_config
)
# Create existing account preferences to test deletion
NotificationPreference.objects.create(
user=self.user1,
app='old_app',
type='old_type',
web=True,
push=False,
email=False,
email_cadence='Daily'
)
with patch.object(Command, '_process_batch') as mock_process:
mock_process.return_value = [
NotificationPreference(
user_id=self.user1.id,
app='grading',
type='test_type',
web=True,
push=False,
email=True,
email_cadence='Daily'
)
]
call_command('migrate_preferences_to_account_level_model', '--batch-size=1')
# Verify old preferences were deleted and new ones created
self.assertEqual(NotificationPreference.objects.count(), 1)
new_pref = NotificationPreference.objects.first()
self.assertEqual(new_pref.app, 'grading')
self.assertEqual(new_pref.type, 'test_type')
@patch(f'{COMMAND_MODULE}.transaction.atomic')
def test_migrate_preferences_to_account_level_model(self, mock_atomic):
"""Test that users are processed in batches correctly."""
# Mock atomic to avoid transaction issues during testing
mock_atomic.return_value.__enter__ = Mock()
mock_atomic.return_value.__exit__ = Mock(return_value=None)
# Create course preferences for multiple users
for i, user in enumerate([self.user1, self.user2, self.user3]):
CourseNotificationPreference.objects.create(
user=user,
course_id=f'course-v1:Test+Course+{i}',
notification_preference_config=self.sample_config
)
call_command('migrate_preferences_to_account_level_model', '--batch-size=2')
# Check that preferences were created for each user
self.assertEqual(NotificationPreference.objects.count(), 18)
def test_command_arguments(self):
"""Test that command arguments are handled correctly."""
command = Command()
parser = command.create_parser('test', 'migrate_preferences_to_account_level_model')
# Test default arguments
options = parser.parse_args([])
self.assertEqual(options.batch_size, 1000)
self.assertFalse(options.dry_run)
# Test custom arguments
options = parser.parse_args(['--batch-size', '500', '--dry-run'])
self.assertEqual(options.batch_size, 500)
self.assertTrue(options.dry_run)
@patch(f'{COMMAND_MODULE}.aggregate_notification_configs')
def test_process_user_preferences_with_core_types(self, mock_aggregate):
"""Test processing of core notification types specifically."""
CourseNotificationPreference.objects.create(
user=self.user1,
course_id='course-v1:Test+Course+1',
notification_preference_config=self.sample_config
)
mock_aggregate.return_value = {
'discussion': {
'notification_types': {
'core': {'web': False, 'push': True, 'email': False, 'email_cadence': 'Weekly'}
},
'core_notification_types': ['new_response', 'new_comment', None, 123] # Include invalid types
}
}
command = Command()
with self.assertLogs(level='WARNING') as log:
preferences = command._process_batch([self.user1.id])
# Should create 2 valid core preferences (ignoring None and 123)
valid_prefs = [p for p in preferences if p.type in ['new_response', 'new_comment']]
self.assertEqual(len(valid_prefs), 2)
# Check that invalid core types were logged as warnings
warning_logs = [log for log in log.output if 'Skipping malformed core_type_name' in log]
self.assertEqual(len(warning_logs), 2)
def test_progress_logging(self):
"""Test that progress is logged at appropriate intervals."""
# Create enough users to trigger progress logging
users = []
for i in range(10):
user = User.objects.create_user(username=f'userX{i}', email=f'userx{i}@example.com')
users.append(user)
CourseNotificationPreference.objects.create(
user=user,
course_id=f'course-v1:Test+Course+{i}',
notification_preference_config=self.sample_config
)
with patch.object(Command, '_process_batch') as mock_process:
mock_process.return_value = []
with patch(f'{COMMAND_MODULE}.logger') as mock_logger:
call_command('migrate_preferences_to_account_level_model', '--batch-size=1')
# Check that progress was logged (every 5 batches)
progress_calls = [call for call in mock_logger.info.call_args_list
if 'PROGRESS:' in str(call)]
self.assertGreater(len(progress_calls), 0)
def test_empty_batch_handling(self):
"""Test handling when no preferences need to be created."""
CourseNotificationPreference.objects.create(
user=self.user1,
course_id='course-v1:Test+Course+1',
notification_preference_config=self.sample_config
)
with patch.object(Command, '_process_batch') as mock_process:
mock_process.return_value = [] # No preferences to create
with patch(f'{COMMAND_MODULE}.logger') as mock_logger:
call_command('migrate_preferences_to_account_level_model', '--batch-size=1')
# Should log that no preferences were created
no_prefs_calls = [call for call in mock_logger.info.call_args_list
if 'No preferences to create' in str(call)]
self.assertEqual(len(no_prefs_calls), 1)

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

@@ -5,10 +5,8 @@ Serializers for the notifications API.
from django.core.exceptions import ValidationError
from rest_framework import serializers
from common.djangoapps.student.models import CourseEnrollment
from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
Notification,
get_additional_notification_channel_settings,
get_notification_channels
@@ -16,7 +14,6 @@ from openedx.core.djangoapps.notifications.models import (
from .base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES, EmailCadence
from .email.utils import is_notification_type_channel_editable
from .utils import remove_preferences_with_no_access
def add_info_to_notification_config(config_obj):
@@ -84,164 +81,6 @@ class CourseOverviewSerializer(serializers.ModelSerializer):
fields = ('id', 'display_name')
class NotificationCourseEnrollmentSerializer(serializers.ModelSerializer):
"""
Serializer for CourseEnrollment model.
"""
course = CourseOverviewSerializer()
class Meta:
model = CourseEnrollment
fields = ('course',)
class UserCourseNotificationPreferenceSerializer(serializers.ModelSerializer):
"""
Serializer for user notification preferences.
"""
course_name = serializers.SerializerMethodField(read_only=True)
class Meta:
model = CourseNotificationPreference
fields = ('id', 'course_name', 'course_id', 'notification_preference_config',)
read_only_fields = ('id', 'course_name', 'course_id',)
write_only_fields = ('notification_preference_config',)
def to_representation(self, instance):
"""
Override to_representation to add info of all notification types
"""
preferences = super().to_representation(instance)
course_id = self.context['course_id']
user = self.context['user']
preferences = add_info_to_notification_config(preferences)
preferences = remove_preferences_with_no_access(preferences, user)
preferences['notification_preference_config'] = add_non_editable_in_preference(
preferences.get('notification_preference_config', {})
)
return preferences
def get_course_name(self, obj):
"""
Returns course name from course id.
"""
return CourseOverview.get_from_id(obj.course_id).display_name
class UserNotificationPreferenceUpdateSerializer(serializers.Serializer):
"""
Serializer for user notification preferences update.
"""
notification_app = serializers.CharField()
value = serializers.BooleanField(required=False)
notification_type = serializers.CharField(required=False)
notification_channel = serializers.CharField(required=False)
email_cadence = serializers.CharField(required=False)
def validate(self, attrs):
"""
Validation for notification preference update form
"""
notification_app = attrs.get('notification_app')
notification_type = attrs.get('notification_type')
notification_channel = attrs.get('notification_channel')
notification_email_cadence = attrs.get('email_cadence')
notification_app_config = self.instance.notification_preference_config
if notification_email_cadence:
if not notification_type:
raise ValidationError(
'notification_type is required for email_cadence.'
)
if EmailCadence.get_email_cadence_value(notification_email_cadence) is None:
raise ValidationError(
f'{attrs.get("value")} is not a valid email cadence.'
)
if notification_type and not notification_channel:
raise ValidationError(
'notification_channel is required for notification_type.'
)
if not notification_app_config.get(notification_app, None):
raise ValidationError(
f'{notification_app} is not a valid notification app.'
)
if notification_type:
notification_types = notification_app_config.get(notification_app).get('notification_types')
if not notification_types.get(notification_type, None):
raise ValidationError(
f'{notification_type} is not a valid notification type.'
)
if (
notification_channel and
notification_channel not in get_notification_channels()
and notification_channel not in get_additional_notification_channel_settings()
):
raise ValidationError(
f'{notification_channel} is not a valid notification channel setting.'
)
if notification_app and notification_type and notification_channel:
if not is_notification_type_channel_editable(
notification_app,
notification_type,
'email' if notification_channel == 'email_cadence' else notification_channel
):
raise ValidationError({
'notification_channel': (
f'{notification_channel} is not editable for notification type '
f'{notification_type}.'
)
})
return attrs
def update(self, instance, validated_data):
"""
Update notification preference config.
"""
notification_app = validated_data.get('notification_app')
notification_type = validated_data.get('notification_type')
notification_channel = validated_data.get('notification_channel')
value = validated_data.get('value')
notification_email_cadence = validated_data.get('email_cadence')
user_notification_preference_config = instance.notification_preference_config
# Notification email cadence update
if notification_email_cadence and notification_type:
user_notification_preference_config[notification_app]['notification_types'][notification_type][
'email_cadence'] = notification_email_cadence
# Notification type channel update
elif notification_type and notification_channel:
# Update the notification preference for specific notification type
user_notification_preference_config[
notification_app]['notification_types'][notification_type][notification_channel] = value
# Notification app-wide channel update
elif notification_channel and not notification_type:
app_prefs = user_notification_preference_config[notification_app]
for notification_type_name, notification_type_preferences in app_prefs['notification_types'].items():
non_editable_channels = get_non_editable_channels(notification_app).get(notification_type_name, [])
if notification_channel not in non_editable_channels:
app_prefs['notification_types'][notification_type_name][notification_channel] = value
# Notification app update
else:
# Update the notification preference for notification_app
user_notification_preference_config[notification_app]['enabled'] = value
instance.save()
return instance
class NotificationSerializer(serializers.ModelSerializer):
"""
Serializer for the Notification model.

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

@@ -8,12 +8,10 @@ from celery import shared_task
from celery.utils.log import get_task_logger
from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import transaction
from edx_django_utils.monitoring import set_code_owner_attribute
from opaque_keys.edx.keys import CourseKey
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,
@@ -35,10 +33,8 @@ from openedx.core.djangoapps.notifications.grouping_notifications import (
group_user_notifications
)
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
@@ -46,31 +42,6 @@ from openedx.core.djangoapps.notifications.utils import clean_arguments, get_lis
logger = get_task_logger(__name__)
@shared_task(bind=True, ignore_result=True)
@set_code_owner_attribute
@transaction.atomic
def create_course_notification_preferences_for_courses(self, course_ids):
"""
This task creates Course Notification Preferences for users in courses.
"""
newly_created = 0
for course_id in course_ids:
enrollments = CourseEnrollment.objects.filter(course_id=course_id, is_active=True)
logger.debug(f'Found {enrollments.count()} enrollments for course {course_id}')
logger.debug(f'Creating Course Notification Preferences for course {course_id}')
for enrollment in enrollments:
_, created = CourseNotificationPreference.objects.get_or_create(
user=enrollment.user, course_id=course_id
)
if created:
newly_created += 1
logger.debug(
f'CourseNotificationPreference back-fill completed for course {course_id}.\n'
f'Newly created course preferences: {newly_created}.\n'
)
@shared_task(ignore_result=True)
@set_code_owner_attribute
def delete_notifications(kwargs):
@@ -247,16 +218,6 @@ def is_notification_valid(notification_type, context):
return True
def update_user_preference(preference: CourseNotificationPreference, user_id, course_id):
"""
Update user preference if config version is changed.
"""
current_version = get_course_notification_preference_config_version()
if preference.config_version != current_version:
return preference.get_user_course_preference(user_id, course_id)
return preference
def update_account_user_preference(user_id: int) -> None:
"""
Update account level user preferences to ensure all notification types are present.
@@ -333,26 +294,6 @@ def _get_channel_default(is_core: bool, notification_type: str, channel: str) ->
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.
"""
new_preferences = []
for user_id in user_ids:
if not any(preference.user_id == int(user_id) for preference in preferences):
new_preferences.append(CourseNotificationPreference(
user_id=user_id,
course_id=course_id,
))
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
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.

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

@@ -1,516 +0,0 @@
"""
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_NOTIFICATIONS, ENABLE_PUSH_NOTIFICATIONS
from ..models import CourseNotificationPreference, Notification
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)
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
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)
@override_waffle_flag(ENABLE_PUSH_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)
@override_waffle_flag(ENABLE_PUSH_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()
send_notifications([self.user.id], str(self.course_1.id), app_name, notification_type, context, content_url)
self.assertEqual(len(Notification.objects.all()), 1)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@override_waffle_flag(ENABLE_PUSH_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()
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(), 0)
user_notifications_mock.assert_not_called()
@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
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, 10, 3),
(settings.NOTIFICATION_CREATION_BATCH_SIZE + 10, 12, 5),
(settings.NOTIFICATION_CREATION_BATCH_SIZE - 10, 10, 3),
)
@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
discussion_config['notification_types'][notification_type]['push'] = 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(10):
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)
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 override_waffle_flag(ENABLE_PUSH_NOTIFICATIONS, active=True):
with self.assertNumQueries(12):
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)
@override_waffle_flag(ENABLE_PUSH_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)
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,
)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.data(
(False, False, 0),
(False, True, 0),
(True, False, 0),
(True, True, 0),
)
@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

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

@@ -18,78 +18,12 @@ from xmodule.modulestore.tests.factories import CourseFactory
from ..config.waffle import 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)
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
class SendNotificationsTest(ModuleStoreTestCase):
"""
@@ -282,9 +216,9 @@ class SendBatchNotificationsTest(ModuleStoreTestCase):
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.data(
(settings.NOTIFICATION_CREATION_BATCH_SIZE, 10, 3),
(settings.NOTIFICATION_CREATION_BATCH_SIZE + 10, 12, 5),
(settings.NOTIFICATION_CREATION_BATCH_SIZE - 10, 10, 3),
(settings.NOTIFICATION_CREATION_BATCH_SIZE, 12, 3),
(settings.NOTIFICATION_CREATION_BATCH_SIZE + 10, 14, 5),
(settings.NOTIFICATION_CREATION_BATCH_SIZE - 10, 12, 3),
)
@ddt.unpack
def test_notification_is_send_in_batch(self, creation_size, prefs_query_count, notifications_query_count):
@@ -334,7 +268,7 @@ class SendBatchNotificationsTest(ModuleStoreTestCase):
"username": "Test Author"
}
with override_waffle_flag(ENABLE_NOTIFICATIONS, active=True):
with self.assertNumQueries(10):
with self.assertNumQueries(12):
send_notifications(user_ids, str(self.course.id), notification_app, notification_type,
context, "http://test.url")
@@ -354,7 +288,7 @@ class SendBatchNotificationsTest(ModuleStoreTestCase):
"replier_name": "Replier Name"
}
with override_waffle_flag(ENABLE_NOTIFICATIONS, active=True):
with self.assertNumQueries(12):
with self.assertNumQueries(14):
send_notifications(user_ids, str(self.course.id), notification_app, notification_type,
context, "http://test.url")

349
openedx/core/djangoapps/notifications/tests/test_utils.py
View File

@@ -1,349 +0,0 @@
"""
Test cases for the notification utility functions.
"""
import copy
import unittest
import pytest
from common.djangoapps.student.tests.factories import UserFactory
from openedx.core.djangoapps.django_comment_common.models import assign_role, FORUM_ROLE_MODERATOR
from openedx.core.djangoapps.notifications.utils import aggregate_notification_configs, \
filter_out_visible_preferences_by_course_ids
class TestAggregateNotificationConfigs(unittest.TestCase):
"""
Test cases for the aggregate_notification_configs function.
"""
def test_empty_configs_list_returns_default(self):
"""
If the configs list is empty, the default config should be returned.
"""
default_config = [{
"grading": {
"enabled": False,
"non_editable": {},
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Daily"
}
}
}
}]
result = aggregate_notification_configs(default_config)
assert result == default_config[0]
def test_enable_notification_type(self):
"""
If a config enables a notification type, it should be enabled in the result.
"""
config_list = [
{
"grading": {
"enabled": False,
"non_editable": {},
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Weekly"
}
}
}
},
{
"grading": {
"enabled": True,
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Weekly"
}
}
}
}]
result = aggregate_notification_configs(config_list)
assert result["grading"]["enabled"] is True
assert result["grading"]["notification_types"]["core"]["web"] is True
assert result["grading"]["notification_types"]["core"]["push"] is True
assert result["grading"]["notification_types"]["core"]["email"] is True
# Use default email_cadence
assert result["grading"]["notification_types"]["core"]["email_cadence"] == "Weekly"
def test_merge_core_notification_types(self):
"""
Core notification types should be merged across configs.
"""
config_list = [
{
"discussion": {
"enabled": True,
"core_notification_types": ["new_comment"],
"notification_types": {}
}
},
{
"discussion": {
"core_notification_types": ["new_response", "new_comment"]
}
}]
result = aggregate_notification_configs(config_list)
assert set(result["discussion"]["core_notification_types"]) == {
"new_comment", "new_response"
}
def test_multiple_configs_aggregate(self):
"""
Multiple configs should be aggregated together.
"""
config_list = [
{
"updates": {
"enabled": False,
"notification_types": {
"course_updates": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Weekly"
}
}
}
},
{
"updates": {
"enabled": True,
"notification_types": {
"course_updates": {
"web": True,
"email_cadence": "Weekly"
}
}
}
},
{
"updates": {
"notification_types": {
"course_updates": {
"push": True,
"email_cadence": "Weekly"
}
}
}
}
]
result = aggregate_notification_configs(config_list)
assert result["updates"]["enabled"] is True
assert result["updates"]["notification_types"]["course_updates"]["web"] is True
assert result["updates"]["notification_types"]["course_updates"]["push"] is True
assert result["updates"]["notification_types"]["course_updates"]["email"] is False
# Use default email_cadence
assert result["updates"]["notification_types"]["course_updates"]["email_cadence"] == "Weekly"
def test_ignore_unknown_notification_types(self):
"""
Unknown notification types should be ignored.
"""
config_list = [
{
"grading": {
"enabled": False,
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Daily"
}
}
}
},
{
"grading": {
"notification_types": {
"unknown_type": {
"web": True,
"push": True,
"email": True
}
}
}
}]
result = aggregate_notification_configs(config_list)
assert "unknown_type" not in result["grading"]["notification_types"]
assert result["grading"]["notification_types"]["core"]["web"] is False
def test_ignore_unknown_categories(self):
"""
Unknown categories should be ignored.
"""
config_list = [
{
"grading": {
"enabled": False,
"notification_types": {}
}
},
{
"unknown_category": {
"enabled": True,
"notification_types": {}
}
}]
result = aggregate_notification_configs(config_list)
assert "unknown_category" not in result
assert result["grading"]["enabled"] is False
def test_preserves_default_structure(self):
"""
The resulting config should have the same structure as the default config.
"""
config_list = [
{
"discussion": {
"enabled": False,
"non_editable": {"core": ["web"]},
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Weekly"
}
},
"core_notification_types": []
}
},
{
"discussion": {
"enabled": True,
"extra_field": "should_not_appear"
}
}
]
result = aggregate_notification_configs(config_list)
assert set(result["discussion"].keys()) == {
"enabled", "non_editable", "notification_types", "core_notification_types"
}
assert "extra_field" not in result["discussion"]
def test_if_email_cadence_has_diff_set_mix_as_value(self):
"""
If email_cadence is different in the configs, set it to "Mixed".
"""
config_list = [
{
"grading": {
"enabled": False,
"notification_types": {
"core": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Daily"
}
}
}
},
{
"grading": {
"enabled": True,
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Weekly"
}
}
}
},
{
"grading": {
"notification_types": {
"core": {
"email_cadence": "Monthly"
}
}
}
}
]
result = aggregate_notification_configs(config_list)
assert result["grading"]["notification_types"]["core"]["email_cadence"] == "Mixed"
@pytest.mark.django_db
class TestVisibilityFilter(unittest.TestCase):
"""
Test cases for the filter_out_visible_preferences_by_course_ids function.
"""
def setUp(self):
self.user = UserFactory()
self.course_key = "course-v1:edX+DemoX+Demo_Course"
self.mock_preferences = {
'discussion': {
'enabled': True,
'non_editable': {'core': ['web']},
'notification_types': {
'core': {'web': True, 'push': True, 'email': True, 'email_cadence': 'Daily'},
'content_reported': {'web': True, 'push': True, 'email': True, 'email_cadence': 'Daily'},
'new_question_post': {'web': False, 'push': False, 'email': False, 'email_cadence': 'Daily'},
'new_discussion_post': {'web': False, 'push': False, 'email': False, 'email_cadence': 'Daily'},
'new_instructor_all_learners_post': {
'web': True, 'push': False, 'email': False, 'email_cadence': 'Daily'
}
},
'core_notification_types': [
'new_response', 'comment_on_followed_post',
'response_endorsed_on_thread', 'new_comment_on_response',
'new_comment', 'response_on_followed_post', 'response_endorsed'
]
}
}
def test_visibility_filter_with_no_role(self):
"""
Test that the preferences are filtered out correctly when the user has no role.
"""
updated_preferences = filter_out_visible_preferences_by_course_ids(
self.user,
copy.deepcopy(self.mock_preferences),
[self.course_key]
)
assert updated_preferences != self.mock_preferences
assert not updated_preferences["discussion"]["notification_types"].get("content_reported", False)
def test_visibility_filter_with_instructor_role(self):
"""
Instructors should see all preferences.
"""
updated_preferences = filter_out_visible_preferences_by_course_ids(
self.user,
self.mock_preferences,
[self.course_key]
)
assign_role(self.course_key, self.user, FORUM_ROLE_MODERATOR)
assert updated_preferences == self.mock_preferences

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

@@ -1,12 +1,8 @@
"""
Tests for the views in the notifications app.
"""
import itertools
import json
from copy import deepcopy
from datetime import datetime, timedelta
from unittest import mock
from unittest.mock import patch
import ddt
from django.conf import settings
@@ -14,8 +10,6 @@ from django.contrib.auth import get_user_model
from django.test.utils import override_settings
from django.urls import reverse
from edx_toggles.toggles.testutils import override_waffle_flag
from openedx_events.learning.data import CourseData, CourseEnrollmentData, UserData, UserPersonalData
from openedx_events.learning.signals import COURSE_ENROLLMENT_CREATED
from pytz import UTC
from rest_framework import status
from rest_framework.test import APIClient, APITestCase
@@ -31,578 +25,22 @@ from openedx.core.djangoapps.django_comment_common.models import (
FORUM_ROLE_MODERATOR
)
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx.core.djangoapps.notifications.email import ONE_CLICK_EMAIL_UNSUB_KEY
from openedx.core.djangoapps.notifications.email.utils import encrypt_object, encrypt_string
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
Notification,
get_course_notification_preference_config_version, NotificationPreference
)
from openedx.core.djangoapps.notifications.serializers import NotificationCourseEnrollmentSerializer, \
add_non_editable_in_preference
from openedx.core.djangoapps.user_api.models import UserPreference
from openedx.core.djangoapps.notifications.email.utils import update_user_preferences_from_patch
from openedx.core.djangoapps.notifications.serializers import add_non_editable_in_preference
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
from ..base_notification import COURSE_NOTIFICATION_APPS, COURSE_NOTIFICATION_TYPES, NotificationAppManager, \
NotificationTypeManager
from ..base_notification import COURSE_NOTIFICATION_APPS, NotificationTypeManager
from ..utils import get_notification_types_with_visibility_settings, exclude_inaccessible_preferences
User = get_user_model()
@ddt.ddt
class CourseEnrollmentListViewTest(ModuleStoreTestCase):
"""
Tests for the CourseEnrollmentListView.
"""
def setUp(self):
"""
Set up the test.
"""
super().setUp()
self.client = APIClient()
self.user = UserFactory()
course_1 = CourseFactory.create(
org='testorg',
number='testcourse',
run='testrun'
)
course_2 = CourseFactory.create(
org='testorg',
number='testcourse_two',
run='testrun'
)
course_overview_1 = CourseOverviewFactory.create(id=course_1.id, org='AwesomeOrg')
course_overview_2 = CourseOverviewFactory.create(id=course_2.id, org='AwesomeOrg')
self.enrollment1 = CourseEnrollment.objects.create(
user=self.user,
course=course_overview_1,
is_active=True,
mode='audit'
)
self.enrollment2 = CourseEnrollment.objects.create(
user=self.user,
course=course_overview_2,
is_active=False,
mode='honor'
)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.unpack
def test_course_enrollment_list_view(self):
"""
Test the CourseEnrollmentListView.
"""
self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
url = reverse('enrollment-list')
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.data['results']
enrollments = CourseEnrollment.objects.filter(user=self.user, is_active=True)
expected_data = NotificationCourseEnrollmentSerializer(enrollments, many=True).data
self.assertEqual(len(data), 1)
self.assertEqual(data, expected_data)
self.assertEqual(response.data['show_preferences'], True)
def test_course_enrollment_api_permission(self):
"""
Calls api without login.
Check is 401 is returned
"""
url = reverse('enrollment-list')
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.ddt
class CourseEnrollmentPostSaveTest(ModuleStoreTestCase):
"""
Tests for the post_save signal for CourseEnrollment.
"""
def setUp(self):
"""
Set up the test.
"""
super().setUp()
self.user = UserFactory()
self.course = CourseFactory.create(
org='testorg',
number='testcourse',
run='testrun'
)
course_overview = CourseOverviewFactory.create(id=self.course.id, org='AwesomeOrg')
self.course_enrollment = CourseEnrollment.objects.create(
user=self.user,
course=course_overview,
is_active=True,
mode='audit'
)
def test_course_enrollment_post_save(self):
"""
Test the post_save signal for CourseEnrollment.
"""
# Emit post_save signal
enrollment_data = CourseEnrollmentData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=CourseData(
course_key=self.course.id,
display_name=self.course.display_name,
),
mode=self.course_enrollment.mode,
is_active=self.course_enrollment.is_active,
creation_date=self.course_enrollment.created,
)
COURSE_ENROLLMENT_CREATED.send_event(
enrollment=enrollment_data
)
# Assert that CourseNotificationPreference object was created with correct attributes
notification_preferences = CourseNotificationPreference.objects.all()
self.assertEqual(notification_preferences.count(), 1)
self.assertEqual(notification_preferences[0].user, self.user)
def test_disabled_email_preference_is_generated_after_unsubscribe(self):
"""
Test the post_save signal for CourseEnrollment for user with one-click unsubscribe.
"""
UserPreference.objects.create(user_id=self.user.id, key=ONE_CLICK_EMAIL_UNSUB_KEY)
enrollment_data = CourseEnrollmentData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=CourseData(
course_key=self.course.id,
display_name=self.course.display_name,
),
mode=self.course_enrollment.mode,
is_active=self.course_enrollment.is_active,
creation_date=self.course_enrollment.created,
)
COURSE_ENROLLMENT_CREATED.send_event(
enrollment=enrollment_data
)
notification_preferences = CourseNotificationPreference.objects.all()
self.assertEqual(notification_preferences.count(), 1)
self.assertEqual(notification_preferences[0].user, self.user)
email_preferences = [
notification["email"]
for app in notification_preferences[0].notification_preference_config.values()
for notification in app["notification_types"].values()
]
self.assertEqual(email_preferences, [False] * len(email_preferences))
@ddt.data(*itertools.product(('web', 'email'), (True, False)))
@ddt.unpack
def test_course_preference_creation_for_inactive_enrollments_on_unsub(
self,
channel,
value
):
"""
Test that unsubscribing through one click email does not create new course preferences for inactive enrollments
if not already exists.
"""
self.course_enrollment.is_active = False
self.course_enrollment.save()
encrypted_username = encrypt_string(self.user.username)
encrypted_patch = encrypt_object({
'channel': channel,
'value': value
})
update_user_preferences_from_patch(encrypted_username, encrypted_patch)
self.assertEqual(CourseNotificationPreference.objects.all().count(), 0)
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
@ddt.ddt
class UserNotificationPreferenceAPITest(ModuleStoreTestCase):
"""
Test for user notification preference API.
"""
def setUp(self):
super().setUp()
self.user = UserFactory()
self.course = CourseFactory.create(
org='testorg',
number='testcourse',
run='testrun'
)
course_overview = CourseOverviewFactory.create(id=self.course.id, org='AwesomeOrg')
self.course_enrollment = CourseEnrollment.objects.create(
user=self.user,
course=course_overview,
is_active=True,
mode='audit'
)
self.client = APIClient()
self.path = reverse('notification-preferences', kwargs={'course_key_string': self.course.id})
enrollment_data = CourseEnrollmentData(
user=UserData(
pii=UserPersonalData(
username=self.user.username,
email=self.user.email,
name=self.user.profile.name,
),
id=self.user.id,
is_active=self.user.is_active,
),
course=CourseData(
course_key=self.course.id,
display_name=self.course.display_name,
),
mode=self.course_enrollment.mode,
is_active=self.course_enrollment.is_active,
creation_date=self.course_enrollment.created,
)
COURSE_ENROLLMENT_CREATED.send_event(
enrollment=enrollment_data
)
def _expected_api_response(self, is_staff=False):
"""
Helper method to return expected API response.
"""
response = {
'id': 1,
'course_name': 'course-v1:testorg+testcourse+testrun Course',
'course_id': 'course-v1:testorg+testcourse+testrun',
'notification_preference_config': {
'discussion': {
'enabled': True,
'core_notification_types': [
'new_comment_on_response',
'new_comment',
'new_response',
'response_on_followed_post',
'comment_on_followed_post',
'response_endorsed_on_thread',
'response_endorsed'
],
'notification_types': {
'new_discussion_post': {
'web': False,
'email': False,
'push': False,
'email_cadence': 'Daily',
'info': ''
},
'new_question_post': {
'web': False,
'email': False,
'push': False,
'email_cadence': 'Daily',
'info': ''
},
'core': {
'web': True,
'email': True,
'push': True,
'email_cadence': 'Daily',
'info': 'Notifications for responses and comments on your posts, and the ones you’re '
'following, including endorsements to your responses and on your posts.'
},
'content_reported': {
'web': True,
'email': True,
'push': False,
'info': '',
'email_cadence': 'Daily',
},
'new_instructor_all_learners_post': {
'web': True,
'email': False,
'push': False,
'email_cadence': 'Daily',
'info': ''
},
},
'non_editable': {
'new_discussion_post': ['push'],
'new_question_post': ['push'],
'new_instructor_all_learners_post': ['push'],
}
},
'updates': {
'enabled': True,
'core_notification_types': [],
'notification_types': {
'course_updates': {
'web': True,
'email': False,
'push': False,
'email_cadence': 'Daily',
'info': ''
},
'core': {
'web': True,
'email': True,
'push': True,
'email_cadence': 'Daily',
'info': 'Notifications for new announcements and updates from the course team.'
}
},
'non_editable': {
'course_updates': ['push']
}
},
'grading': {
'enabled': True,
'core_notification_types': [],
'notification_types': {
'ora_staff_notifications': {
'web': True,
'email': False,
'push': False,
'email_cadence': 'Daily',
'info': 'Notifications for when a submission is made for ORA that includes staff grading '
'step.'
},
'core': {
'web': True,
'email': True,
'push': True,
'email_cadence': 'Daily',
'info': 'Notifications for submission grading.'
},
'ora_grade_assigned': {
'web': True,
'email': True,
'push': False,
'email_cadence': 'Daily',
'info': ''
},
},
'non_editable': {
'ora_grade_assigned': ['push']
}
},
"enrollments": {
"enabled": True,
"core_notification_types": [],
"notification_types": {
"audit_access_expiring_soon": {
"web": True,
"email": False,
"push": False,
"email_cadence": "Daily",
"info": ""
},
"core": {
"web": True,
"email": True,
"push": True,
"email_cadence": "Daily",
"info": "Notifications for enrollments."
}
},
"non_editable": {}
}
}
}
if is_staff:
response['notification_preference_config']['grading']['non_editable'] = {
'ora_staff_notifications': ['push'],
'ora_grade_assigned': ['push']
}
return response
def test_get_user_notification_preference_without_login(self):
"""
Test get user notification preference without login.
"""
response = self.client.get(self.path)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
@mock.patch("eventtracking.tracker.emit")
def test_get_user_notification_preference(self, mock_emit):
"""
Test get user notification preference.
"""
self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
response = self.client.get(self.path)
self.assertEqual(response.status_code, status.HTTP_200_OK)
expected_response = self._expected_api_response()
expected_response = remove_notifications_with_visibility_settings(expected_response)
self.assertEqual(response.data, expected_response)
event_name, event_data = mock_emit.call_args[0]
self.assertEqual(event_name, 'edx.notifications.preferences.viewed')
@mock.patch("eventtracking.tracker.emit")
@mock.patch.dict(COURSE_NOTIFICATION_TYPES, {
**COURSE_NOTIFICATION_TYPES,
**{
'content_reported': {
'name': 'content_reported',
'visible_to': [FORUM_ROLE_MODERATOR, FORUM_ROLE_COMMUNITY_TA, FORUM_ROLE_ADMINISTRATOR]
}
}
})
@ddt.data(
FORUM_ROLE_MODERATOR,
FORUM_ROLE_COMMUNITY_TA,
FORUM_ROLE_ADMINISTRATOR,
None
)
def test_get_user_notification_preference_with_visibility_settings(self, role, mock_emit):
"""
Test get user notification preference.
"""
if role:
CourseStaffRole(self.course.id).add_users(self.user)
self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
role_instance = None
if role:
role_instance = RoleFactory(name=role, course_id=self.course.id)
role_instance.users.add(self.user)
response = self.client.get(self.path)
self.assertEqual(response.status_code, status.HTTP_200_OK)
expected_response = self._expected_api_response(is_staff=bool(role))
if not role:
expected_response = remove_notifications_with_visibility_settings(expected_response)
self.assertEqual(response.data, expected_response)
event_name, event_data = mock_emit.call_args[0]
self.assertEqual(event_name, 'edx.notifications.preferences.viewed')
if role_instance:
role_instance.users.clear()
@ddt.data(
('discussion', None, None, True, status.HTTP_200_OK, 'app_update'),
('discussion', None, None, False, status.HTTP_200_OK, 'app_update'),
('invalid_notification_app', None, None, True, status.HTTP_400_BAD_REQUEST, None),
('discussion', 'core', 'email', True, status.HTTP_200_OK, 'type_update'),
('discussion', 'core', 'email', False, status.HTTP_200_OK, 'type_update'),
# Test for email cadence update
('discussion', 'core', 'email_cadence', 'Daily', status.HTTP_200_OK, 'type_update'),
('discussion', 'core', 'email_cadence', 'Weekly', status.HTTP_200_OK, 'type_update'),
# Test for app-wide channel update
('discussion', None, 'email', True, status.HTTP_200_OK, 'app-wide-channel-update'),
('discussion', None, 'email', False, status.HTTP_200_OK, 'app-wide-channel-update'),
('discussion', 'invalid_notification_type', 'email', True, status.HTTP_400_BAD_REQUEST, None),
('discussion', 'new_comment', 'invalid_notification_channel', False, status.HTTP_400_BAD_REQUEST, None),
)
@ddt.unpack
@mock.patch("eventtracking.tracker.emit")
def test_patch_user_notification_preference(
self, notification_app, notification_type, notification_channel, value, expected_status, update_type, mock_emit,
):
"""
Test update of user notification preference.
"""
self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
payload = {
'notification_app': notification_app,
'value': value,
}
if notification_type:
payload['notification_type'] = notification_type
if notification_channel:
payload['notification_channel'] = notification_channel
response = self.client.patch(self.path, json.dumps(payload), content_type='application/json')
self.assertEqual(response.status_code, expected_status)
expected_data = self._expected_api_response()
if update_type == 'app_update':
expected_data = self._expected_api_response()
expected_data = remove_notifications_with_visibility_settings(expected_data)
expected_data['notification_preference_config'][notification_app]['enabled'] = value
self.assertEqual(response.data, expected_data)
elif update_type == 'type_update':
expected_data = self._expected_api_response()
expected_data = remove_notifications_with_visibility_settings(expected_data)
expected_data['notification_preference_config'][notification_app][
'notification_types'][notification_type][notification_channel] = value
self.assertEqual(response.data, expected_data)
elif update_type == 'app-wide-channel-update':
expected_data = remove_notifications_with_visibility_settings(expected_data)
app_prefs = expected_data['notification_preference_config'][notification_app]
for notification_type_name, notification_type_preferences in app_prefs['notification_types'].items():
non_editable_channels = app_prefs['non_editable'].get(notification_type_name, [])
if notification_channel not in non_editable_channels:
app_prefs['notification_types'][notification_type_name][notification_channel] = value
self.assertEqual(response.data, expected_data)
if expected_status == status.HTTP_200_OK:
event_name, event_data = mock_emit.call_args[0]
self.assertEqual(event_name, 'edx.notifications.preferences.updated')
self.assertEqual(event_data['notification_app'], notification_app)
self.assertEqual(event_data['notification_type'], notification_type or '')
self.assertEqual(event_data['notification_channel'], notification_channel or '')
self.assertEqual(event_data['value'], value)
def test_info_is_not_saved_in_json(self):
default_prefs = NotificationAppManager().get_notification_app_preferences()
for notification_app, app_prefs in default_prefs.items():
for _, type_prefs in app_prefs.get('notification_types', {}).items():
assert 'info' not in type_prefs.keys()
def test_non_editable_is_not_saved_in_json(self):
default_prefs = NotificationAppManager().get_notification_app_preferences()
for app_prefs in default_prefs.values():
assert 'non_editable' not in app_prefs.keys()
@ddt.data(*itertools.product(('email', 'web'), (True, False)))
@ddt.unpack
def test_unsub_user_preferences_removal_on_email_enabled(self, channel, value):
"""
Test one click unsub user preference should be removed on email enable for any app.
"""
UserPreference.objects.create(user=self.user, key=ONE_CLICK_EMAIL_UNSUB_KEY)
self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
payload = {
'notification_app': 'discussion',
'notification_type': 'core',
'notification_channel': channel,
'value': value
}
self.client.patch(self.path, json.dumps(payload), content_type='application/json')
result = 0 if channel == 'email' and value else 1
self.assertEqual(UserPreference.objects.count(), result)
@ddt.ddt
class NotificationListAPIViewTest(APITestCase):
"""
@@ -1108,361 +546,6 @@ def remove_notifications_with_visibility_settings(expected_response):
return expected_response
@ddt.ddt
class UpdateAllNotificationPreferencesViewTests(APITestCase):
"""
Tests for the UpdateAllNotificationPreferencesView.
"""
def setUp(self):
# Create test user
self.user = User.objects.create_user(
username='testuser',
password='testpass123'
)
self.client = APIClient()
self.client.force_authenticate(user=self.user)
self.url = reverse('update-all-notification-preferences')
# Complex notification config structure
self.base_config = {
"grading": {
"enabled": True,
"non_editable": {},
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
},
"ora_staff_notifications": {
"web": False,
"push": False,
"email": False,
"email_cadence": "Daily"
}
},
"core_notification_types": []
},
"updates": {
"enabled": True,
"non_editable": {},
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
},
"course_updates": {
"web": True,
"push": True,
"email": False,
"email_cadence": "Daily"
}
},
"core_notification_types": []
},
"discussion": {
"enabled": True,
"non_editable": {
"core": ["web"]
},
"notification_types": {
"core": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
},
"content_reported": {
"web": True,
"push": True,
"email": True,
"email_cadence": "Daily"
},
"new_question_post": {
"web": True,
"push": False,
"email": False,
"email_cadence": "Daily"
},
"new_discussion_post": {
"web": True,
"push": False,
"email": False,
"email_cadence": "Daily"
}
},
"core_notification_types": [
"new_comment_on_response",
"new_comment",
"new_response",
"response_on_followed_post",
"comment_on_followed_post",
"response_endorsed_on_thread",
"response_endorsed"
]
}
}
# Create test notification preferences
self.preferences = []
for i in range(3):
pref = CourseNotificationPreference.objects.create(
user=self.user,
course_id=f'course-v1:TestX+Test{i}+2024',
notification_preference_config=deepcopy(self.base_config),
is_active=True
)
self.preferences.append(pref)
# Create an inactive preference
self.inactive_pref = CourseNotificationPreference.objects.create(
user=self.user,
course_id='course-v1:TestX+Inactive+2024',
notification_preference_config=deepcopy(self.base_config),
is_active=False
)
def test_update_discussion_notification(self):
"""
Test updating discussion notification settings
"""
data = {
'notification_app': 'discussion',
'notification_type': 'core',
'notification_channel': 'web',
'value': False
}
response = self.client.post(self.url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['status'], 'success')
self.assertEqual(response.data['data']['total_updated'], 3)
# Verify database updates
for pref in CourseNotificationPreference.objects.filter(is_active=True):
self.assertFalse(
pref.notification_preference_config['discussion'][
'notification_types']['core']['web']
)
def test_update_non_editable_field(self):
"""
Test attempting to update a non-editable field
"""
data = {
'notification_app': 'discussion',
'notification_type': 'core',
'notification_channel': 'web',
'value': False
}
response = self.client.post(self.url, data, format='json')
# Should fail because 'web' is non-editable for 'core' in discussion
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['status'], 'success')
# Verify database remains unchanged
for pref in CourseNotificationPreference.objects.filter(is_active=True):
self.assertFalse(
pref.notification_preference_config['discussion']['notification_types']['core']['web']
)
def test_update_email_cadence(self):
"""
Test updating email cadence setting
"""
data = {
'notification_app': 'discussion',
'notification_type': 'content_reported',
'email_cadence': 'Weekly'
}
response = self.client.post(self.url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['status'], 'success')
# Verify database updates
for pref in CourseNotificationPreference.objects.filter(is_active=True):
notification_type = pref.notification_preference_config['discussion']['notification_types'][
'content_reported']
self.assertEqual(
notification_type['email_cadence'],
'Weekly'
)
@patch.dict('openedx.core.djangoapps.notifications.serializers.COURSE_NOTIFICATION_APPS', {
**COURSE_NOTIFICATION_APPS,
'grading': {
'enabled': False,
'core_info': 'Notifications for submission grading.',
'core_web': True,
'core_email': True,
'core_push': True,
'core_email_cadence': 'Daily',
'non_editable': []
}
})
def test_update_disabled_app(self):
"""
Test updating notification for a disabled app
"""
# Disable the grading app in all preferences
for pref in self.preferences:
config = pref.notification_preference_config
config['grading']['enabled'] = False
pref.notification_preference_config = config
pref.save()
data = {
'notification_app': 'grading',
'notification_type': 'core',
'notification_channel': 'email',
'value': False
}
response = self.client.post(self.url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data['status'], 'error')
def test_invalid_serializer_data(self):
"""
Test handling of invalid input data
"""
test_cases = [
{
'notification_app': 'invalid_app',
'notification_type': 'core',
'notification_channel': 'push',
'value': False
},
{
'notification_app': 'discussion',
'notification_type': 'invalid_type',
'notification_channel': 'push',
'value': False
},
{
'notification_app': 'discussion',
'notification_type': 'core',
'notification_channel': 'invalid_channel',
'value': False
},
{
'notification_app': 'discussion',
'notification_type': 'core',
'notification_channel': 'email_cadence',
'value': 'Invalid_Cadence'
}
]
for test_case in test_cases:
response = self.client.post(self.url, test_case, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
@ddt.data(*itertools.product(('email', 'web'), (True, False)))
@ddt.unpack
def test_unsub_user_preferences_removal_on_account_email_enabled(self, channel, value):
"""
Test one click unsub user preference should be removed on email enable for any app through account preferences
"""
UserPreference.objects.create(user=self.user, key=ONE_CLICK_EMAIL_UNSUB_KEY)
payload = {
'notification_app': 'grading',
'notification_type': 'core',
'notification_channel': channel,
'value': value
}
self.client.post(self.url, payload, format='json')
result = 0 if channel == 'email' and value else 1
self.assertEqual(UserPreference.objects.count(), result)
class GetAggregateNotificationPreferencesTest(APITestCase):
"""
Tests for the GetAggregateNotificationPreferences API view.
"""
def setUp(self):
# Set up a user and API client
self.user = User.objects.create_user(username='testuser', password='testpass')
self.client = APIClient()
self.client.force_authenticate(user=self.user)
self.url = reverse('notification-preferences-aggregated') # Adjust with the actual name
def test_no_active_notification_preferences(self):
"""
Test case: No active notification preferences found for the user
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertEqual(response.data['status'], 'error')
self.assertEqual(response.data['message'], 'No active notification preferences found')
@patch('openedx.core.djangoapps.notifications.views.aggregate_notification_configs')
def test_with_active_notification_preferences(self, mock_aggregate):
"""
Test case: Active notification preferences found for the user
"""
# Mock aggregate_notification_configs for a controlled output
mock_aggregate.return_value = {'mocked': {'notification_types': {}}}
# Create active notification preferences for the user
CourseNotificationPreference.objects.create(
user=self.user,
is_active=True,
notification_preference_config={'example': 'config'}
)
response = self.client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['status'], 'success')
self.assertEqual(response.data['message'], 'Notification preferences retrieved')
self.assertDictEqual(response.data['data'], {'mocked': {'notification_types': {}, 'non_editable': {}}})
def test_unauthenticated_user(self):
"""
Test case: Request without authentication
"""
# Test case: Request without authentication
self.client.logout() # Remove authentication
response = self.client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
@mock.patch.dict(COURSE_NOTIFICATION_APPS, {
**COURSE_NOTIFICATION_APPS,
**{
'discussion': {
'name': 'content_reported',
'non_editable': ["web"]
}
}
})
@mock.patch.dict(COURSE_NOTIFICATION_TYPES, {
**COURSE_NOTIFICATION_TYPES,
**{
'course_updates': {
**COURSE_NOTIFICATION_TYPES['course_updates'],
'non_editable': ["email"]
}
}
})
def test_non_editable_is_added_in_api_response(self):
CourseNotificationPreference.objects.create(user=self.user, is_active=True)
response = self.client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
prefs = response.data['data']
self.assertDictEqual(prefs['updates']['non_editable'], {'course_updates': ['email']})
self.assertDictEqual(prefs['discussion']['non_editable'], {
'new_discussion_post': ['push'],
'new_question_post': ['push'],
'new_instructor_all_learners_post': ['push'],
'core': ['web']
})
@ddt.ddt
class TestNotificationPreferencesView(ModuleStoreTestCase):
"""

25
openedx/core/djangoapps/notifications/urls.py
View File

@@ -1,35 +1,21 @@
"""
URLs for the notifications API.
"""
from django.conf import settings
from django.urls import path, re_path
from django.urls import path
from rest_framework import routers
from .views import (
CourseEnrollmentListView,
MarkNotificationsSeenAPIView,
NotificationCountView,
NotificationListAPIView,
NotificationReadAPIView,
UpdateAllNotificationPreferencesView,
UserNotificationPreferenceView,
preference_update_from_encrypted_username_view, AggregatedNotificationPreferences, NotificationPreferencesView
preference_update_from_encrypted_username_view,
NotificationPreferencesView,
)
router = routers.DefaultRouter()
urlpatterns = [
path('enrollments/', CourseEnrollmentListView.as_view(), name='enrollment-list'),
re_path(
fr'^configurations/{settings.COURSE_KEY_PATTERN}$',
UserNotificationPreferenceView.as_view(),
name='notification-preferences'
),
path(
'configurations/',
AggregatedNotificationPreferences.as_view(),
name='notification-preferences-aggregated'
),
path(
'v2/configurations/',
NotificationPreferencesView.as_view(),
@@ -45,11 +31,6 @@ urlpatterns = [
path('read/', NotificationReadAPIView.as_view(), name='notifications-read'),
path('preferences/update/<str:username>/<str:patch>/', preference_update_from_encrypted_username_view,
name='preference_update_from_encrypted_username_view'),
path(
'preferences/update-all/',
UpdateAllNotificationPreferencesView.as_view(),
name='update-all-notification-preferences'
),
]
urlpatterns += router.urls

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

@@ -1,13 +1,11 @@
"""
Utils function for notifications app
"""
import copy
from typing import Dict, List, Set
from common.djangoapps.student.models import CourseAccessRole, CourseEnrollment
from openedx.core.djangoapps.django_comment_common.models import Role
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
from openedx.core.lib.cache_utils import request_cached
@@ -230,64 +228,6 @@ def process_app_config(
update_notification_types(app_config, user_app_config)
def aggregate_notification_configs(existing_user_configs: List[Dict]) -> Dict:
"""
Update default notification config with values from other configs.
Rules:
1. Start with default config as base
2. If any value is True in other configs, make it True
3. Set email_cadence to "Mixed" if different cadences found, else use default
Args:
existing_user_configs: List of notification config dictionaries to apply
Returns:
Updated config following the same structure
"""
if not existing_user_configs:
return {}
result_config = copy.deepcopy(existing_user_configs[0])
apps = result_config.keys()
for app in apps:
app_config = result_config[app]
for user_config in existing_user_configs:
process_app_config(app_config, user_config, app, existing_user_configs[0])
# if email_cadence is mixed, set it to "Mixed"
for app in result_config:
for type_key, type_config in result_config[app]["notification_types"].items():
if len(type_config.get("email_cadence", [])) > 1:
result_config[app]["notification_types"][type_key]["email_cadence"] = "Mixed"
else:
if result_config[app]["notification_types"][type_key].get('email_cadence'):
result_config[app]["notification_types"][type_key]["email_cadence"] = (
result_config[app]["notification_types"][type_key]["email_cadence"].pop())
else:
result_config[app]["notification_types"][type_key]["email_cadence"] = EmailCadence.DAILY
return result_config
def filter_out_visible_preferences_by_course_ids(user, preferences: Dict, course_ids: List) -> Dict:
"""
Filter out notifications visible to forum roles from user preferences.
"""
forum_roles = Role.objects.filter(users__id=user.id).values_list('name', flat=True)
course_roles = CourseAccessRole.objects.filter(
user=user,
course_id__in=course_ids
).values_list('role', flat=True)
notification_types_with_visibility = get_notification_types_with_visibility_settings()
return filter_out_visible_notifications(
preferences,
notification_types_with_visibility,
forum_roles,
course_roles
)
def get_user_forum_access_roles(user_id: int) -> List[str]:
"""
Get forum roles for the given user in all course.

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

@@ -1,15 +1,12 @@
"""
Views for the notifications API.
"""
import copy
from datetime import datetime, timedelta
from django.conf import settings
from django.db import transaction
from django.db.models import Count
from django.shortcuts import get_object_or_404
from django.utils.translation import gettext as _
from opaque_keys.edx.keys import CourseKey
from pytz import UTC
from rest_framework import generics, status
from rest_framework.decorators import api_view
@@ -17,245 +14,31 @@ from rest_framework.generics import UpdateAPIView
from rest_framework.response import Response
from rest_framework.views import APIView
from common.djangoapps.student.models import CourseEnrollment
from openedx.core.djangoapps.notifications.email import ONE_CLICK_EMAIL_UNSUB_KEY
from openedx.core.djangoapps.notifications.email.utils import update_user_preferences_from_patch
from openedx.core.djangoapps.notifications.models import get_course_notification_preference_config_version, \
NotificationPreference
from openedx.core.djangoapps.notifications.models import NotificationPreference
from openedx.core.djangoapps.notifications.permissions import allow_any_authenticated_user
from openedx.core.djangoapps.notifications.serializers import add_info_to_notification_config
from openedx.core.djangoapps.user_api.models import UserPreference
from .base_notification import COURSE_NOTIFICATION_APPS, NotificationAppManager, COURSE_NOTIFICATION_TYPES, \
NotificationTypeManager
from .config.waffle import ENABLE_NOTIFICATIONS
from .events import (
notification_preference_update_event,
notification_preferences_viewed_event,
notification_read_event,
notification_tray_opened_event,
notifications_app_all_read_event
)
from .models import CourseNotificationPreference, Notification
from .models import Notification
from .serializers import (
NotificationCourseEnrollmentSerializer,
NotificationSerializer,
UserCourseNotificationPreferenceSerializer,
UserNotificationPreferenceUpdateAllSerializer,
UserNotificationPreferenceUpdateSerializer,
add_non_editable_in_preference
)
from .tasks import create_notification_preference
from .utils import (
aggregate_notification_configs,
filter_out_visible_preferences_by_course_ids,
get_show_notifications_tray,
exclude_inaccessible_preferences
)
@allow_any_authenticated_user()
class CourseEnrollmentListView(generics.ListAPIView):
"""
API endpoint to get active CourseEnrollments for requester.
**Permissions**: User must be authenticated.
**Response Format** (paginated):
{
"next": (str) url_to_next_page_of_courses,
"previous": (str) url_to_previous_page_of_courses,
"count": (int) total_number_of_courses,
"num_pages": (int) total_number_of_pages,
"current_page": (int) current_page_number,
"start": (int) index_of_first_course_on_page,
"results" : [
{
"course": {
"id": (int) course_id,
"display_name": (str) course_display_name
},
},
...
],
}
Response Error Codes:
- 403: The requester cannot access resource.
"""
serializer_class = NotificationCourseEnrollmentSerializer
def get_paginated_response(self, data):
"""
Return a response given serialized page data with show_preferences flag.
"""
response = super().get_paginated_response(data)
response.data["show_preferences"] = get_show_notifications_tray(self.request.user)
return response
def get_queryset(self):
user = self.request.user
return CourseEnrollment.objects.filter(user=user, is_active=True)
def list(self, request, *args, **kwargs):
"""
Returns the list of active course enrollments for which ENABLE_NOTIFICATIONS
Waffle flag is enabled
"""
queryset = self.filter_queryset(self.get_queryset())
course_ids = queryset.values_list('course_id', flat=True)
for course_id in course_ids:
if not ENABLE_NOTIFICATIONS.is_enabled(course_id):
queryset = queryset.exclude(course_id=course_id)
queryset = queryset.select_related('course').order_by('-id')
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
return Response({
"show_preferences": get_show_notifications_tray(request.user),
"results": self.get_serializer(queryset, many=True).data
})
@allow_any_authenticated_user()
class UserNotificationPreferenceView(APIView):
"""
Supports retrieving and patching the UserNotificationPreference
model.
**Example Requests**
GET /api/notifications/configurations/{course_id}
PATCH /api/notifications/configurations/{course_id}
**Example Response**:
{
'id': 1,
'course_name': 'testcourse',
'course_id': 'course-v1:testorg+testcourse+testrun',
'notification_preference_config': {
'discussion': {
'enabled': False,
'core': {
'info': '',
'web': False,
'push': False,
'email': False,
},
'notification_types': {
'new_post': {
'info': '',
'web': False,
'push': False,
'email': False,
},
},
'not_editable': {},
},
}
}
"""
def get(self, request, course_key_string):
"""
Returns notification preference for user for a course.
Parameters:
request (Request): The request object.
course_key_string (int): The ID of the course to retrieve notification preference.
Returns:
{
'id': 1,
'course_name': 'testcourse',
'course_id': 'course-v1:testorg+testcourse+testrun',
'notification_preference_config': {
'discussion': {
'enabled': False,
'core': {
'info': '',
'web': False,
'push': False,
'email': False,
},
'notification_types': {
'new_post': {
'info': '',
'web': False,
'push': False,
'email': False,
},
},
'not_editable': {},
},
}
}
"""
course_id = CourseKey.from_string(course_key_string)
user_preference = CourseNotificationPreference.get_updated_user_course_preferences(request.user, course_id)
serializer_context = {
'course_id': course_id,
'user': request.user
}
serializer = UserCourseNotificationPreferenceSerializer(user_preference, context=serializer_context)
notification_preferences_viewed_event(request, course_id)
return Response(serializer.data)
def patch(self, request, course_key_string):
"""
Update an existing user notification preference with the data in the request body.
Parameters:
request (Request): The request object
course_key_string (int): The ID of the course of the notification preference to be updated.
Returns:
200: The updated preference, serialized using the UserNotificationPreferenceSerializer
404: If the preference does not exist
403: If the user does not have permission to update the preference
400: Validation error
"""
course_id = CourseKey.from_string(course_key_string)
user_course_notification_preference = CourseNotificationPreference.objects.get(
user=request.user,
course_id=course_id,
is_active=True,
)
if user_course_notification_preference.config_version != get_course_notification_preference_config_version():
return Response(
{'error': _('The notification preference config version is not up to date.')},
status=status.HTTP_409_CONFLICT,
)
if request.data.get('notification_channel', '') == 'email_cadence':
request.data['email_cadence'] = request.data['value']
del request.data['value']
preference_update = UserNotificationPreferenceUpdateSerializer(
user_course_notification_preference, data=request.data, partial=True
)
preference_update.is_valid(raise_exception=True)
updated_notification_preferences = preference_update.save()
if request.data.get('notification_channel', '') == 'email' and request.data.get('value', False):
UserPreference.objects.filter(
user_id=request.user.id,
key=ONE_CLICK_EMAIL_UNSUB_KEY
).delete()
notification_preference_update_event(request.user, course_id, preference_update.validated_data)
serializer_context = {
'course_id': course_id,
'user': request.user
}
serializer = UserCourseNotificationPreferenceSerializer(updated_notification_preferences,
context=serializer_context)
return Response(serializer.data, status=status.HTTP_200_OK)
@allow_any_authenticated_user()
class NotificationListAPIView(generics.ListAPIView):
"""
@@ -462,170 +245,6 @@ def preference_update_from_encrypted_username_view(request, username, patch):
return Response({"result": "success"}, status=status.HTTP_200_OK)
@allow_any_authenticated_user()
class UpdateAllNotificationPreferencesView(APIView):
"""
API view for updating all notification preferences for the current user.
"""
def post(self, request):
"""
Update all notification preferences for the current user.
"""
# check if request have required params
serializer = UserNotificationPreferenceUpdateAllSerializer(data=request.data)
if not serializer.is_valid():
return Response({
'status': 'error',
'message': serializer.errors
}, status=status.HTTP_400_BAD_REQUEST)
# check if required config is not editable
try:
with transaction.atomic():
# Get all active notification preferences for the current user
notification_preferences = (
CourseNotificationPreference.objects
.select_for_update()
.filter(
user=request.user,
is_active=True
)
)
if not notification_preferences.exists():
return Response({
'status': 'error',
'message': 'No active notification preferences found'
}, status=status.HTTP_404_NOT_FOUND)
data = serializer.validated_data
app = data['notification_app']
email_cadence = data.get('email_cadence', None)
channel = data.get('notification_channel', 'email_cadence' if email_cadence else None)
notification_type = data['notification_type']
value = data.get('value', email_cadence if email_cadence else None)
updated_courses = []
errors = []
# Update each preference
for preference in notification_preferences:
try:
# Create a deep copy of the current config
updated_config = copy.deepcopy(preference.notification_preference_config)
# Check if the path exists and update the value
if (
updated_config.get(app, {})
.get('notification_types', {})
.get(notification_type, {})
.get(channel)
) is not None:
# Update the specific setting in the config
updated_config[app]['notification_types'][notification_type][channel] = value
# Update the notification preference
preference.notification_preference_config = updated_config
preference.save()
updated_courses.append({
'course_id': str(preference.course_id),
'current_setting': updated_config[app]['notification_types'][notification_type]
})
else:
errors.append({
'course_id': str(preference.course_id),
'error': f'Invalid path: {app}.notification_types.{notification_type}.{channel}'
})
except (KeyError, AttributeError, ValueError) as e:
errors.append({
'course_id': str(preference.course_id),
'error': str(e)
})
if channel == 'email' and value:
UserPreference.objects.filter(
user_id=request.user,
key=ONE_CLICK_EMAIL_UNSUB_KEY
).delete()
response_data = {
'status': 'success' if updated_courses else 'partial_success' if errors else 'error',
'message': 'Notification preferences update completed',
'data': {
'updated_value': value,
'notification_type': notification_type,
'channel': channel,
'app': app,
'successfully_updated_courses': updated_courses,
'total_updated': len(updated_courses),
'total_courses': notification_preferences.count()
}
}
if errors:
response_data['errors'] = errors
event_data = {
'notification_app': app,
'notification_type': notification_type,
'notification_channel': channel,
'value': value,
'email_cadence': value
}
notification_preference_update_event(
request.user,
[course['course_id'] for course in updated_courses],
event_data
)
return Response(
response_data,
status=status.HTTP_200_OK if updated_courses else status.HTTP_400_BAD_REQUEST
)
except (KeyError, AttributeError, ValueError) as e:
return Response({
'status': 'error',
'message': str(e)
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
@allow_any_authenticated_user()
class AggregatedNotificationPreferences(APIView):
"""
API view for getting the aggregate notification preferences for the current user.
"""
def get(self, request):
"""
API view for getting the aggregate notification preferences for the current user.
"""
notification_preferences = CourseNotificationPreference.get_user_notification_preferences(request.user)
if not notification_preferences.exists():
return Response({
'status': 'error',
'message': 'No active notification preferences found'
}, status=status.HTTP_404_NOT_FOUND)
notification_configs = notification_preferences.values_list('notification_preference_config', flat=True)
notification_configs = aggregate_notification_configs(
notification_configs
)
course_ids = notification_preferences.values_list('course_id', flat=True)
filter_out_visible_preferences_by_course_ids(
request.user,
notification_configs,
course_ids,
)
notification_preferences_viewed_event(request)
notification_configs = add_info_to_notification_config(notification_configs)
return Response({
'status': 'success',
'message': 'Notification preferences retrieved',
'data': add_non_editable_in_preference(notification_configs)
}, status=status.HTTP_200_OK)
@allow_any_authenticated_user()
class NotificationPreferencesView(APIView):
"""