Merge branch 'master' into list_forum_members-to-drf

lms/djangoapps/discussion/rest_api/discussions_notifications.py

@@ -3,7 +3,10 @@
 """
 import re
 
+from bs4 import BeautifulSoup
 from django.conf import settings
+from django.utils.text import Truncator
+
 from lms.djangoapps.discussion.django_comment_client.permissions import get_team
 from openedx_events.learning.data import UserNotificationData, CourseNotificationData
 from openedx_events.learning.signals import USER_NOTIFICATION_REQUESTED, COURSE_NOTIFICATION_REQUESTED
@@ -27,13 +30,24 @@ class DiscussionNotificationSender:
     Class to send notifications to users who are subscribed to the thread.
     """
 
-    def __init__(self, thread, course, creator, parent_id=None):
+    def __init__(self, thread, course, creator, parent_id=None, comment_id=None):
         self.thread = thread
         self.course = course
         self.creator = creator
         self.parent_id = parent_id
+        self.comment_id = comment_id
         self.parent_response = None
+        self.comment = None
         self._get_parent_response()
+        self._get_comment()
+
+    def _get_comment(self):
+        """
+        Get comment object
+        """
+        if not self.comment_id:
+            return
+        self.comment = Comment(id=self.comment_id).retrieve()
 
     def _send_notification(self, user_ids, notification_type, extra_context=None):
         """
@@ -99,7 +113,10 @@ def send_new_response_notification(self):
         there is a response to the main thread.
         """
         if not self.parent_id and self.creator.id != int(self.thread.user_id):
-            self._send_notification([self.thread.user_id], "new_response")
+            context = {
+                'email_content': clean_thread_html_body(self.comment.body),
+            }
+            self._send_notification([self.thread.user_id], "new_response", extra_context=context)
 
     def _response_and_thread_has_same_creator(self) -> bool:
         """
@@ -136,6 +153,7 @@ def send_new_comment_notification(self):
             context = {
                 "author_name": str(author_name),
                 "author_pronoun": str(author_pronoun),
+                "email_content": clean_thread_html_body(self.comment.body),
             }
             self._send_notification([self.thread.user_id], "new_comment", extra_context=context)
 
@@ -149,7 +167,14 @@ def send_new_comment_on_response_notification(self):
             self.creator.id != int(self.parent_response.user_id) and not
             self._response_and_thread_has_same_creator()
         ):
-            self._send_notification([self.parent_response.user_id], "new_comment_on_response")
+            context = {
+                "email_content": clean_thread_html_body(self.comment.body),
+            }
+            self._send_notification(
+                [self.parent_response.user_id],
+                "new_comment_on_response",
+                extra_context=context
+            )
 
     def _check_if_subscriber_is_not_thread_or_content_creator(self, subscriber_id) -> bool:
         """
@@ -190,7 +215,12 @@ def send_response_on_followed_post_notification(self):
         # Remove duplicate users from the list of users to send notification
         users = list(set(users))
         if not self.parent_id:
-            self._send_notification(users, "response_on_followed_post")
+            self._send_notification(
+                users,
+                "response_on_followed_post",
+                extra_context={
+                    "email_content": clean_thread_html_body(self.comment.body),
+                })
         else:
             author_name = f"{self.parent_response.username}'s"
             # use 'their' if comment author is also response author.
@@ -206,6 +236,7 @@ def send_response_on_followed_post_notification(self):
                 extra_context={
                     "author_name": str(author_name),
                     "author_pronoun": str(author_pronoun),
+                    "email_content": clean_thread_html_body(self.comment.body),
                 }
             )
 
@@ -289,7 +320,8 @@ def send_new_thread_created_notification(self):
             ]
         context = {
             'username': self.creator.username,
-            'post_title': self.thread.title
+            'post_title': self.thread.title,
+            "email_content": clean_thread_html_body(self.thread.body),
         }
         self._send_course_wide_notification(notification_type, audience_filters, context)
 
@@ -339,3 +371,26 @@ def is_discussion_cohorted(course_key_str):
 def remove_html_tags(text):
     clean = re.compile('<.*?>')
     return re.sub(clean, '', text)
+
+
+def clean_thread_html_body(html_body):
+    """
+    Get post body with tags removed and limited to 500 characters
+    """
+    html_body = BeautifulSoup(Truncator(html_body).chars(500, html=True), 'html.parser')
+
+    tags_to_remove = [
+        "a", "link",  # Link Tags
+        "img", "picture", "source",  # Image Tags
+        "video", "track",  # Video Tags
+        "audio",  # Audio Tags
+        "embed", "object", "iframe",  # Embedded Content
+        "script"
+    ]
+
+    # Remove the specified tags while keeping their content
+    for tag in tags_to_remove:
+        for match in html_body.find_all(tag):
+            match.unwrap()
+
+    return str(html_body)

