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

feat: added unsubsribe url for email notifications (#34967)

Browse Source
This commit is contained in:
Muhammad Adeel Tajamul
2024-06-28 11:22:09 +05:00
committed by GitHub
parent 6945bfae61
commit 98dfb12943
7 changed files with 426 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -92,8 +92,8 @@ def send_digest_email_to_user(user, cadence_type, course_language='en', courses_
logger.info(f'<Email Cadence> No filtered notification for {user.username} ==Temp Log==')
return
apps_dict = create_app_notifications_dict(notifications)
message_context = create_email_digest_context(apps_dict, start_date, end_date, cadence_type,
courses_data=courses_data)
message_context = create_email_digest_context(apps_dict, user.username, start_date, end_date,
cadence_type, courses_data=courses_data)
recipient = Recipient(user.id, user.email)
message = EmailNotificationMessageType(
app_label="notifications", name="email_digest"

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

@@ -4,21 +4,32 @@ Test utils.py
import datetime
import ddt
from itertools import product
from pytz import utc
from waffle import get_waffle_flag_model # pylint: disable=invalid-django-waffle-import
from common.djangoapps.student.tests.factories import UserFactory
from openedx.core.djangoapps.notifications.base_notification import (
COURSE_NOTIFICATION_APPS,
COURSE_NOTIFICATION_TYPES,
)
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_EMAIL_NOTIFICATIONS
from openedx.core.djangoapps.notifications.models import Notification
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, Notification
from openedx.core.djangoapps.notifications.email.utils import (
add_additional_attributes_to_notifications,
create_app_notifications_dict,
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,
is_email_notification_flag_enabled,
update_user_preferences_from_patch,
)
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
@@ -102,8 +113,9 @@ class TestContextFunctions(ModuleStoreTestCase):
"""
Tests common header and footer context
"""
context = create_email_template_context()
keys = ['platform_name', 'mailing_address', 'logo_url', 'social_media', 'notification_settings_url']
context = create_email_template_context(self.user.username)
keys = ['platform_name', 'mailing_address', 'logo_url', 'social_media',
'notification_settings_url', 'unsubscribe_url']
for key in keys:
assert key in context
@@ -121,6 +133,7 @@ class TestContextFunctions(ModuleStoreTestCase):
end_date = datetime.datetime(2024, 3, 24, 12, 0)
params = {
"app_notifications_dict": app_dict,
"username": self.user.username,
"start_date": end_date - datetime.timedelta(days=0 if digest_frequency == "Daily" else 6),
"end_date": end_date,
"digest_frequency": digest_frequency,
@@ -194,3 +207,227 @@ class TestWaffleFlag(ModuleStoreTestCase):
assert is_email_notification_flag_enabled() is False
assert is_email_notification_flag_enabled(self.user_1) is False
assert is_email_notification_flag_enabled(self.user_2) is False
class TestEncryption(ModuleStoreTestCase):
"""
Tests all encryption methods
"""
def test_string_encryption(self):
"""
Tests if decrypted string is equal original string
"""
string = "edx"
encrypted = encrypt_string(string)
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):
"""
Tests if preferences are update according to patch data
"""
def setUp(self):
"""
Setup test cases
"""
super().setUp()
self.user = UserFactory()
self.course_1 = CourseFactory.create(display_name='test course 1', run="Testing_course_1")
self.course_2 = CourseFactory.create(display_name='test course 2', run="Testing_course_2")
self.preference_1 = CourseNotificationPreference(course_id=self.course_1.id, user=self.user)
self.preference_2 = CourseNotificationPreference(course_id=self.course_2.id, user=self.user)
self.preference_1.save()
self.preference_2.save()
self.default_json = self.preference_1.notification_preference_config
def is_channel_editable(self, app_name, notification_type, channel):
"""
Returns if channel is editable
"""
if notification_type == 'core':
return channel not in COURSE_NOTIFICATION_APPS[app_name]['non_editable']
return channel not in COURSE_NOTIFICATION_TYPES[notification_type]['non_editable']
def get_default_cadence_value(self, app_name, notification_type):
"""
Returns default email cadence value
"""
if notification_type == 'core':
return COURSE_NOTIFICATION_APPS[app_name]['core_email_cadence']
return COURSE_NOTIFICATION_TYPES[notification_type]['email_cadence']
@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 = EmailCadence.NEVER
if new_value:
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 = EmailCadence.NEVER
if new_value:
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 = EmailCadence.NEVER
if new_value:
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 self.assertNumQueries(1):
update_user_preferences_from_patch(enc_username, enc_patch)

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

@@ -2,14 +2,22 @@
Email Notifications Utils
"""
import datetime
import json
from django.conf import settings
from django.urls import reverse
from pytz import utc
from waffle import get_waffle_flag_model # pylint: disable=invalid-django-waffle-import
from lms.djangoapps.branding.api import get_logo_url_for_email
from lms.djangoapps.discussion.notification_prefs.views import UsernameCipher
from openedx.core.djangoapps.notifications.base_notification import (
COURSE_NOTIFICATION_APPS,
COURSE_NOTIFICATION_TYPES,
)
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_EMAIL_NOTIFICATIONS
from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference
from xmodule.modulestore.django import modulestore
from .notification_icons import NotificationTypeIcons
@@ -51,7 +59,21 @@ def get_icon_url_for_notification_type(notification_type):
return NotificationTypeIcons.get_icon_url_for_notification_type(notification_type)
def create_email_template_context():
def get_unsubscribe_link(username, patch):
"""
Returns unsubscribe url for username with patch preferences
"""
encrypted_username = encrypt_string(username)
encrypted_patch = encrypt_object(patch)
kwargs = {
'username': encrypted_username,
'patch': encrypted_patch
}
relative_url = reverse('preference_update_from_encrypted_username_view', kwargs=kwargs)
return f"{settings.LMS_BASE}{relative_url}"
def create_email_template_context(username):
"""
Creates email context for header and footer
"""
@@ -65,16 +87,21 @@ def create_email_template_context():
for social_platform in social_media_urls.keys()
if social_media_icons.get(social_platform)
}
patch = {
'channel': 'email',
'value': False
}
return {
"platform_name": settings.PLATFORM_NAME,
"mailing_address": settings.CONTACT_MAILING_ADDRESS,
"logo_url": get_logo_url_for_email(),
"social_media": social_media_info,
"notification_settings_url": f"{settings.ACCOUNT_MICROFRONTEND_URL}/notifications",
"unsubscribe_url": get_unsubscribe_link(username, patch)
}
def create_email_digest_context(app_notifications_dict, start_date, end_date=None, digest_frequency="Daily",
def create_email_digest_context(app_notifications_dict, username, start_date, end_date=None, digest_frequency="Daily",
courses_data=None):
"""
Creates email context based on content
@@ -84,7 +111,7 @@ def create_email_digest_context(app_notifications_dict, start_date, end_date=Non
digest_frequency: EmailCadence.DAILY or EmailCadence.WEEKLY
courses_data: Dictionary to cache course info (avoid additional db calls)
"""
context = create_email_template_context()
context = create_email_template_context(username)
start_date_str = create_datetime_string(start_date)
end_date_str = create_datetime_string(end_date if end_date else start_date)
email_digest_updates = [{
@@ -243,3 +270,99 @@ def filter_notification_with_email_enabled_preferences(notifications, preference
if notification.notification_type in enabled_course_prefs[notification.course_id]:
filtered_notifications.append(notification)
return filtered_notifications
def encrypt_string(string):
"""
Encrypts input string
"""
return UsernameCipher.encrypt(string)
def decrypt_string(string):
"""
Decrypts input 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):
"""
Decrypt username and patch and updates user preferences
Allowed parameters for decrypted patch
app_name: name of app
notification_type: notification type name
channel: channel name ('web', 'push', 'email')
value: True or False
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))
kwargs = {'user__username': username}
if 'course_id' in patch.keys():
kwargs['course_id'] = patch['course_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
def is_editable(app_name, notification_type, channel):
"""
Returns if notification type channel is editable
"""
if notification_type == 'core':
return channel not in COURSE_NOTIFICATION_APPS[app_name]['non_editable']
return channel not in COURSE_NOTIFICATION_TYPES[notification_type]['non_editable']
def get_default_cadence_value(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']
preferences = CourseNotificationPreference.objects.filter(**kwargs)
# pylint: disable=too-many-nested-blocks
for preference in preferences:
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_editable(app_name, noti_type, channel):
type_prefs[channel] = pref_value
if channel == 'email':
cadence_value = get_default_cadence_value(app_name, noti_type)\
if pref_value else EmailCadence.NEVER
type_prefs['email_cadence'] = cadence_value
preference.save()

3
openedx/core/djangoapps/notifications/templates/notifications/digest_footer.html
View File

@@ -37,6 +37,9 @@
<a href="{{notification_settings_url}}" rel="noopener noreferrer" target="_blank" style="color: black">
Notification Settings
</a>
<a href="{{unsubscribe_url}}" rel="noopener noreferrer" target="_blank" style="color: black; margin-left: 1rem">
Unsubscribe
</a>
</p>
<p>
&copy; {% now "Y" %} {{ platform_name }}. All Rights Reserved <br/>

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

@@ -7,6 +7,7 @@ from unittest import mock
import ddt
from django.conf import settings
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
@@ -26,8 +27,10 @@ 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_notifications import EmailCadence
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, Notification
from openedx.core.djangoapps.notifications.serializers import NotificationCourseEnrollmentSerializer
from openedx.core.djangoapps.notifications.email.utils import get_unsubscribe_link
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
@@ -1080,6 +1083,40 @@ class NotificationReadAPIViewTestCase(APITestCase):
self.assertEqual(response.data, {'error': 'Invalid app_name or notification_id.'})
@ddt.ddt
class UpdatePreferenceFromEncryptedDataView(ModuleStoreTestCase):
"""
Tests if preference is updated when encrypted url is hit
"""
def setUp(self):
"""
Setup test case
"""
super().setUp()
password = 'password'
self.user = UserFactory(password=password)
self.client.login(username=self.user.username, password=password)
self.course = CourseFactory.create(display_name='test course 1', run="Testing_course_1")
CourseNotificationPreference(course_id=self.course.id, user=self.user).save()
@override_settings(LMS_BASE="")
@ddt.data('get', 'post')
def test_if_preference_is_updated(self, request_type):
"""
Tests if preference is updated when url is hit
"""
url = get_unsubscribe_link(self.user.username, {'channel': 'email', 'value': False})
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
assert type_prefs['email_cadence'] == EmailCadence.NEVER
def remove_notifications_with_visibility_settings(expected_response):
"""
Remove notifications with visibility settings from the expected response.

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

@@ -11,7 +11,9 @@ from .views import (
NotificationCountView,
NotificationListAPIView,
NotificationReadAPIView,
UserNotificationPreferenceView, UserNotificationChannelPreferenceView,
UserNotificationChannelPreferenceView,
UserNotificationPreferenceView,
preference_update_from_encrypted_username_view,
)
router = routers.DefaultRouter()
@@ -37,7 +39,8 @@ urlpatterns = [
name='mark-notifications-seen'
),
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'),
]
urlpatterns += router.urls

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

@@ -5,16 +5,19 @@ from datetime import datetime, timedelta
from django.conf import settings
from django.db.models import Count
from django.http import HttpResponse
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
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.utils import update_user_preferences_from_patch
from openedx.core.djangoapps.notifications.models import (
CourseNotificationPreference,
get_course_notification_preference_config_version
@@ -479,3 +482,13 @@ class NotificationReadAPIView(APIView):
return Response({'message': _('Notifications marked read.')}, status=status.HTTP_200_OK)
return Response({'error': _('Invalid app_name or notification_id.')}, status=status.HTTP_400_BAD_REQUEST)
@api_view(['GET', 'POST'])
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)
return HttpResponse("<!DOCTYPE html><html><body>Success</body></html>", status=status.HTTP_200_OK)