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

feat: added new_question_post and new_discussion_post notification (#33043)

Browse Source
This commit is contained in:
Muhammad Adeel Tajamul
2023-08-31 14:29:21 +05:00
committed by GitHub
parent 9db024c435
commit 4efd54a3fd
10 changed files with 424 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
lms/djangoapps/discussion/rest_api/api.py
View File

@@ -120,6 +120,7 @@ from .serializers import (
UserStatsSerializer,
get_context
)
from .tasks import send_thread_created_notification
from .utils import (
AttributeDict,
add_stats_for_users_with_no_discussion_content,
@@ -127,7 +128,9 @@ from .utils import (
discussion_open_for_user,
get_usernames_for_course,
get_usernames_from_search_string,
set_attribute, send_response_notifications, is_posting_allowed
is_posting_allowed,
send_response_notifications,
set_attribute,
)
@@ -1466,6 +1469,8 @@ def create_thread(request, thread_data):
track_thread_created_event(request, course, cc_thread, actions_form.cleaned_data["following"],
from_mfe_sidebar)
thread_id = cc_thread.attributes['id']
send_thread_created_notification.apply_async(args=[thread_id, str(course.id), request.user.id])
return api_thread

28
lms/djangoapps/discussion/rest_api/tasks.py Normal file
View File

@@ -0,0 +1,28 @@
"""
Contain celery tasks
"""
from celery import shared_task
from django.contrib.auth import get_user_model
from opaque_keys.edx.locator import CourseKey
from lms.djangoapps.courseware.courses import get_course_with_access
from openedx.core.djangoapps.django_comment_common.comment_client.thread import Thread
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from .utils import DiscussionNotificationSender
User = get_user_model()
@shared_task
def send_thread_created_notification(thread_id, course_key_str, user_id):
"""
Send notification when a new thread is created
"""
course_key = CourseKey.from_string(course_key_str)
if not ENABLE_NOTIFICATIONS.is_enabled(course_key):
return
thread = Thread(id=thread_id).retrieve()
user = User.objects.get(id=user_id)
course = get_course_with_access(user, 'load', course_key, check_if_enrolled=True)
notification_sender = DiscussionNotificationSender(thread, course, user)
notification_sender.send_new_thread_created_notification()

36
lms/djangoapps/discussion/rest_api/tests/test_api.py
View File

@@ -1849,7 +1849,10 @@ class CreateThreadTest(
}
@mock.patch("eventtracking.tracker.emit")
def test_basic(self, mock_emit):
@mock.patch(
'lms.djangoapps.discussion.rest_api.tasks.send_thread_created_notification.apply_async'
)
def test_basic(self, mock_notification_task, mock_emit):
cs_thread = make_minimal_cs_thread({
"id": "test_id",
"username": self.user.username,
@@ -1865,6 +1868,7 @@ class CreateThreadTest(
"read": True,
})
assert actual == expected
mock_notification_task.assert_called_once()
assert parsed_body(httpretty.last_request()) == {
'course_id': [str(self.course.id)],
'commentable_id': ['test_topic'],
@@ -1907,7 +1911,10 @@ class CreateThreadTest(
self.assertEqual(assertion.exception.detail, "Discussions are in blackout period.")
@mock.patch("eventtracking.tracker.emit")
def test_basic_in_blackout_period_with_user_access(self, mock_emit):
@mock.patch(
'lms.djangoapps.discussion.rest_api.tasks.send_thread_created_notification.apply_async'
)
def test_basic_in_blackout_period_with_user_access(self, mock_notification_task, mock_emit):
"""
Test case when course is in blackout period and user has special privileges.
"""
@@ -1947,6 +1954,7 @@ class CreateThreadTest(
],
})
assert actual == expected
mock_notification_task.assert_called_once()
self.assertEqual(
parsed_body(httpretty.last_request()),
{
@@ -2030,7 +2038,10 @@ class CreateThreadTest(
)
)
@ddt.unpack
def test_group_id(self, role_name, course_is_cohorted, topic_is_cohorted, data_group_state):
@mock.patch(
'lms.djangoapps.discussion.rest_api.tasks.send_thread_created_notification.apply_async'
)
def test_group_id(self, role_name, course_is_cohorted, topic_is_cohorted, data_group_state, mock_notification_task):
"""
Tests whether the user has permission to create a thread with certain
group_id values.
@@ -2068,6 +2079,7 @@ class CreateThreadTest(
try:
create_thread(self.request, data)
assert not expected_error
mock_notification_task.assert_called_once()
actual_post_data = parsed_body(httpretty.last_request())
if data_group_state == "group_is_set":
assert actual_post_data['group_id'] == [str(data['group_id'])]
@@ -2079,19 +2091,26 @@ class CreateThreadTest(
if not expected_error:
self.fail(f"Unexpected validation error: {ex}")
def test_following(self):
@mock.patch(
'lms.djangoapps.discussion.rest_api.tasks.send_thread_created_notification.apply_async'
)
def test_following(self, mock_notification_task):
self.register_post_thread_response({"id": "test_id", "username": self.user.username})
self.register_subscription_response(self.user)
data = self.minimal_data.copy()
data["following"] = "True"
result = create_thread(self.request, data)
assert result['following'] is True
mock_notification_task.assert_called_once()
cs_request = httpretty.last_request()
assert urlparse(cs_request.path).path == f"/api/v1/users/{self.user.id}/subscriptions" # lint-amnesty, pylint: disable=no-member
assert cs_request.method == 'POST'
assert parsed_body(cs_request) == {'source_type': ['thread'], 'source_id': ['test_id']}
def test_voted(self):
@mock.patch(
'lms.djangoapps.discussion.rest_api.tasks.send_thread_created_notification.apply_async'
)
def test_voted(self, mock_notification_task):
self.register_post_thread_response({"id": "test_id", "username": self.user.username})
self.register_thread_votes_response("test_id")
data = self.minimal_data.copy()
@@ -2099,12 +2118,16 @@ class CreateThreadTest(
with self.assert_signal_sent(api, 'thread_voted', sender=None, user=self.user, exclude_args=('post',)):
result = create_thread(self.request, data)
assert result['voted'] is True
mock_notification_task.assert_called_once()
cs_request = httpretty.last_request()
assert urlparse(cs_request.path).path == '/api/v1/threads/test_id/votes' # lint-amnesty, pylint: disable=no-member
assert cs_request.method == 'PUT'
assert parsed_body(cs_request) == {'user_id': [str(self.user.id)], 'value': ['up']}
def test_abuse_flagged(self):
@mock.patch(
'lms.djangoapps.discussion.rest_api.tasks.send_thread_created_notification.apply_async'
)
def test_abuse_flagged(self, mock_notification_task):
self.register_post_thread_response({"id": "test_id", "username": self.user.username})
self.register_thread_flag_response("test_id")
data = self.minimal_data.copy()
@@ -2112,6 +2135,7 @@ class CreateThreadTest(
result = create_thread(self.request, data)
assert result['abuse_flagged'] is True
cs_request = httpretty.last_request()
mock_notification_task.assert_called_once()
assert urlparse(cs_request.path).path == '/api/v1/threads/test_id/abuse_flag' # lint-amnesty, pylint: disable=no-member
assert cs_request.method == 'PUT'
assert parsed_body(cs_request) == {'user_id': [str(self.user.id)]}

232
lms/djangoapps/discussion/rest_api/tests/test_tasks.py Normal file
View File

@@ -0,0 +1,232 @@
"""
Test cases for tasks.py
"""
from unittest import mock
from edx_toggles.toggles.testutils import override_waffle_flag
import ddt
import httpretty
from common.djangoapps.student.models import CourseEnrollment
from common.djangoapps.student.tests.factories import StaffFactory, UserFactory
from lms.djangoapps.discussion.django_comment_client.tests.factories import RoleFactory
from lms.djangoapps.discussion.rest_api.tasks import send_thread_created_notification
from lms.djangoapps.discussion.rest_api.tests.utils import make_minimal_cs_thread
from openedx.core.djangoapps.course_groups.models import CohortMembership, CourseCohortsSettings
from openedx.core.djangoapps.course_groups.tests.helpers import CohortFactory
from openedx.core.djangoapps.django_comment_common.models import (
CourseDiscussionSettings,
FORUM_ROLE_COMMUNITY_TA,
FORUM_ROLE_GROUP_MODERATOR,
FORUM_ROLE_MODERATOR,
FORUM_ROLE_STUDENT,
)
from openedx.core.djangoapps.discussions.models import DiscussionTopicLink
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx_events.learning.signals import USER_NOTIFICATION_REQUESTED
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
from .test_views import DiscussionAPIViewTestMixin
@ddt.ddt
@httpretty.activate
@mock.patch.dict("django.conf.settings.FEATURES", {"ENABLE_DISCUSSION_SERVICE": True})
@override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
class TestNewThreadCreatedNotification(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
"""
Test cases related to new_discussion_post and new_question_post notification types
"""
def setUp(self):
"""
Setup test case
"""
super().setUp()
# Creating a course
self.course = CourseFactory.create()
# Creating relative discussion and cohort settings
CourseCohortsSettings.objects.create(course_id=str(self.course.id))
CourseDiscussionSettings.objects.create(course_id=str(self.course.id), _divided_discussions='[]')
self.first_cohort = self.second_cohort = None
# Duplicating roles
self.student_role = RoleFactory(name=FORUM_ROLE_STUDENT, course_id=self.course.id)
self.moderator_role = RoleFactory(name=FORUM_ROLE_MODERATOR, course_id=self.course.id)
self.ta_role = RoleFactory(name=FORUM_ROLE_COMMUNITY_TA, course_id=self.course.id)
self.group_community_ta_role = RoleFactory(name=FORUM_ROLE_GROUP_MODERATOR, course_id=self.course.id)
# Creating users for with roles
self.author = StaffFactory(course_key=self.course.id, username='Author')
self.staff = StaffFactory(course_key=self.course.id, username='Staff')
self.moderator = UserFactory(username='Moderator')
self.moderator_role.users.add(self.moderator)
self.ta = UserFactory(username='TA')
self.ta_role.users.add(self.ta)
self.group_ta_cohort_1 = UserFactory(username='Group TA 1')
self.group_ta_cohort_2 = UserFactory(username='Group TA 2')
self.group_community_ta_role.users.add(self.group_ta_cohort_1)
self.group_community_ta_role.users.add(self.group_ta_cohort_2)
self.learner_cohort_1 = UserFactory(username='Learner 1')
self.learner_cohort_2 = UserFactory(username='Learner 2')
self.student_role.users.add(self.learner_cohort_1)
self.student_role.users.add(self.learner_cohort_2)
# Creating a topic
self.topic_id = 'test_topic'
usage_key = self.course.id.make_usage_key('vertical', self.topic_id)
self.topic = DiscussionTopicLink(
context_key=self.course.id,
usage_key=usage_key,
title=f"Discussion on {self.topic_id}",
external_id=self.topic_id,
provider_id="openedx",
ordering=1,
enabled_in_context=True,
)
self.notification_to_all_users = [
self.learner_cohort_1, self.learner_cohort_2, self.staff,
self.moderator, self.ta, self.group_ta_cohort_1, self.group_ta_cohort_2
]
self.privileged_users = [
self.staff, self.moderator, self.ta
]
self.cohort_1_users = [self.learner_cohort_1, self.group_ta_cohort_1] + self.privileged_users
self.cohort_2_users = [self.learner_cohort_2, self.group_ta_cohort_2] + self.privileged_users
self.thread = self._create_thread()
def _configure_cohorts(self):
"""
Configure cohort for course and assign membership to users
"""
course_key_str = str(self.course.id)
cohort_settings = CourseCohortsSettings.objects.get(course_id=course_key_str)
cohort_settings.is_cohorted = True
cohort_settings.save()
discussion_settings = CourseDiscussionSettings.objects.get(course_id=course_key_str)
discussion_settings.always_divide_inline_discussions = True
discussion_settings.save()
self.first_cohort = CohortFactory(course_id=self.course.id, name="FirstCohort")
self.second_cohort = CohortFactory(course_id=self.course.id, name="SecondCohort")
CohortMembership.assign(cohort=self.first_cohort, user=self.learner_cohort_1)
CohortMembership.assign(cohort=self.first_cohort, user=self.group_ta_cohort_1)
CohortMembership.assign(cohort=self.second_cohort, user=self.learner_cohort_2)
CohortMembership.assign(cohort=self.second_cohort, user=self.group_ta_cohort_2)
def _assign_enrollments(self):
"""
Enrolls all the user in the course
"""
user_list = [self.author] + self.notification_to_all_users
for user in user_list:
CourseEnrollment.enroll(user, self.course.id)
def _create_thread(self, thread_type="discussion", group_id=None):
"""
Create a thread
"""
thread = make_minimal_cs_thread({
'id': 1,
'course_id': str(self.course.id),
"commentable_id": self.topic_id,
"username": self.author.username,
"user_id": str(self.author.id),
"thread_type": thread_type,
"group_id": group_id,
"title": "Test Title",
})
self.register_get_thread_response(thread)
return thread
def assert_users_id_list(self, user_ids_1, user_ids_2):
"""
Assert whether the user ids in two lists are same
"""
assert len(user_ids_1) == len(user_ids_2)
for user_id in user_ids_1:
assert user_id in user_ids_2
def test_basic(self):
"""
Left empty intentionally. This test case is inherited from DiscussionAPIViewTestMixin
"""
def test_not_authenticated(self):
"""
Left empty intentionally. This test case is inherited from DiscussionAPIViewTestMixin
"""
def test_no_notification_if_course_has_no_enrollments(self):
"""
Tests no notification is send if course has no enrollments
"""
handler = mock.Mock()
USER_NOTIFICATION_REQUESTED.connect(handler)
send_thread_created_notification(self.thread['id'], str(self.course.id), self.author.id)
self.assertEqual(handler.call_count, 0)
@ddt.data(
('new_question_post',),
('new_discussion_post',),
)
@ddt.unpack
def test_notification_is_send_to_all_enrollments(self, notification_type):
"""
Tests notification is send to all users if course is not cohorted
"""
self._assign_enrollments()
thread_type = (
"discussion"
if notification_type == "new_discussion_post"
else ("question" if notification_type == "new_question_post" else "")
)
thread = self._create_thread(thread_type=thread_type)
handler = mock.Mock()
USER_NOTIFICATION_REQUESTED.connect(handler)
send_thread_created_notification(thread['id'], str(self.course.id), self.author.id)
self.assertEqual(handler.call_count, 1)
assert notification_type == handler.call_args[1]['notification_data'].notification_type
user_ids_list = [user.id for user in self.notification_to_all_users]
self.assert_users_id_list(user_ids_list, handler.call_args[1]['notification_data'].user_ids)
@ddt.data(
('cohort_1', 'new_question_post'),
('cohort_1', 'new_discussion_post'),
('cohort_2', 'new_question_post'),
('cohort_2', 'new_discussion_post'),
)
@ddt.unpack
def test_notification_is_send_to_cohort_ids(self, cohort_text, notification_type):
"""
Tests if notification is send only to privileged users and cohort members if the
course is cohorted
"""
self._assign_enrollments()
self._configure_cohorts()
cohort, audience = (
(self.first_cohort, self.cohort_1_users)
if cohort_text == "cohort_1"
else ((self.second_cohort, self.cohort_2_users) if cohort_text == "cohort_2" else None)
)
thread_type = (
"discussion"
if notification_type == "new_discussion_post"
else ("question" if notification_type == "new_question_post" else "")
)
cohort_id = cohort.id
thread = self._create_thread(group_id=cohort_id, thread_type=thread_type)
handler = mock.Mock()
USER_NOTIFICATION_REQUESTED.connect(handler)
send_thread_created_notification(thread['id'], str(self.course.id), self.author.id)
assert notification_type == handler.call_args[1]['notification_data'].notification_type
self.assertEqual(handler.call_count, 1)
user_ids_list = [user.id for user in audience]
self.assert_users_id_list(user_ids_list, handler.call_args[1]['notification_data'].user_ids)

6
lms/djangoapps/discussion/rest_api/tests/test_views.py
View File

@@ -1360,7 +1360,10 @@ class ThreadViewSetCreateTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
super().setUp()
self.url = reverse("thread-list")
def test_basic(self):
@mock.patch(
'lms.djangoapps.discussion.rest_api.tasks.send_thread_created_notification.apply_async'
)
def test_basic(self, mock_notification_task):
self.register_get_user_response(self.user)
cs_thread = make_minimal_cs_thread({
"id": "test_thread",
@@ -1381,6 +1384,7 @@ class ThreadViewSetCreateTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
content_type="application/json"
)
assert response.status_code == 200
mock_notification_task.assert_called_once()
response_data = json.loads(response.content.decode('utf-8'))
assert response_data == self.expected_thread_data({
"read": True,

86
lms/djangoapps/discussion/rest_api/utils.py
View File

@@ -10,11 +10,15 @@ from django.contrib.auth.models import User # lint-amnesty, pylint: disable=imp
from django.core.paginator import Paginator
from django.db.models.functions import Length
from common.djangoapps.student.models import CourseEnrollment
from common.djangoapps.student.roles import CourseStaffRole, CourseInstructorRole
from lms.djangoapps.discussion.django_comment_client.utils import has_discussion_privileges
from openedx.core.djangoapps.course_groups.models import CourseCohortsSettings, CourseUserGroup
from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration, PostingRestriction
from openedx.core.djangoapps.discussions.utils import get_divided_discussions
from openedx.core.djangoapps.django_comment_common.models import (
Role,
CourseDiscussionSettings,
FORUM_ROLE_ADMINISTRATOR,
FORUM_ROLE_MODERATOR,
FORUM_ROLE_GROUP_MODERATOR,
@@ -375,6 +379,15 @@ def send_response_notifications(thread, course, creator, parent_id=None):
notification_sender.send_new_comment_on_response_notification()
def is_discussion_cohorted(course_key_str):
"""
Returns if the discussion is divided by cohorts
"""
cohort_settings = CourseCohortsSettings.objects.get(course_id=course_key_str)
discussion_settings = CourseDiscussionSettings.objects.get(course_id=course_key_str)
return cohort_settings.is_cohorted and discussion_settings.always_divide_inline_discussions
class DiscussionNotificationSender:
"""
Class to send notifications to users who are subscribed to the thread.
@@ -422,6 +435,59 @@ class DiscussionNotificationSender:
return self.parent_response
def _create_cohort_course_audience(self):
"""
Creates audience based on user cohort and role
"""
course_key_str = str(self.course.id)
discussion_cohorted = is_discussion_cohorted(course_key_str)
# Retrieves cohort divided discussion
discussion_settings = CourseDiscussionSettings.objects.get(course_id=course_key_str)
divided_course_wide_discussions, divided_inline_discussions = get_divided_discussions(
self.course,
discussion_settings
)
# Checks if post has any cohort assigned
group_id = self.thread.attributes['group_id']
if group_id is not None:
group_id = int(group_id)
# Course wide topics
topic_id = self.thread.attributes['commentable_id']
all_topics = divided_inline_discussions + divided_course_wide_discussions
topic_divided = topic_id in all_topics or discussion_settings.always_divide_inline_discussions
user_ids = []
if discussion_cohorted and topic_divided and group_id is not None:
users_in_cohort = CourseUserGroup.objects.filter(
course_id=course_key_str, id=group_id
).values_list('users__id', flat=True)
user_ids.extend(users_in_cohort)
privileged_roles = [FORUM_ROLE_ADMINISTRATOR, FORUM_ROLE_MODERATOR, FORUM_ROLE_COMMUNITY_TA]
privileged_users = Role.objects.filter(
name__in=privileged_roles,
course_id=course_key_str
).values_list('users__id', flat=True)
user_ids.extend(privileged_users)
staff_users = CourseStaffRole(self.course.id).users_with_role().values_list('id', flat=True)
user_ids.extend(staff_users)
admin_users = CourseInstructorRole(self.course.id).users_with_role().values_list('id', flat=True)
user_ids.extend(admin_users)
else:
user_ids = CourseEnrollment.objects.filter(
course__id=course_key_str, is_active=True
).values_list('user__id', flat=True)
unique_user_ids = list(set(user_ids))
if self.creator.id in unique_user_ids:
unique_user_ids.remove(self.creator.id)
return unique_user_ids
def send_new_response_notification(self):
"""
Send notification to users who are subscribed to the main thread/post i.e.
@@ -464,6 +530,26 @@ class DiscussionNotificationSender:
):
self._send_notification([self.parent_response.user_id], "new_comment_on_response")
def send_new_thread_created_notification(self):
"""
Send notification based on notification_type
"""
thread_type = self.thread.attributes['thread_type']
notification_type = (
"new_question_post"
if thread_type == "question"
else ("new_discussion_post" if thread_type == "discussion" else "")
)
if notification_type not in ['new_discussion_post', 'new_question_post']:
raise ValueError(f'Invalid notification type {notification_type}')
user_ids = self._create_cohort_course_audience()
context = {
'username': self.creator.username,
'post_title': self.thread.title
}
self._send_notification(user_ids, notification_type, context)
def is_posting_allowed(posting_restrictions: str, blackout_schedules: List):
"""

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

@@ -13,7 +13,6 @@ COURSE_NOTIFICATION_TYPES = {
'notification_app': 'discussion',
'name': 'new_comment_on_response',
'is_core': True,
'info': 'Comment on response',
'content_template': _('<{p}><{strong}>{replier_name}</{strong}> commented on your response to the post '
'<{strong}>{post_title}</{strong}></{p}>'),
'content_context': {
@@ -26,8 +25,6 @@ COURSE_NOTIFICATION_TYPES = {
'notification_app': 'discussion',
'name': 'new_comment',
'is_core': True,
'info': 'Comment on post',
'non_editable': ['web', 'email'],
'content_template': _('<{p}><{strong}>{replier_name}</{strong}> commented on <{strong}>{author_name}\'s'
'</{strong}> response to your post <{strong}>{post_title}</{strong}></{p}>'),
'content_context': {
@@ -41,8 +38,6 @@ COURSE_NOTIFICATION_TYPES = {
'notification_app': 'discussion',
'name': 'new_response',
'is_core': True,
'info': 'Response on post',
'non_editable': [],
'content_template': _('<{p}><{strong}>{replier_name}</{strong}> responded to your '
'post <{strong}>{post_title}</{strong}></{p}>'),
'content_context': {
@@ -51,6 +46,38 @@ COURSE_NOTIFICATION_TYPES = {
},
'email_template': '',
},
'new_discussion_post': {
'notification_app': 'discussion',
'name': 'new_discussion_post',
'is_core': False,
'info': '',
'web': False,
'email': False,
'push': False,
'non_editable': [],
'content_template': _('<{p}><{strong}>{username}</{strong}> posted <{strong}>{post_title}</{strong}></{p}>'),
'content_context': {
'post_title': 'Post title',
'username': 'Post author name',
},
'email_template': '',
},
'new_question_post': {
'notification_app': 'discussion',
'name': 'new_question_post',
'is_core': False,
'info': '',
'web': False,
'email': False,
'push': False,
'non_editable': [],
'content_template': _('<{p}><{strong}>{username}</{strong}> asked <{strong}>{post_title}</{strong}></{p}>'),
'content_context': {
'post_title': 'Post title',
'username': 'Post author name',
},
'email_template': '',
}
}
COURSE_NOTIFICATION_APPS = {

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

@@ -21,7 +21,7 @@ log = logging.getLogger(__name__)
NOTIFICATION_CHANNELS = ['web', 'push', 'email']
# Update this version when there is a change to any course specific notification type or app.
COURSE_NOTIFICATION_CONFIG_VERSION = 2
COURSE_NOTIFICATION_CONFIG_VERSION = 3
def get_course_notification_preference_config():

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

@@ -276,7 +276,7 @@ class NotificationPreferenceValidationTest(ModuleStoreTestCase):
Tests if COURSE_NOTIFICATION_TYPES constant has all required keys with valid
data type for core notification type
"""
str_keys = ['notification_app', 'name', 'info', 'email_template']
str_keys = ['notification_app', 'name', 'email_template']
notification_types = base_notification.COURSE_NOTIFICATION_TYPES
assert "" not in notification_types.keys()
for notification_type in notification_types.values():

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

@@ -224,7 +224,9 @@ class UserNotificationPreferenceAPITest(ModuleStoreTestCase):
'email': True,
'push': True,
'info': ''
}
},
'new_discussion_post': {'web': False, 'email': False, 'push': False, 'info': ''},
'new_question_post': {'web': False, 'email': False, 'push': False, 'info': ''}
},
'non_editable': {
'core': ['web']