lms/djangoapps/discussion/rest_api/tasks.py

@@ -33,7 +33,7 @@ def send_thread_created_notification(thread_id, course_key_str, user_id):
 
 @shared_task
 @set_code_owner_attribute
-def send_response_notifications(thread_id, course_key_str, user_id, parent_id=None):
+def send_response_notifications(thread_id, course_key_str, user_id, comment_id, parent_id=None):
     """
     Send notifications to users who are subscribed to the thread.
     """
@@ -43,7 +43,7 @@ def send_response_notifications(thread_id, course_key_str, user_id, parent_id=No
     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, parent_id)
+    notification_sender = DiscussionNotificationSender(thread, course, user, parent_id, comment_id)
     notification_sender.send_new_comment_notification()
     notification_sender.send_new_response_notification()
     notification_sender.send_new_comment_on_response_notification()

lms/djangoapps/discussion/rest_api/tests/test_discussions_notifications.py

@@ -1,13 +1,14 @@
 """
 Unit tests for the DiscussionNotificationSender class
 """
-
+import re
 import unittest
 from unittest.mock import MagicMock, patch
 
 import pytest
 
-from lms.djangoapps.discussion.rest_api.discussions_notifications import DiscussionNotificationSender
+from lms.djangoapps.discussion.rest_api.discussions_notifications import DiscussionNotificationSender, \
+    clean_thread_html_body
 
 
 @patch('lms.djangoapps.discussion.rest_api.discussions_notifications.DiscussionNotificationSender'
@@ -88,3 +89,82 @@ def test_send_reported_content_notification_for_thread(self, mock_send_notificat
         self.notification_sender.send_reported_content_notification()
 
         self._assert_send_notification_called_with(mock_send_notification, 'thread')
+
+
+class TestCleanThreadHtmlBody(unittest.TestCase):
+    """
+    Tests for the clean_thread_html_body function
+    """
+
+    def test_html_tags_removal(self):
+        """
+        Test that the clean_thread_html_body function removes unwanted HTML tags
+        """
+        html_body = """
+        <p>This is a <a href="#">link</a> to a page.</p>
+        <p>Here is an image: <img src="image.jpg" alt="image"></p>
+        <p>Embedded video: <iframe src="video.mp4"></iframe></p>
+        <p>Script test: <script>alert('hello');</script></p>
+        <p>Some other content that should remain.</p>
+        """
+        expected_output = ("<p>This is a link to a page.</p>"
+                           "<p>Here is an image: </p>"
+                           "<p>Embedded video: </p>"
+                           "<p>Script test: alert('hello');</p>"
+                           "<p>Some other content that should remain.</p>")
+
+        result = clean_thread_html_body(html_body)
+
+        def normalize_html(text):
+            """
+             Normalize the output by removing extra whitespace, newlines, and spaces between tags
+            """
+            text = re.sub(r'\s+', ' ', text).strip()  # Replace any sequence of whitespace with a single space
+            text = re.sub(r'>\s+<', '><', text)  # Remove spaces between HTML tags
+            return text
+
+        normalized_result = normalize_html(result)
+        normalized_expected_output = normalize_html(expected_output)
+
+        self.assertEqual(normalized_result, normalized_expected_output)
+
+    def test_truncate_html_body(self):
+        """
+        Test that the clean_thread_html_body function truncates the HTML body to 500 characters
+        """
+        html_body = """
+        <p>This is a long text that should be truncated to 500 characters.</p>
+        """ * 20  # Repeat to exceed 500 characters
+
+        result = clean_thread_html_body(html_body)
+        self.assertGreaterEqual(500, len(result))
+
+    def test_no_tags_to_remove(self):
+        """
+        Test that the clean_thread_html_body function does not remove any tags if there are no unwanted tags
+        """
+        html_body = "<p>This paragraph has no tags to remove.</p>"
+        expected_output = "<p>This paragraph has no tags to remove.</p>"
+
+        result = clean_thread_html_body(html_body)
+        self.assertEqual(result, expected_output)
+
+    def test_empty_html_body(self):
+        """
+        Test that the clean_thread_html_body function returns an empty string if the input is an empty string
+        """
+        html_body = ""
+        expected_output = ""
+
+        result = clean_thread_html_body(html_body)
+        self.assertEqual(result, expected_output)
+
+    def test_only_script_tag(self):
+        """
+        Test that the clean_thread_html_body function removes the script tag and its content
+        """
+        html_body = "<script>alert('Hello');</script>"
+        expected_output = "alert('Hello');"
+
+        result = clean_thread_html_body(html_body)
+        self.assertEqual(result.strip(), expected_output)