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

feat: updated one click unsubscribe to use account level preference model (#37161)

Browse Source
This commit is contained in:
Muhammad Adeel Tajamul
2025-08-12 16:44:42 +05:00
committed by GitHub
parent 983cdf9274
commit e8b58f770e
6 changed files with 93 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -6,7 +6,6 @@ import ddt
import pytest
from django.http.response import Http404
from itertools import product
from pytz import utc
from waffle import get_waffle_flag_model # pylint: disable=invalid-django-waffle-import
@@ -24,9 +23,7 @@ from openedx.core.djangoapps.notifications.email.utils import (
create_datetime_string,
create_email_digest_context,
create_email_template_context,
decrypt_object,
decrypt_string,
encrypt_object,
encrypt_string,
get_course_info,
get_time_ago,
@@ -241,17 +238,6 @@ class TestEncryption(ModuleStoreTestCase):
decrypted = decrypt_string(encrypted)
assert string == decrypted
def test_object_encryption(self):
"""
Tests if decrypted object is equal to original object
"""
obj = {
'org': 'edx'
}
encrypted = encrypt_object(obj)
decrypted = decrypt_object(encrypted)
assert obj == decrypted
@ddt.ddt
class TestUpdatePreferenceFromPatch(ModuleStoreTestCase):
@@ -289,173 +275,21 @@ class TestUpdatePreferenceFromPatch(ModuleStoreTestCase):
return COURSE_NOTIFICATION_APPS[app_name]['core_email_cadence']
return COURSE_NOTIFICATION_TYPES[notification_type]['email_cadence']
@ddt.data(True, False)
def test_value_param(self, new_value):
"""
Tests if value is updated for all notification types and for all channels
"""
encrypted_username = encrypt_string(self.user.username)
encrypted_patch = encrypt_object({
'value': new_value
})
update_user_preferences_from_patch(encrypted_username, encrypted_patch)
preference_1 = CourseNotificationPreference.objects.get(course_id=self.course_1.id, user=self.user)
preference_2 = CourseNotificationPreference.objects.get(course_id=self.course_2.id, user=self.user)
for preference in [preference_1, preference_2]:
config = preference.notification_preference_config
for app_name, app_prefs in config.items():
for noti_type, type_prefs in app_prefs['notification_types'].items():
for channel in ['web', 'email', 'push']:
if self.is_channel_editable(app_name, noti_type, channel):
assert type_prefs[channel] == new_value
else:
default_app_json = self.default_json[app_name]
default_notification_type_json = default_app_json['notification_types'][noti_type]
assert type_prefs[channel] == default_notification_type_json[channel]
@ddt.data(*product(['web', 'email', 'push'], [True, False]))
@ddt.unpack
def test_value_with_channel_param(self, param_channel, new_value):
"""
Tests if value is updated only for channel
"""
encrypted_username = encrypt_string(self.user.username)
encrypted_patch = encrypt_object({
'channel': param_channel,
'value': new_value
})
update_user_preferences_from_patch(encrypted_username, encrypted_patch)
preference_1 = CourseNotificationPreference.objects.get(course_id=self.course_1.id, user=self.user)
preference_2 = CourseNotificationPreference.objects.get(course_id=self.course_2.id, user=self.user)
# pylint: disable=too-many-nested-blocks
for preference in [preference_1, preference_2]:
config = preference.notification_preference_config
for app_name, app_prefs in config.items():
for noti_type, type_prefs in app_prefs['notification_types'].items():
for channel in ['web', 'email', 'push']:
if not self.is_channel_editable(app_name, noti_type, channel):
continue
if channel == param_channel:
assert type_prefs[channel] == new_value
if channel == 'email':
cadence_value = self.get_default_cadence_value(app_name, noti_type)
assert type_prefs['email_cadence'] == cadence_value
else:
default_app_json = self.default_json[app_name]
default_notification_type_json = default_app_json['notification_types'][noti_type]
assert type_prefs[channel] == default_notification_type_json[channel]
@ddt.data(True, False)
def test_value_with_course_id_param(self, new_value):
"""
Tests if value is updated for a single course only
"""
encrypted_username = encrypt_string(self.user.username)
encrypted_patch = encrypt_object({
'value': new_value,
'course_id': str(self.course_1.id),
})
update_user_preferences_from_patch(encrypted_username, encrypted_patch)
preference_2 = CourseNotificationPreference.objects.get(course_id=self.course_2.id, user=self.user)
self.assertDictEqual(preference_2.notification_preference_config, self.default_json)
preference_1 = CourseNotificationPreference.objects.get(course_id=self.course_1.id, user=self.user)
config = preference_1.notification_preference_config
for app_name, app_prefs in config.items():
for noti_type, type_prefs in app_prefs['notification_types'].items():
for channel in ['web', 'email', 'push']:
if self.is_channel_editable(app_name, noti_type, channel):
assert type_prefs[channel] == new_value
else:
default_app_json = self.default_json[app_name]
default_notification_type_json = default_app_json['notification_types'][noti_type]
assert type_prefs[channel] == default_notification_type_json[channel]
@ddt.data(*product(['discussion', 'updates'], [True, False]))
@ddt.unpack
def test_value_with_app_name_param(self, param_app_name, new_value):
"""
Tests if value is updated only for channel
"""
encrypted_username = encrypt_string(self.user.username)
encrypted_patch = encrypt_object({
'app_name': param_app_name,
'value': new_value
})
update_user_preferences_from_patch(encrypted_username, encrypted_patch)
preference_1 = CourseNotificationPreference.objects.get(course_id=self.course_1.id, user=self.user)
preference_2 = CourseNotificationPreference.objects.get(course_id=self.course_2.id, user=self.user)
# pylint: disable=too-many-nested-blocks
for preference in [preference_1, preference_2]:
config = preference.notification_preference_config
for app_name, app_prefs in config.items():
for noti_type, type_prefs in app_prefs['notification_types'].items():
for channel in ['web', 'email', 'push']:
if not self.is_channel_editable(app_name, noti_type, channel):
continue
if app_name == param_app_name:
assert type_prefs[channel] == new_value
if channel == 'email':
cadence_value = self.get_default_cadence_value(app_name, noti_type)
assert type_prefs['email_cadence'] == cadence_value
else:
default_app_json = self.default_json[app_name]
default_notification_type_json = default_app_json['notification_types'][noti_type]
assert type_prefs[channel] == default_notification_type_json[channel]
@ddt.data(*product(['new_discussion_post', 'content_reported'], [True, False]))
@ddt.unpack
def test_value_with_notification_type_param(self, param_notification_type, new_value):
"""
Tests if value is updated only for channel
"""
encrypted_username = encrypt_string(self.user.username)
encrypted_patch = encrypt_object({
'notification_type': param_notification_type,
'value': new_value
})
update_user_preferences_from_patch(encrypted_username, encrypted_patch)
preference_1 = CourseNotificationPreference.objects.get(course_id=self.course_1.id, user=self.user)
preference_2 = CourseNotificationPreference.objects.get(course_id=self.course_2.id, user=self.user)
# pylint: disable=too-many-nested-blocks
for preference in [preference_1, preference_2]:
config = preference.notification_preference_config
for app_name, app_prefs in config.items():
for noti_type, type_prefs in app_prefs['notification_types'].items():
for channel in ['web', 'email', 'push']:
if not self.is_channel_editable(app_name, noti_type, channel):
continue
if noti_type == param_notification_type:
assert type_prefs[channel] == new_value
if channel == 'email':
cadence_value = self.get_default_cadence_value(app_name, noti_type)
assert type_prefs['email_cadence'] == cadence_value
else:
default_app_json = self.default_json[app_name]
default_notification_type_json = default_app_json['notification_types'][noti_type]
assert type_prefs[channel] == default_notification_type_json[channel]
def test_preference_not_updated_if_invalid_username(self):
"""
Tests if no preference is updated when username is not valid
"""
username = f"{self.user.username}-updated"
enc_username = encrypt_string(username)
enc_patch = encrypt_object({"value": True})
with pytest.raises(Http404):
update_user_preferences_from_patch(enc_username, enc_patch)
update_user_preferences_from_patch(enc_username)
def test_user_preference_created_on_email_unsubscribe(self):
"""
Test that the user's email unsubscribe preference is correctly created after unsubscribing digest email.
"""
encrypted_username = encrypt_string(self.user.username)
encrypted_patch = encrypt_object({
'channel': 'email',
'value': False
})
update_user_preferences_from_patch(encrypted_username, encrypted_patch)
update_user_preferences_from_patch(encrypted_username)
self.assertTrue(
UserPreference.objects.filter(user=self.user, key=ONE_CLICK_EMAIL_UNSUB_KEY).exists()
)

106
openedx/core/djangoapps/notifications/email/utils.py
View File

@@ -2,7 +2,6 @@
Email Notifications Utils
"""
import datetime
import json
from bs4 import BeautifulSoup
from django.conf import settings
@@ -12,7 +11,6 @@ from django.utils.translation import gettext as _
from pytz import utc
from waffle import get_waffle_flag_model # pylint: disable=invalid-django-waffle-import
from common.djangoapps.student.models import CourseEnrollment
from lms.djangoapps.branding.api import get_logo_url_for_email
from lms.djangoapps.discussion.notification_prefs.views import UsernameCipher
from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY
@@ -21,11 +19,7 @@ from openedx.core.djangoapps.notifications.config.waffle import ENABLE_EMAIL_NOT
from openedx.core.djangoapps.notifications.email import ONE_CLICK_EMAIL_UNSUB_KEY
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
from openedx.core.djangoapps.notifications.events import notification_preference_unsubscribe_event
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
NotificationPreference,
get_course_notification_preference_config_version
)
from openedx.core.djangoapps.notifications.models import NotificationPreference
from openedx.core.djangoapps.user_api.models import UserPreference
from xmodule.modulestore.django import modulestore
@@ -71,13 +65,12 @@ def get_icon_url_for_notification_type(notification_type):
return NotificationTypeIcons.get_icon_url_for_notification_type(notification_type)
def get_unsubscribe_link(username, patch):
def get_unsubscribe_link(username):
"""
Returns unsubscribe url for username with patch preferences
"""
encrypted_username = encrypt_string(username)
encrypted_patch = encrypt_object(patch)
return f"{settings.LEARNING_MICROFRONTEND_URL}/preferences-unsubscribe/{encrypted_username}/{encrypted_patch}"
return f"{settings.LEARNING_MICROFRONTEND_URL}/preferences-unsubscribe/{encrypted_username}/"
def create_email_template_context(username):
@@ -106,7 +99,7 @@ def create_email_template_context(username):
"logo_notification_cadence_url": settings.NOTIFICATION_DIGEST_LOGO,
"social_media": social_media_info,
"notification_settings_url": f"{account_base_url}/#notifications",
"unsubscribe_url": get_unsubscribe_link(username, patch)
"unsubscribe_url": get_unsubscribe_link(username)
}
@@ -391,23 +384,7 @@ def decrypt_string(string):
return UsernameCipher.decrypt(string).decode()
def encrypt_object(obj):
"""
Returns hashed string of object
"""
string = json.dumps(obj)
return encrypt_string(string)
def decrypt_object(string):
"""
Decrypts input string and returns an object
"""
decoded = decrypt_string(string)
return json.loads(decoded)
def update_user_preferences_from_patch(encrypted_username, encrypted_patch):
def update_user_preferences_from_patch(encrypted_username):
"""
Decrypt username and patch and updates user preferences
Allowed parameters for decrypted patch
@@ -418,78 +395,15 @@ def update_user_preferences_from_patch(encrypted_username, encrypted_patch):
course_id: course key string
"""
username = decrypt_string(encrypted_username)
patch = decrypt_object(encrypted_patch)
app_value = patch.get("app_name")
type_value = patch.get("notification_type")
channel_value = patch.get("channel")
pref_value = bool(patch.get("value", False))
user = get_object_or_404(User, username=username)
kwargs = {'user': user}
if 'course_id' in patch.keys():
kwargs['course_id'] = patch['course_id']
NotificationPreference.create_default_preferences_for_user(user.id)
def is_name_match(name, param_name):
"""
Name is match if strings are equal or param_name is None
"""
return True if param_name is None else name == param_name
updated_count = NotificationPreference.objects.filter(user=user, email=True).update(email=False)
is_preference_updated = updated_count > 0
def get_default_cadence_value(app_name, notification_type):
"""
Returns default email cadence value
"""
if notification_type == 'core':
return COURSE_NOTIFICATION_APPS[app_name]['core_email_cadence']
return COURSE_NOTIFICATION_TYPES[notification_type]['email_cadence']
def get_updated_preference(pref):
"""
Update preference if config version doesn't match
"""
if pref.config_version != get_course_notification_preference_config_version():
pref = pref.get_user_course_preference(pref.user_id, pref.course_id)
return pref
course_ids = CourseEnrollment.objects.filter(user=user, is_active=True).values_list('course_id', flat=True)
CourseNotificationPreference.objects.bulk_create(
[
CourseNotificationPreference(user=user, course_id=course_id)
for course_id in course_ids
],
ignore_conflicts=True
)
preferences = CourseNotificationPreference.objects.filter(**kwargs)
is_preference_updated = False
# pylint: disable=too-many-nested-blocks
for preference in preferences:
preference = get_updated_preference(preference)
preference_json = preference.notification_preference_config
for app_name, app_prefs in preference_json.items():
if not is_name_match(app_name, app_value):
continue
for noti_type, type_prefs in app_prefs['notification_types'].items():
if not is_name_match(noti_type, type_value):
continue
for channel in ['web', 'email', 'push']:
if not is_name_match(channel, channel_value):
continue
if is_notification_type_channel_editable(app_name, noti_type, channel):
if type_prefs[channel] != pref_value:
type_prefs[channel] = pref_value
is_preference_updated = True
if channel == 'email' and pref_value and type_prefs.get('email_cadence') == EmailCadence.NEVER:
default_cadence = get_default_cadence_value(app_name, noti_type)
if type_prefs['email_cadence'] != default_cadence:
type_prefs['email_cadence'] = default_cadence
is_preference_updated = True
preference.save()
notification_preference_unsubscribe_event(user, is_preference_updated)
if app_value is None and type_value is None and channel_value == 'email' and not pref_value:
UserPreference.objects.get_or_create(user_id=user.id, key=ONE_CLICK_EMAIL_UNSUB_KEY)
notification_preference_unsubscribe_event(user, is_preference_updated)
UserPreference.objects.get_or_create(user_id=user.id, key=ONE_CLICK_EMAIL_UNSUB_KEY)
def is_notification_type_channel_editable(app_name, notification_type, channel):

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

@@ -12,7 +12,9 @@ from opaque_keys.edx.django.models import CourseKeyField
from openedx.core.djangoapps.notifications.base_notification import (
NotificationAppManager,
NotificationPreferenceSyncManager,
get_notification_content
get_notification_content,
COURSE_NOTIFICATION_APPS,
COURSE_NOTIFICATION_TYPES
)
from openedx.core.djangoapps.notifications.email import ONE_CLICK_EMAIL_UNSUB_KEY
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
@@ -96,6 +98,41 @@ def get_additional_notification_channel_settings():
return ADDITIONAL_NOTIFICATION_CHANNEL_SETTINGS
def create_notification_preference(user_id: int, notification_type: str):
"""
Create a single notification preference with appropriate defaults.
Args:
user_id: ID of the user
notification_type: Type of notification
Returns:
NotificationPreference instance
"""
notification_config = COURSE_NOTIFICATION_TYPES.get(notification_type, {})
is_core = notification_config.get('is_core', False)
app = notification_config['notification_app']
kwargs = {
"web": notification_config.get('web', True),
"push": notification_config.get('push', False),
"email": notification_config.get('email', False),
"email_cadence": notification_config.get('email_cadence', EmailCadence.DAILY),
}
if is_core:
app_config = COURSE_NOTIFICATION_APPS[app]
kwargs = {
"web": app_config.get("core_web", True),
"push": app_config.get("core_push", False),
"email": app_config.get("core_email", False),
"email_cadence": app_config.get("core_email_cadence", EmailCadence.DAILY),
}
return NotificationPreference(
user_id=user_id,
type=notification_type,
app=app,
**kwargs,
)
class Notification(TimeStampedModel):
"""
Model to store notifications for users
@@ -149,6 +186,30 @@ class NotificationPreference(TimeStampedModel):
email_cadence = models.CharField(max_length=64, choices=EmailCadenceChoices.choices, null=False, blank=False)
is_active = models.BooleanField(default=True)
def __str__(self):
return f"{self.user_id} {self.type} (Web:{self.web}) (Push:{self.push})"\
f"(Email:{self.email}, {self.email_cadence})"
@classmethod
def create_default_preferences_for_user(cls, user_id) -> list:
"""
Creates all preferences for user
Note: It creates preferences using bulk create, so primary key will be missing for newly created
preference. Refetch if primary key is needed
"""
preferences = list(NotificationPreference.objects.filter(user_id=user_id))
user_preferences_map = {pref.type: pref for pref in preferences}
diff = set(COURSE_NOTIFICATION_TYPES.keys()) - set(user_preferences_map.keys())
if diff:
missing_types = [
create_notification_preference(user_id=user_id, notification_type=missing_type)
for missing_type in diff
]
new_preferences = NotificationPreference.objects.bulk_create(missing_types)
preferences = preferences + list(new_preferences)
return preferences
def is_enabled_for_any_channel(self, *args, **kwargs) -> bool:
"""
Returns True if the notification preference is enabled for any channel.

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

@@ -25,17 +25,15 @@ 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.utils import encrypt_object, encrypt_string
from openedx.core.djangoapps.notifications.email.utils import encrypt_string
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
Notification,
get_course_notification_preference_config_version, NotificationPreference
CourseNotificationPreference, Notification, NotificationPreference
)
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, NotificationTypeManager
from ..base_notification import COURSE_NOTIFICATION_APPS, NotificationTypeManager, COURSE_NOTIFICATION_TYPES
from ..utils import get_notification_types_with_visibility_settings, exclude_inaccessible_preferences
User = get_user_model()
@@ -496,39 +494,33 @@ class UpdatePreferenceFromEncryptedDataView(ModuleStoreTestCase):
"""
Tests if preference is updated when url is hit
"""
prefs = NotificationPreference.create_default_preferences_for_user(self.user.id)
assert any(pref.email for pref in prefs)
user_hash = encrypt_string(self.user.username)
patch_hash = encrypt_object({'channel': 'email', 'value': False})
url_params = {
"username": user_hash,
"patch": patch_hash
}
url = reverse("preference_update_from_encrypted_username_view", kwargs=url_params)
url = reverse("preference_update_view", kwargs=url_params)
func = getattr(self.client, request_type)
response = func(url)
assert response.status_code == status.HTTP_200_OK
preference = CourseNotificationPreference.objects.get(user=self.user, course_id=self.course.id)
config = preference.notification_preference_config
for app_name, app_prefs in config.items():
for type_prefs in app_prefs['notification_types'].values():
assert type_prefs['email'] is False
preferences = NotificationPreference.objects.filter(user=self.user)
for preference in preferences:
assert preference.email is False
def test_if_config_version_is_updated(self):
def test_creation_of_missing_preference(self):
"""
Tests if preference version is updated before applying patch data
Tests if missing preferences are created when unsubscribe is clicked
"""
preference = CourseNotificationPreference.objects.get(user=self.user, course_id=self.course.id)
preference.config_version -= 1
preference.save()
NotificationPreference.objects.filter(user=self.user).delete()
user_hash = encrypt_string(self.user.username)
patch_hash = encrypt_object({'channel': 'email', 'value': False})
url_params = {
"username": user_hash,
"patch": patch_hash
}
url = reverse("preference_update_from_encrypted_username_view", kwargs=url_params)
url = reverse("preference_update_view", kwargs=url_params)
self.client.get(url)
preference = CourseNotificationPreference.objects.get(user=self.user, course_id=self.course.id)
assert preference.config_version == get_course_notification_preference_config_version()
preferences = NotificationPreference.objects.filter(user=self.user)
assert preferences.count() == len(COURSE_NOTIFICATION_TYPES.keys())
def remove_notifications_with_visibility_settings(expected_response):

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

@@ -29,6 +29,8 @@ urlpatterns = [
name='mark-notifications-seen'
),
path('read/', NotificationReadAPIView.as_view(), name='notifications-read'),
path('preferences/update/<str:username>/', preference_update_from_encrypted_username_view,
name='preference_update_view'),
path('preferences/update/<str:username>/<str:patch>/', preference_update_from_encrypted_username_view,
name='preference_update_from_encrypted_username_view'),
]

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

@@ -236,12 +236,12 @@ class NotificationReadAPIView(APIView):
@api_view(['GET', 'POST'])
def preference_update_from_encrypted_username_view(request, username, patch):
def preference_update_from_encrypted_username_view(request, username, patch=""):
"""
View to update user preferences from encrypted username and patch.
username and patch must be string
"""
update_user_preferences_from_patch(username, patch)
update_user_preferences_from_patch(username)
return Response({"result": "success"}, status=status.HTTP_200_OK)