Merge branch 'master' into iamsobanjaved/django-42-lts

openedx/core/djangoapps/content_tagging/api.py

@@ -3,8 +3,6 @@ Content Tagging APIs
 """
 from __future__ import annotations
 
-from typing import Iterator
-
 import openedx_tagging.core.tagging.api as oel_tagging
 from django.db.models import Q, QuerySet, Exists, OuterRef
 from openedx_tagging.core.tagging.models import Taxonomy
@@ -101,7 +99,7 @@ def get_taxonomies_for_org(
     return oel_tagging.get_taxonomies(enabled=enabled).filter(
         Exists(
             TaxonomyOrg.get_relationships(
-                taxonomy=OuterRef("pk"),
+                taxonomy=OuterRef("pk"),  # type: ignore
                 rel_type=TaxonomyOrg.RelType.OWNER,
                 org_short_name=org_short_name,
             )
@@ -130,7 +128,7 @@ def get_unassigned_taxonomies(enabled=True) -> QuerySet:
 def get_content_tags(
     object_key: ContentKey,
     taxonomy_id: int | None = None,
-) -> Iterator[ContentObjectTag]:
+) -> QuerySet:
     """
     Generates a list of content tags for a given object.
 
@@ -147,7 +145,7 @@ def tag_content_object(
     object_key: ContentKey,
     taxonomy: Taxonomy,
     tags: list,
-) -> Iterator[ContentObjectTag]:
+) -> QuerySet:
     """
     This is the main API to use when you want to add/update/delete tags from a content object (e.g. an XBlock or
     course).

openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 
 from urllib.parse import parse_qs, urlparse
 import json
+from unittest.mock import MagicMock
 
 import abc
 import ddt
@@ -33,6 +34,7 @@ from openedx.core.djangoapps.content_libraries.api import (
     create_library,
     set_library_user_permissions,
 )
+from openedx.core.djangoapps.content_tagging import api as tagging_api
 from openedx.core.djangoapps.content_tagging.models import TaxonomyOrg
 from openedx.core.djangolib.testing.utils import skip_unless_cms
 from openedx.core.lib import blockstore_api
@@ -192,7 +194,7 @@ class TestTaxonomyObjectsMixin:
             rel_type=TaxonomyOrg.RelType.OWNER,
         )
 
-        # Global taxonomy
+        # Global taxonomy, which contains tags
         self.t1 = Taxonomy.objects.create(name="t1", enabled=True)
         TaxonomyOrg.objects.create(
             taxonomy=self.t1,
@@ -203,6 +205,12 @@ class TestTaxonomyObjectsMixin:
             taxonomy=self.t2,
             rel_type=TaxonomyOrg.RelType.OWNER,
         )
+        root1 = Tag.objects.create(taxonomy=self.t1, value="ALPHABET")
+        Tag.objects.create(taxonomy=self.t1, value="android", parent=root1)
+        Tag.objects.create(taxonomy=self.t1, value="abacus", parent=root1)
+        Tag.objects.create(taxonomy=self.t1, value="azure", parent=root1)
+        Tag.objects.create(taxonomy=self.t1, value="aardvark", parent=root1)
+        Tag.objects.create(taxonomy=self.t1, value="anvil", parent=root1)
 
         # OrgA taxonomy
         self.tA1 = Taxonomy.objects.create(name="tA1", enabled=True)
@@ -278,7 +286,8 @@ class TestTaxonomyListCreateViewSet(TestTaxonomyObjectsMixin, APITestCase):
             expected_taxonomies: list[str],
             enabled_parameter: bool | None = None,
             org_parameter: str | None = None,
-            unassigned_parameter: bool | None = None
+            unassigned_parameter: bool | None = None,
+            page_size: int | None = None,
     ) -> None:
         """
         Helper function to call the list endpoint and check the response
@@ -293,6 +302,7 @@ class TestTaxonomyListCreateViewSet(TestTaxonomyObjectsMixin, APITestCase):
             "enabled": enabled_parameter,
             "org": org_parameter,
             "unassigned": unassigned_parameter,
+            "page_size": page_size,
         }.items() if v is not None}
 
         response = self.client.get(url, query_params, format="json")
@@ -304,11 +314,12 @@ class TestTaxonomyListCreateViewSet(TestTaxonomyObjectsMixin, APITestCase):
         """
         Tests that staff users see all taxonomies
         """
-        # Default page_size=10, and so "tBA1" and "tBA2" appear on the second page
+        # page_size=10, and so "tBA1" and "tBA2" appear on the second page
         expected_taxonomies = ["ot1", "ot2", "st1", "st2", "t1", "t2", "tA1", "tA2", "tB1", "tB2"]
         self._test_list_taxonomy(
             user_attr="staff",
             expected_taxonomies=expected_taxonomies,
+            page_size=10,
         )
 
     @ddt.data(
@@ -476,6 +487,29 @@ class TestTaxonomyListCreateViewSet(TestTaxonomyObjectsMixin, APITestCase):
             if user_attr == "staffA":
                 assert response.data["orgs"] == [self.orgA.short_name]
 
+    def test_list_taxonomy_query_count(self):
+        """
+        Test how many queries are used when retrieving taxonomies and permissions
+        """
+        url = TAXONOMY_ORG_LIST_URL + f'?org=${self.orgA.short_name}&enabled=true'
+
+        self.client.force_authenticate(user=self.staff)
+        with self.assertNumQueries(16):  # TODO Why so many queries?
+            response = self.client.get(url)
+
+        assert response.status_code == 200
+        assert response.data["can_add_taxonomy"]
+        assert len(response.data["results"]) == 2
+        for taxonomy in response.data["results"]:
+            if taxonomy["system_defined"]:
+                assert not taxonomy["can_change_taxonomy"]
+                assert not taxonomy["can_delete_taxonomy"]
+                assert taxonomy["can_tag_object"]
+            else:
+                assert taxonomy["can_change_taxonomy"]
+                assert taxonomy["can_delete_taxonomy"]
+                assert taxonomy["can_tag_object"]
+
 
 @ddt.ddt
 class TestTaxonomyDetailExportMixin(TestTaxonomyObjectsMixin):
@@ -787,7 +821,14 @@ class TestTaxonomyDetailViewSet(TestTaxonomyDetailExportMixin, APITestCase):
         assert response.status_code == expected_status, reason
 
         if status.is_success(expected_status):
-            check_taxonomy(response.data, taxonomy.pk, **(TaxonomySerializer(taxonomy.cast()).data))
+            request = MagicMock()
+            request.user = user
+            context = {"request": request}
+            check_taxonomy(
+                response.data,
+                taxonomy.pk,
+                **(TaxonomySerializer(taxonomy.cast(), context=context)).data,
+            )
 
 
 @skip_unless_cms
@@ -1538,12 +1579,12 @@ class TestObjectTagViewSet(TestObjectTagMixin, APITestCase):
 
         # Fetch this object's tags for a single taxonomy
         expected_tags = [{
-            'editable': True,
             'name': 'Multiple Taxonomy',
             'taxonomy_id': taxonomy.pk,
+            'can_tag_object': True,
             'tags': [
-                {'value': 'Tag 1', 'lineage': ['Tag 1']},
-                {'value': 'Tag 2', 'lineage': ['Tag 2']},
+                {'value': 'Tag 1', 'lineage': ['Tag 1'], 'can_delete_objecttag': True},
+                {'value': 'Tag 2', 'lineage': ['Tag 2'], 'can_delete_objecttag': True},
             ],
         }]
 
@@ -1560,6 +1601,28 @@ class TestObjectTagViewSet(TestObjectTagMixin, APITestCase):
         assert status.is_success(response3.status_code)
         assert response3.data[str(self.courseA)]["taxonomies"] == expected_tags
 
+    def test_object_tags_query_count(self):
+        """
+        Test how many queries are used when retrieving object tags and permissions
+        """
+        object_key = self.courseA
+        object_id = str(object_key)
+        tagging_api.tag_content_object(object_key=object_key, taxonomy=self.t1, tags=["anvil", "android"])
+        expected_tags = [
+            {"value": "android", "lineage": ["ALPHABET", "android"], "can_delete_objecttag": True},
+            {"value": "anvil", "lineage": ["ALPHABET", "anvil"], "can_delete_objecttag": True},
+        ]
+
+        url = OBJECT_TAGS_URL.format(object_id=object_id)
+        self.client.force_authenticate(user=self.staff)
+        with self.assertNumQueries(7):  # TODO Why so many queries?
+            response = self.client.get(url)
+
+        assert response.status_code == 200
+        assert len(response.data[object_id]["taxonomies"]) == 1
+        assert response.data[object_id]["taxonomies"][0]["can_tag_object"]
+        assert response.data[object_id]["taxonomies"][0]["tags"] == expected_tags
+
 
 @skip_unless_cms
 @ddt.ddt
@@ -2029,3 +2092,27 @@ class TestImportTagsView(ImportTaxonomyMixin, APITestCase):
         assert len(tags) == len(self.old_tags)
         for i, tag in enumerate(tags):
             assert tag["value"] == self.old_tags[i].value
+
+
+@skip_unless_cms
+@ddt.ddt
+class TestTaxonomyTagsViewSet(TestTaxonomyObjectsMixin, APITestCase):
+    """
+    Test cases for TaxonomyTagsViewSet retrive action.
+    """
+    def test_taxonomy_tags_query_count(self):
+        """
+        Test how many queries are used when retrieving small taxonomies+tags and permissions
+        """
+        url = f"{TAXONOMY_TAGS_URL}?search_term=an&parent_tag=ALPHABET".format(pk=self.t1.id)
+
+        self.client.force_authenticate(user=self.staff)
+        with self.assertNumQueries(13):  # TODO Why so many queries?
+            response = self.client.get(url)
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.data["can_add_tag"]
+        assert len(response.data["results"]) == 2
+        for taxonomy in response.data["results"]:
+            assert taxonomy["can_change_tag"]
+            assert taxonomy["can_delete_tag"]

openedx/core/djangoapps/content_tagging/rest_api/v1/urls.py

@@ -3,6 +3,7 @@ Taxonomies API v1 URLs.
 """
 
 from rest_framework.routers import DefaultRouter
+from openedx_tagging.core.tagging.rest_api.v1.views import ObjectTagCountsView
 
 from django.urls.conf import path, include
 
@@ -16,6 +17,7 @@ from . import views
 router = DefaultRouter()
 router.register("taxonomies", views.TaxonomyOrgView, basename="taxonomy")
 router.register("object_tags", views.ObjectTagOrgView, basename="object_tag")
+router.register("object_tag_counts", ObjectTagCountsView, basename="object_tag_counts")
 
 urlpatterns = [
     path(

openedx/core/djangoapps/content_tagging/rest_api/v1/views.py

@@ -81,11 +81,11 @@ class TaxonomyOrgView(TaxonomyView):
         serializer.instance = create_taxonomy(**serializer.validated_data, orgs=user_admin_orgs)
 
     @action(detail=False, url_path="import", methods=["post"])
-    def create_import(self, request: Request, **kwargs) -> Response:
+    def create_import(self, request: Request, **kwargs) -> Response:  # type: ignore
         """
         Creates a new taxonomy with the given orgs and imports the tags from the uploaded file.
         """
-        response = super().create_import(request, **kwargs)
+        response = super().create_import(request=request, **kwargs)  # type: ignore
 
         # If creation was successful, set the orgs for the new taxonomy
         if status.is_success(response.status_code):

openedx/core/djangoapps/content_tagging/rules.py

@@ -219,7 +219,7 @@ def can_change_object_tag_objectid(user: UserType, object_id: str) -> bool:
     Everyone that has permission to edit the object should be able to tag it.
     """
     if not object_id:
-        raise ValueError("object_id must be provided")
+        return True
     try:
         usage_key = UsageKey.from_string(object_id)
         if not usage_key.course_key.is_course:
@@ -274,7 +274,7 @@ def can_change_taxonomy_tag(user: UserType, tag: oel_tagging.Tag | None = None)
     return oel_tagging.is_taxonomy_admin(user) and (
         not tag
         or not taxonomy
-        or (taxonomy and not taxonomy.allow_free_text and not taxonomy.system_defined)
+        or (bool(taxonomy) and not taxonomy.allow_free_text and not taxonomy.system_defined)
     )
 
 

openedx/core/djangoapps/notifications/audience_filters.py

@@ -1,7 +1,6 @@
 """
 Audience based filters for notifications
 """
-import logging
 
 from abc import abstractmethod
 
@@ -22,9 +21,6 @@ from openedx.core.djangoapps.django_comment_common.models import (
 )
 
 
-log = logging.getLogger(__name__)
-
-
 class NotificationAudienceFilterBase:
     """
     Base class for notification audience filters
@@ -84,12 +80,10 @@ class CourseRoleAudienceFilter(NotificationAudienceFilterBase):
 
         if 'staff' in course_roles:
             staff_users = CourseStaffRole(course_key).users_with_role().values_list('id', flat=True)
-            log.info(f'Temp: Course wide notification, staff users calculated are {staff_users}')
             user_ids.extend(staff_users)
 
         if 'instructor' in course_roles:
             instructor_users = CourseInstructorRole(course_key).users_with_role().values_list('id', flat=True)
-            log.info(f'Temp: Course wide notification, instructor users calculated are {instructor_users}')
             user_ids.extend(instructor_users)
 
         return user_ids

openedx/core/djangoapps/notifications/base_notification.py

@@ -113,6 +113,25 @@ COURSE_NOTIFICATION_TYPES = {
         'email_template': '',
         'filters': [FILTER_AUDIT_EXPIRED_USERS_WITH_NO_ROLE]
     },
+    'content_reported': {
+        'notification_app': 'discussion',
+        'name': 'content_reported',
+        'is_core': False,
+        'info': '',
+        'web': True,
+        'email': True,
+        'push': True,
+        'non_editable': [],
+        'content_template': _('<p><strong>{username}’s </strong> {content_type} has been reported <strong> {'
+                              'content}</strong></p>'),
+
+        'content_context': {
+            'post_title': 'Post title',
+            'author_name': 'author name',
+            'replier_name': 'replier name',
+        },
+        'email_template': '',
+    },
 }
 
 COURSE_NOTIFICATION_APPS = {

openedx/core/djangoapps/notifications/handlers.py

@@ -96,13 +96,10 @@ def calculate_course_wide_notification_audience(course_key, audience_filters):
             if filter_class:
                 filter_instance = filter_class(course_key)
                 filtered_users = filter_instance.filter(filter_values)
-                log.info(f'Temp: Course-wide notification filtered users are '
-                         f'{filtered_users} for filter type {filter_type}')
                 audience_user_ids.extend(filtered_users)
         else:
             raise ValueError(f"Invalid audience filter type: {filter_type}")
 
-    log.info(f'Temp: Course-wide notification after audience filter is applied, users: {list(set(audience_user_ids))}')
     return list(set(audience_user_ids))
 
 
@@ -131,5 +128,4 @@ def generate_course_notifications(signal, sender, course_notification_data, meta
         'content_url': course_notification_data.get('content_url'),
     }
 
-    log.info(f"Temp: Course-wide notification, user_ids to sent notifications to {notification_data.get('user_ids')}")
     send_notifications.delay(**notification_data)

openedx/core/djangoapps/notifications/models.py

@@ -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 = 4
+COURSE_NOTIFICATION_CONFIG_VERSION = 5
 
 
 def get_course_notification_preference_config():

openedx/core/djangoapps/notifications/tests/test_views.py

@@ -18,6 +18,7 @@ from rest_framework.test import APIClient, APITestCase
 from common.djangoapps.student.models import CourseEnrollment
 from common.djangoapps.student.tests.factories import UserFactory
 from lms.djangoapps.discussion.django_comment_client.tests.factories import RoleFactory
+from lms.djangoapps.discussion.toggles import ENABLE_REPORTED_CONTENT_NOTIFICATIONS
 from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
 from openedx.core.djangoapps.django_comment_common.models import (
     FORUM_ROLE_ADMINISTRATOR,
@@ -169,6 +170,7 @@ class CourseEnrollmentPostSaveTest(ModuleStoreTestCase):
 
 
 @override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
+@override_waffle_flag(ENABLE_REPORTED_CONTENT_NOTIFICATIONS, active=True)
 @ddt.ddt
 class UserNotificationPreferenceAPITest(ModuleStoreTestCase):
     """
@@ -246,6 +248,7 @@ class UserNotificationPreferenceAPITest(ModuleStoreTestCase):
                         },
                         'new_discussion_post': {'web': False, 'email': False, 'push': False, 'info': ''},
                         'new_question_post': {'web': False, 'email': False, 'push': False, 'info': ''},
+                        'content_reported': {'web': True, 'email': True, 'push': True, 'info': ''},
                     },
                     'non_editable': {
                         'core': ['web']

openedx/core/djangoapps/notifications/utils.py

@@ -4,6 +4,7 @@ Utils function for notifications app
 from typing import Dict, List
 
 from common.djangoapps.student.models import CourseEnrollment
+from lms.djangoapps.discussion.toggles import ENABLE_REPORTED_CONTENT_NOTIFICATIONS
 from openedx.core.djangoapps.django_comment_common.models import Role
 from openedx.core.lib.cache_utils import request_cached
 
@@ -65,6 +66,10 @@ def filter_course_wide_preferences(course_key, preferences):
     if ENABLE_COURSEWIDE_NOTIFICATIONS.is_enabled(course_key):
         return preferences
     course_wide_notification_types = ['new_discussion_post', 'new_question_post']
+
+    if not ENABLE_REPORTED_CONTENT_NOTIFICATIONS.is_enabled(course_key):
+        course_wide_notification_types.append('content_reported')
+
     config = preferences['notification_preference_config']
     for app_prefs in config.values():
         notification_types = app_prefs['notification_types']

openedx/core/djangoapps/safe_sessions/middleware.py

@@ -95,7 +95,7 @@ from edx_django_utils.logging import encrypt_for_log
 from edx_django_utils.monitoring import set_custom_attribute
 from edx_toggles.toggles import SettingToggle
 
-from openedx.core.djangoapps.user_authn.cookies import delete_logged_in_cookies
+from openedx.core.djangoapps.user_authn.cookies import delete_logged_in_cookies, set_logged_in_cookies
 from openedx.core.lib.mobile_utils import is_request_from_mobile_app
 
 # .. toggle_name: LOG_REQUEST_USER_CHANGES
@@ -768,6 +768,92 @@ class SafeSessionMiddleware(SessionMiddleware, MiddlewareMixin):
         return encrypt_for_log(str(request.headers), getattr(settings, 'SAFE_SESSIONS_DEBUG_PUBLIC_KEY', None))
 
 
+class EmailChangeMiddleware(MiddlewareMixin):
+    """
+    Middleware responsible for performing the following
+    jobs on detecting an email change
+    1) It will update the session's email and update the JWT cookie
+        to match the new email.
+    2) It will invalidate any future session on other browsers where
+        the user's email does not match its session email.
+
+    This middleware ensures that the sessions in other browsers
+    are invalidated when a user changes their email in one browser.
+    The active session in which the email change is made will remain valid.
+
+    The user's email is stored in their session and JWT cookies during login
+    and gets updated when the user changes their email.
+    This middleware checks for any mismatch between the stored email
+    and the current user's email in each request, and if found,
+    it invalidates/flushes the session and mark cookies for deletion in request
+    which are then deleted in the process_response of SafeSessionMiddleware.
+    """
+
+    def process_request(self, request):
+        """
+        Invalidate the user session if there's a mismatch
+        between the email in the user's session and request.user.email.
+        """
+        if request.user.is_authenticated:
+            user_session_email = request.session.get('email', None)
+            are_emails_mismatched = user_session_email is not None and request.user.email != user_session_email
+            EmailChangeMiddleware._set_session_email_match_custom_attributes(are_emails_mismatched)
+            if settings.ENFORCE_SESSION_EMAIL_MATCH and are_emails_mismatched:
+                # Flush the session and mark cookies for deletion.
+                log.info(
+                    f'EmailChangeMiddleware invalidating session for user: {request.user.id} due to email mismatch.'
+                )
+                request.session.flush()
+                request.user = AnonymousUser()
+                _mark_cookie_for_deletion(request)
+
+    def process_response(self, request, response):
+        """
+        1. Update the logged-in cookies if the email change was requested
+        2. Store user's email in session if not already
+        """
+        if request.user.is_authenticated:
+            if request.session.get('email', None) is None:
+                # .. custom_attribute_name: session_with_no_email_found
+                # .. custom_attribute_description: Indicates that user's email was not
+                #      yet stored in the user's session.
+                set_custom_attribute('session_with_no_email_found', True)
+                request.session['email'] = request.user.email
+
+            if request_cache.get_cached_response('email_change_requested').is_found:
+                # Update the JWT cookies with new user email
+                response = set_logged_in_cookies(request, response, request.user)
+
+        return response
+
+    @staticmethod
+    def register_email_change(request, email):
+        """
+        Stores the fact that an email change happened.
+
+        1. Sets the email in session for later comparison.
+        2. Sets a request level variable to mark that the user email change was requested.
+        """
+        request.session['email'] = email
+        request_cache.set('email_change_requested', True)
+
+    @staticmethod
+    def _set_session_email_match_custom_attributes(are_emails_mismatched):
+        """
+        Sets custom attributes of session_email_match
+        """
+        # .. custom_attribute_name: session_email_match
+        # .. custom_attribute_description: Indicates whether there is a match between the
+        #      email in the user's session and the current user's email in the request.
+        set_custom_attribute('session_email_mismatch', are_emails_mismatched)
+
+        # .. custom_attribute_name: is_enforce_session_email_match_enabled
+        # .. custom_attribute_description: Indicates whether session email match was enforced.
+        #       When enforced/enabled, it invalidates sessions in other browsers upon email change,
+        #       while preserving the session validity in the browser where the email change occurs.
+        set_custom_attribute('is_enforce_session_email_match_enabled', settings.ENFORCE_SESSION_EMAIL_MATCH)
+
+
 def obscure_token(value: Union[str, None]) -> Union[str, None]:
     """
     Return a short string that can be used to detect other occurrences

openedx/core/djangoapps/safe_sessions/tests/test_middleware.py

@@ -1,22 +1,29 @@
 """
 Unit tests for SafeSessionMiddleware
 """
-
+import uuid
 from unittest.mock import call, patch, MagicMock
 
 import ddt
 from crum import set_current_request
 from django.conf import settings
 from django.contrib.auth import SESSION_KEY
-from django.contrib.auth.models import AnonymousUser
+from django.contrib.auth.models import AnonymousUser, User  # lint-amnesty, pylint: disable=imported-auth-user
 from django.http import HttpResponse, HttpResponseRedirect, SimpleCookie
 from django.test import TestCase
 from django.test.utils import override_settings
+from django.urls import reverse
+from edx_django_utils.cache import RequestCache
+from edx_rest_framework_extensions.auth.jwt import cookies as jwt_cookies
 
-from openedx.core.djangolib.testing.utils import get_mock_request, CacheIsolationTestCase
+from common.djangoapps.student.models import PendingEmailChange
+from openedx.core.djangolib.testing.utils import get_mock_request, CacheIsolationTestCase, skip_unless_lms
+from openedx.core.djangoapps.user_authn.tests.utils import setup_login_oauth_client
+from openedx.core.djangoapps.user_authn.cookies import ALL_LOGGED_IN_COOKIE_NAMES
 from common.djangoapps.student.tests.factories import UserFactory
 
 from ..middleware import (
+    EmailChangeMiddleware,
     SafeCookieData,
     SafeSessionMiddleware,
     mark_user_change_as_expected,
@@ -615,3 +622,748 @@ class TestTrackRequestUserChanges(TestCase):
         request.user = object()
         assert len(request.debug_user_changes) == 2
         assert "Changing request user but user has no id." in request.debug_user_changes[1]
+
+
+@skip_unless_lms
+class TestEmailChangeMiddleware(TestSafeSessionsLogMixin, TestCase):
+    """
+    Test class for EmailChangeMiddleware
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.EMAIL = 'test@example.com'
+        self.PASSWORD = 'Password1234'
+        self.user = UserFactory.create(email=self.EMAIL, password=self.PASSWORD)
+        self.addCleanup(set_current_request, None)
+        self.request = get_mock_request(self.user)
+        self.request.session = {}
+        self.client.response = HttpResponse()
+        self.client.response.cookies = SimpleCookie()
+        self.addCleanup(RequestCache.clear_all_namespaces)
+
+        self.login_url = reverse("user_api_login_session", kwargs={'api_version': 'v2'})
+        self.register_url = reverse("user_api_registration_v2")
+        self.dashboard_url = reverse('dashboard')
+
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=False)
+    @patch('openedx.core.djangoapps.safe_sessions.middleware._mark_cookie_for_deletion')
+    def test_process_request_user_not_authenticated_with_toggle_disabled(self, mock_mark_cookie_for_deletion):
+        """
+        Calls EmailChangeMiddleware.process_request when no user is authenticated
+        and ENFORCE_SESSION_EMAIL_MATCH toggle is disabled.
+        Verifies that session and cookies are not affected.
+        """
+        # Unauthenticated User
+        self.request.user = AnonymousUser()
+
+        # Call process_request without authenticating a user
+        EmailChangeMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Assert that session and cookies are not affected
+        # Assert that _mark_cookie_for_deletion not called
+        mock_mark_cookie_for_deletion.assert_not_called()
+
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=True)
+    @patch('openedx.core.djangoapps.safe_sessions.middleware._mark_cookie_for_deletion')
+    def test_process_request_user_not_authenticated_with_toggle_enabled(self, mock_mark_cookie_for_deletion):
+        """
+        Calls EmailChangeMiddleware.process_request when no user is authenticated
+        and ENFORCE_SESSION_EMAIL_MATCH toggle is enabled.
+        Verifies that session and cookies are not affected.
+        """
+        # Unauthenticated User
+        self.request.user = AnonymousUser()
+
+        # Call process_request without authenticating a user
+        EmailChangeMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Assert that session and cookies are not affected
+        # Assert that _mark_cookie_for_deletion not called
+        mock_mark_cookie_for_deletion.assert_not_called()
+
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=True)
+    @patch('openedx.core.djangoapps.safe_sessions.middleware._mark_cookie_for_deletion')
+    @patch("openedx.core.djangoapps.safe_sessions.middleware.set_custom_attribute")
+    def test_process_request_emails_match_with_toggle_enabled(
+        self, mock_set_custom_attribute, mock_mark_cookie_for_deletion
+    ):
+        """
+        Calls EmailChangeMiddleware.process_request when user is authenticated,
+        ENFORCE_SESSION_EMAIL_MATCH is enabled and user session and request email also match.
+        Verifies that session and cookies are not affected.
+        """
+        # Log in the user
+        self.client.login(email=self.user.email, password=self.PASSWORD)
+        self.request.session = self.client.session
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # Registering email change (store user's email in session for later comparison by
+        # process_request function of middleware)
+        EmailChangeMiddleware.register_email_change(request=self.request, email=self.user.email)
+
+        # Ensure email is set in the session
+        self.assertEqual(self.request.session.get('email'), self.user.email)
+        # Ensure session cookie exist
+        self.assertEqual(len(self.client.response.cookies), 1)
+
+        # No email change occurred in any browser
+
+        # Call process_request
+        EmailChangeMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Verify that session_email_mismatch and is_enforce_session_email_match_enabled
+        # custom attributes are set
+        mock_set_custom_attribute.assert_has_calls([call('session_email_mismatch', False)])
+        mock_set_custom_attribute.assert_has_calls([call('is_enforce_session_email_match_enabled', True)])
+
+        # Assert that the session and cookies are not affected
+        self.assertEqual(self.request.session.get('email'), self.user.email)
+        self.assertEqual(len(self.client.response.cookies), 1)
+        self.assertEqual(self.client.response.cookies[settings.SESSION_COOKIE_NAME].value, 'authenticated')
+
+        # Assert that _mark_cookie_for_deletion not called
+        mock_mark_cookie_for_deletion.assert_not_called()
+
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=False)
+    @patch('openedx.core.djangoapps.safe_sessions.middleware._mark_cookie_for_deletion')
+    @patch("openedx.core.djangoapps.safe_sessions.middleware.set_custom_attribute")
+    def test_process_request_emails_match_with_toggle_disabled(
+        self, mock_set_custom_attribute, mock_mark_cookie_for_deletion
+    ):
+        """
+        Calls EmailChangeMiddleware.process_request when user is authenticated,
+        ENFORCE_SESSION_EMAIL_MATCH is disabled and user session and request email match.
+        Verifies that session and cookies are not affected.
+        """
+        # Log in the user
+        self.client.login(email=self.user.email, password=self.PASSWORD)
+        self.request.session = self.client.session
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # Registering email change (store user's email in session for later comparison by
+        # process_request function of middleware)
+        EmailChangeMiddleware.register_email_change(request=self.request, email=self.user.email)
+
+        # Ensure email is set in the session
+        self.assertEqual(self.request.session.get('email'), self.user.email)
+        # Ensure session cookie exist
+        self.assertEqual(len(self.client.response.cookies), 1)
+
+        # No email change occurred in any browser
+
+        # Call process_request
+        EmailChangeMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Verify that session_email_mismatch and is_enforce_session_email_match_enabled
+        # custom attributes are set
+        mock_set_custom_attribute.assert_has_calls([call('session_email_mismatch', False)])
+        mock_set_custom_attribute.assert_has_calls([call('is_enforce_session_email_match_enabled', False)])
+
+        # Assert that the session and cookies are not affected
+        self.assertEqual(self.request.session.get('email'), self.user.email)
+        self.assertEqual(len(self.client.response.cookies), 1)
+        self.assertEqual(self.client.response.cookies[settings.SESSION_COOKIE_NAME].value, 'authenticated')
+
+        # Assert that _mark_cookie_for_deletion not called
+        mock_mark_cookie_for_deletion.assert_not_called()
+
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=True)
+    @patch('openedx.core.djangoapps.safe_sessions.middleware._mark_cookie_for_deletion')
+    @patch("openedx.core.djangoapps.safe_sessions.middleware.set_custom_attribute")
+    def test_process_request_emails_mismatch_with_toggle_enabled(
+        self, mock_set_custom_attribute, mock_mark_cookie_for_deletion
+    ):
+        """
+        Calls EmailChangeMiddleware.process_request when user is authenticated,
+        ENFORCE_SESSION_EMAIL_MATCH is enabled and user session and request
+        email mismatch. (Email was changed in some other browser)
+        Verifies that session is flushed and cookies are marked for deletion.
+        """
+        # Log in the user
+        self.client.login(email=self.user.email, password=self.PASSWORD)
+        self.request.session = self.client.session
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # Registering email change (store user's email in session for later comparison by
+        # process_request function of middleware)
+        EmailChangeMiddleware.register_email_change(request=self.request, email=self.user.email)
+
+        # Ensure email is set in the session
+        self.assertEqual(self.request.session.get('email'), self.user.email)
+        # Ensure session cookie exist
+        self.assertEqual(len(self.client.response.cookies), 1)
+
+        # simulating email changed in some other browser
+        self.user.email = 'new_email@test.com'
+        self.user.save()
+
+        # Call process_request
+        EmailChangeMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Verify that session_email_mismatch and is_enforce_session_email_match_enabled
+        # custom attributes are set
+        mock_set_custom_attribute.assert_has_calls([call('session_email_mismatch', True)])
+        mock_set_custom_attribute.assert_has_calls([call('is_enforce_session_email_match_enabled', True)])
+
+        # Assert that the session is flushed and cookies marked for deletion
+        mock_mark_cookie_for_deletion.assert_called()
+        assert self.request.session.get(SESSION_KEY) is None
+        assert self.request.user == AnonymousUser()
+
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=False)
+    @patch('openedx.core.djangoapps.safe_sessions.middleware._mark_cookie_for_deletion')
+    @patch("openedx.core.djangoapps.safe_sessions.middleware.set_custom_attribute")
+    def test_process_request_emails_mismatch_with_toggle_disabled(
+        self, mock_set_custom_attribute, mock_mark_cookie_for_deletion
+    ):
+        """
+        Calls EmailChangeMiddleware.process_request when user is authenticated,
+        ENFORCE_SESSION_EMAIL_MATCH is disabled and user session and request
+        email mismatch. (Email was changed in some other browser)
+        Verifies that session and cookies are not affected.
+        """
+        # Log in the user
+        self.client.login(email=self.user.email, password=self.PASSWORD)
+        self.request.session = self.client.session
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # Registering email change (store user's email in session for later comparison by
+        # process_request function of middleware)
+        EmailChangeMiddleware.register_email_change(request=self.request, email=self.user.email)
+
+        # Ensure email is set in the session
+        self.assertEqual(self.request.session.get('email'), self.user.email)
+        # Ensure session cookie exist
+        self.assertEqual(len(self.client.response.cookies), 1)
+
+        # simulating email changed in some other browser
+        self.user.email = 'new_email@test.com'
+        self.user.save()
+
+        # Call process_request
+        EmailChangeMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Verify that session_email_mismatch and is_enforce_session_email_match_enabled
+        # custom attributes are set
+        mock_set_custom_attribute.assert_has_calls([call('session_email_mismatch', True)])
+        mock_set_custom_attribute.assert_has_calls([call('is_enforce_session_email_match_enabled', False)])
+
+        # Assert that the session and cookies are not affected
+        self.assertNotEqual(self.request.session.get('email'), self.user.email)
+        self.assertEqual(len(self.client.response.cookies), 1)
+        self.assertEqual(self.client.response.cookies[settings.SESSION_COOKIE_NAME].value, 'authenticated')
+
+        # Assert that _mark_cookie_for_deletion not called
+        mock_mark_cookie_for_deletion.assert_not_called()
+
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=True)
+    @patch('openedx.core.djangoapps.safe_sessions.middleware._mark_cookie_for_deletion')
+    def test_process_request_no_email_change_history_with_toggle_enabled(
+        self, mock_mark_cookie_for_deletion
+    ):
+        """
+        Calls EmailChangeMiddleware.process_request when there is no previous
+        history of an email change and ENFORCE_SESSION_EMAIL_MATCH is enabled
+        Verifies that existing sessions are not affected.
+        Test that sessions predating this code are not affected.
+        """
+        # Log in the user (Simulating user logged-in before this code and email was not set in session)
+        self.client.login(email=self.user.email, password=self.PASSWORD)
+        self.request.session = self.client.session
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # Ensure there is no email in the session denoting no previous history of email change
+        self.assertEqual(self.request.session.get('email'), None)
+
+        # Ensure session cookie exist
+        self.assertEqual(len(self.client.response.cookies), 1)
+
+        # simulating email changed in some other browser
+        self.user.email = 'new_email@test.com'
+        self.user.save()
+
+        # Call process_request
+        EmailChangeMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Assert that the session and cookies are not affected
+        self.assertEqual(len(self.client.response.cookies), 1)
+        self.assertEqual(self.client.response.cookies[settings.SESSION_COOKIE_NAME].value, 'authenticated')
+
+        # Assert that _mark_cookie_for_deletion not called
+        mock_mark_cookie_for_deletion.assert_not_called()
+
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=False)
+    @patch('openedx.core.djangoapps.safe_sessions.middleware._mark_cookie_for_deletion')
+    def test_process_request_no_email_change_history_with_toggle_disabled(
+        self, mock_mark_cookie_for_deletion
+    ):
+        """
+        Calls EmailChangeMiddleware.process_request when there is no previous
+        history of an email change and ENFORCE_SESSION_EMAIL_MATCH is disabled
+        Verifies that existing sessions are not affected.
+        Test that sessions predating this code are not affected.
+        """
+        # Log in the user (Simulating user logged-in before this code and email was not set in session)
+        self.client.login(email=self.user.email, password=self.PASSWORD)
+        self.request.session = self.client.session
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # Ensure there is no email in the session denoting no previous history of email change
+        self.assertEqual(self.request.session.get('email'), None)
+
+        # Ensure session cookie exist
+        self.assertEqual(len(self.client.response.cookies), 1)
+
+        # simulating email changed in some other browser
+        self.user.email = 'new_email@test.com'
+        self.user.save()
+
+        # Call process_request
+        EmailChangeMiddleware(get_response=lambda request: None).process_request(self.request)
+
+        # Assert that the session and cookies are not affected
+        self.assertEqual(len(self.client.response.cookies), 1)
+        self.assertEqual(self.client.response.cookies[settings.SESSION_COOKIE_NAME].value, 'authenticated')
+
+        # Assert that _mark_cookie_for_deletion not called
+        mock_mark_cookie_for_deletion.assert_not_called()
+
+    @patch("openedx.core.djangoapps.safe_sessions.middleware.set_logged_in_cookies")
+    def test_process_response_user_not_authenticated(self, mock_set_logged_in_cookies):
+        """
+        Calls EmailChangeMiddleware.process_response when user is not authenticated.
+        Verify that the logged-in cookies are not updated
+        """
+        # return value of mock
+        mock_set_logged_in_cookies.return_value = self.client.response
+
+        # Unauthenticated User
+        self.request.user = AnonymousUser()
+
+        # Call process_response without authenticating a user
+        response = EmailChangeMiddleware(get_response=lambda request: None).process_response(
+            self.request, self.client.response
+        )
+
+        assert response.status_code == 200
+        # Assert that cookies are not updated
+        # Assert that mock_set_logged_in_cookies not called
+        mock_set_logged_in_cookies.assert_not_called()
+
+    @patch("openedx.core.djangoapps.safe_sessions.middleware.set_logged_in_cookies")
+    def test_process_response_user_authenticated_but_email_change_not_requested(self, mock_set_logged_in_cookies):
+        """
+        Calls EmailChangeMiddleware.process_response when user is authenticated but email
+        change was not requested.
+        Verify that the logged-in cookies are not updated
+        """
+        # return value of mock
+        mock_set_logged_in_cookies.return_value = self.client.response
+
+        # Log in the user
+        self.client.login(email=self.user.email, password=self.PASSWORD)
+        self.request.session = self.client.session
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # No call to register_email_change to indicate email was not changed
+
+        # Call process_response
+        response = EmailChangeMiddleware(get_response=lambda request: None).process_response(
+            self.request, self.client.response
+        )
+
+        assert response.status_code == 200
+        # Assert that cookies are not updated
+        # Assert that mock_set_logged_in_cookies not called
+        mock_set_logged_in_cookies.assert_not_called()
+
+    @patch("openedx.core.djangoapps.safe_sessions.middleware.set_logged_in_cookies")
+    def test_process_response_user_authenticated_and_email_change_requested(self, mock_set_logged_in_cookies):
+        """
+        Calls EmailChangeMiddleware.process_response when user is authenticated and email
+        change was requested.
+        Verify that the logged-in cookies are updated
+        """
+        # return value of mock
+        mock_set_logged_in_cookies.return_value = self.client.response
+
+        # Log in the user
+        self.client.login(email=self.user.email, password=self.PASSWORD)
+        self.request.session = self.client.session
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # Registering email change (setting a variable `email_change_requested` to indicate email was changed)
+        # so that process_response can update cookies
+        EmailChangeMiddleware.register_email_change(request=self.request, email=self.user.email)
+
+        # Call process_response
+        response = EmailChangeMiddleware(get_response=lambda request: None).process_response(
+            self.request, self.client.response
+        )
+
+        assert response.status_code == 200
+        # Assert that cookies are updated
+        # Assert that mock_set_logged_in_cookies is called
+        mock_set_logged_in_cookies.assert_called()
+
+    def test_process_response_no_email_in_session(self):
+        """
+        Calls EmailChangeMiddleware.process_response when user is authenticated and
+        user's email was not stored in user's session.
+        Verify that the user's email is stored in session
+        """
+        # Log in the user
+        self.client.login(email=self.user.email, password=self.PASSWORD)
+        self.request.session = self.client.session
+        self.client.response.set_cookie(settings.SESSION_COOKIE_NAME, 'authenticated')  # Add some logged-in cookie
+
+        # Ensure there is no email in the session
+        self.assertEqual(self.request.session.get('email'), None)
+
+        # Call process_response
+        response = EmailChangeMiddleware(get_response=lambda request: None).process_response(
+            self.request, self.client.response
+        )
+
+        assert response.status_code == 200
+        # Verify that email is set in the session
+        self.assertEqual(self.request.session.get('email'), self.user.email)
+
+    @patch.dict("django.conf.settings.FEATURES", {"DISABLE_SET_JWT_COOKIES_FOR_TESTS": False})
+    def test_user_remain_authenticated_on_email_change_in_other_browser_with_toggle_disabled(self):
+        """
+        Integration Test: test that a user remains authenticated upon email change
+        in other browser when ENFORCE_SESSION_EMAIL_MATCH toggle is disabled
+        Verify that the session and cookies are not affected in current browser and
+        user remains authenticated
+        """
+        setup_login_oauth_client()
+
+        # Login the user with 'test@example.com` email and test password in current browser
+        response = self.client.post(self.login_url, {
+            "email_or_username": self.EMAIL,
+            "password": self.PASSWORD,
+        })
+        # Verify that the user is logged in successfully in current browser
+        assert response.status_code == 200
+        # Verify that the logged-in cookies are set in current browser
+        self._assert_logged_in_cookies_present(response)
+
+        # Verify that the authenticated user can access the dashboard in current browser
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 200
+
+        # simulating email changed in some other browser (Email is changed in DB)
+        self.user.email = 'new_email@test.com'
+        self.user.save()
+
+        # Verify that the user remains authenticated in current browser and can access the dashboard
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 200
+
+    @patch.dict("django.conf.settings.FEATURES", {"DISABLE_SET_JWT_COOKIES_FOR_TESTS": False})
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=True)
+    def test_cookies_are_updated_with_new_email_on_email_change_with_toggle_enabled(self):
+        """
+        Integration Test: test that cookies are updated with new email upon email change
+        in current browser regardless of toggle setting
+        Verify that the cookies are updated in current browser and
+        user remains authenticated
+        """
+        setup_login_oauth_client()
+
+        # Login the user with 'test@example.com` email and test password in current browser
+        login_response = self.client.post(self.login_url, {
+            "email_or_username": self.EMAIL,
+            "password": self.PASSWORD,
+        })
+        # Verify that the user is logged in successfully in current browser
+        assert login_response.status_code == 200
+        # Verify that the logged-in cookies are set in current browser
+        self._assert_logged_in_cookies_present(login_response)
+
+        # Verify that the authenticated user can access the dashboard in current browser
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 200
+
+        # simulating email change in current browser
+        activation_key = uuid.uuid4().hex
+        PendingEmailChange.objects.update_or_create(
+            user=self.user,
+            defaults={
+                'new_email': 'new_email@test.com',
+                'activation_key': activation_key,
+            }
+        )
+        email_change_response = self.client.get(
+            reverse('confirm_email_change', kwargs={'key': activation_key}),
+        )
+
+        # Verify that email change is successful
+        assert email_change_response.status_code == 200
+        self._assert_logged_in_cookies_present(email_change_response)
+
+        # Verify that jwt cookies are updated with new email and
+        # not equal to old logged-in cookies in current browser
+        self.assertNotEqual(
+            login_response.cookies[jwt_cookies.jwt_cookie_header_payload_name()].value,
+            email_change_response.cookies[jwt_cookies.jwt_cookie_header_payload_name()].value
+        )
+        self.assertNotEqual(
+            login_response.cookies[jwt_cookies.jwt_cookie_signature_name()].value,
+            email_change_response.cookies[jwt_cookies.jwt_cookie_signature_name()].value
+        )
+
+    @patch.dict("django.conf.settings.FEATURES", {"DISABLE_SET_JWT_COOKIES_FOR_TESTS": False})
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=False)
+    def test_cookies_are_updated_with_new_email_on_email_change_with_toggle_disabled(self):
+        """
+        Integration Test: test that cookies are updated with new email upon email change
+        in current browser regardless of toggle setting
+        Verify that the cookies are updated in current browser and
+        user remains authenticated
+        """
+        setup_login_oauth_client()
+
+        # Login the user with 'test@example.com` email and test password in current browser
+        login_response = self.client.post(self.login_url, {
+            "email_or_username": self.EMAIL,
+            "password": self.PASSWORD,
+        })
+        # Verify that the user is logged in successfully in current browser
+        assert login_response.status_code == 200
+        # Verify that the logged-in cookies are set in current browser
+        self._assert_logged_in_cookies_present(login_response)
+
+        # Verify that the authenticated user can access the dashboard in current browser
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 200
+
+        # simulating email change in current browser
+        activation_key = uuid.uuid4().hex
+        PendingEmailChange.objects.update_or_create(
+            user=self.user,
+            defaults={
+                'new_email': 'new_email@test.com',
+                'activation_key': activation_key,
+            }
+        )
+        email_change_response = self.client.get(
+            reverse('confirm_email_change', kwargs={'key': activation_key}),
+        )
+
+        # Verify that email change is successful
+        assert email_change_response.status_code == 200
+        self._assert_logged_in_cookies_present(email_change_response)
+
+        # Verify that jwt cookies are updated with new email and
+        # not equal to old logged-in cookies in current browser
+        self.assertNotEqual(
+            login_response.cookies[jwt_cookies.jwt_cookie_header_payload_name()].value,
+            email_change_response.cookies[jwt_cookies.jwt_cookie_header_payload_name()].value
+        )
+        self.assertNotEqual(
+            login_response.cookies[jwt_cookies.jwt_cookie_signature_name()].value,
+            email_change_response.cookies[jwt_cookies.jwt_cookie_signature_name()].value
+        )
+
+    @patch.dict("django.conf.settings.FEATURES", {"DISABLE_SET_JWT_COOKIES_FOR_TESTS": False})
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=True)
+    def test_logged_in_user_unauthenticated_on_email_change_in_other_browser(self):
+        """
+        Integration Test: Test that a user logged-in in one browser gets unauthenticated
+        when the email is changed in some other browser and the request and session emails mismatch.
+        Verify that the session is invalidated and cookies are deleted in current browser
+        and user gets unauthenticated.
+        """
+        setup_login_oauth_client()
+
+        # Login the user with 'test@example.com` email and test password in current browser
+        response = self.client.post(self.login_url, {
+            "email_or_username": self.EMAIL,
+            "password": self.PASSWORD,
+        })
+        # Verify that the user is logged in successfully in current browser
+        assert response.status_code == 200
+        # Verify that the logged-in cookies are set in current browser
+        self._assert_logged_in_cookies_present(response)
+
+        # Verify that the authenticated user can access the dashboard in current browser
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 200
+
+        # simulating email changed in some other browser (Email is changed in DB)
+        self.user.email = 'new_email@test.com'
+        self.user.save()
+
+        # Verify that the user gets unauthenticated in current browser and cannot access the dashboard
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 302
+        self._assert_logged_in_cookies_not_present(response)
+
+    @patch.dict("django.conf.settings.FEATURES", {"DISABLE_SET_JWT_COOKIES_FOR_TESTS": False})
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=True)
+    def test_logged_in_user_remains_authenticated_on_email_change_in_same_browser(self):
+        """
+        Integration Test: test that a user logged-in in some browser remains authenticated
+        when the email is changed in same browser.
+        Verify that the session and cookies are updated in current browser and
+        user remains authenticated
+        """
+        setup_login_oauth_client()
+
+        # Login the user with 'test@example.com` email and test password in current browser
+        response = self.client.post(self.login_url, {
+            "email_or_username": self.EMAIL,
+            "password": self.PASSWORD,
+        })
+        # Verify that the user is logged in successfully in current browser
+        assert response.status_code == 200
+        # Verify that the logged-in cookies are set in current browser
+        self._assert_logged_in_cookies_present(response)
+
+        # Verify that the authenticated user can access the dashboard in current browser
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 200
+
+        # simulating email change in current browser
+        activation_key = uuid.uuid4().hex
+        PendingEmailChange.objects.update_or_create(
+            user=self.user,
+            defaults={
+                'new_email': 'new_email@test.com',
+                'activation_key': activation_key,
+            }
+        )
+        email_change_response = self.client.get(
+            reverse('confirm_email_change', kwargs={'key': activation_key}),
+        )
+
+        # Verify that email change is successful and all logged-in
+        # cookies are set in current browser
+        assert email_change_response.status_code == 200
+        self._assert_logged_in_cookies_present(email_change_response)
+
+        # Verify that the user remains authenticated in current browser and can access the dashboard
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 200
+
+    @patch.dict("django.conf.settings.FEATURES", {"DISABLE_SET_JWT_COOKIES_FOR_TESTS": False})
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=True)
+    def test_registered_user_unauthenticated_on_email_change_in_other_browser(self):
+        """
+        Integration Test: Test that a user registered in one browser gets unauthenticated
+        when the email is changed in some other browser and the request and session emails mismatch.
+        Verify that the session is invalidated and cookies are deleted in current browser
+        and user gets unauthenticated
+        """
+        setup_login_oauth_client()
+
+        # Register the user with 'john_doe@example.com` email and test password in current browser
+        response = self.client.post(self.register_url, {
+            "email": 'john_doe@example.com',
+            "name": 'John Doe',
+            "username": 'john_doe',
+            "password": 'password',
+            "honor_code": "true",
+        })
+        # Verify that the user is logged in successfully in current browser
+        assert response.status_code == 200
+        # Verify that the logged-in cookies are set in current browser
+        self._assert_logged_in_cookies_present(response)
+
+        # Verify that the authenticated user can access the dashboard in current browser
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 200
+
+        # simulating email changed in some other browser (Email is changed in DB)
+        registered_user = User.objects.get(email='john_doe@example.com')
+        registered_user.email = 'new_email@test.com'
+        registered_user.save()
+
+        # Verify that the user get unauthenticated in current browser and cannot access the dashboard
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 302
+        self._assert_logged_in_cookies_not_present(response)
+
+    @patch.dict("django.conf.settings.FEATURES", {"DISABLE_SET_JWT_COOKIES_FOR_TESTS": False})
+    @override_settings(ENFORCE_SESSION_EMAIL_MATCH=True)
+    def test_registered_user_remain_authenticated_on_email_change_in_same_browser(self):
+        """
+        Integration Test: test that a user registered in one browser remains
+        authenticated in current browser when the email is changed in same browser.
+        Verify that the session and cookies updated and user remains
+        authenticated in current browser
+        """
+        setup_login_oauth_client()
+
+        # Register the user with 'john_doe@example.com` email and test password in current browser
+        response = self.client.post(self.register_url, {
+            "email": 'john_doe@example.com',
+            "name": 'John Doe',
+            "username": 'john_doe',
+            "password": 'password',
+            "honor_code": "true",
+        })
+        # Verify that the user is logged in successfully in current browser
+        assert response.status_code == 200
+        # Verify that the logged-in cookies are set in current browser
+        self._assert_logged_in_cookies_present(response)
+
+        # Verify that the authenticated user can access the dashboard in current browser
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 200
+
+        # getting newly created user
+        registered_user = User.objects.get(email='john_doe@example.com')
+
+        # simulating email change in current browser
+        activation_key = uuid.uuid4().hex
+        PendingEmailChange.objects.update_or_create(
+            user=registered_user,
+            defaults={
+                'new_email': 'new_email@test.com',
+                'activation_key': activation_key,
+            }
+        )
+        email_change_response = self.client.get(
+            reverse('confirm_email_change', kwargs={'key': activation_key}),
+        )
+
+        # Verify that email change is successful and all logged-in
+        # cookies are updated with new email in current browser
+        assert email_change_response.status_code == 200
+        self._assert_logged_in_cookies_present(email_change_response)
+
+        # Verify that the user remains authenticated in current browser and can access the dashboard
+        response = self.client.get(self.dashboard_url)
+        assert response.status_code == 200
+
+    def _assert_logged_in_cookies_present(self, response):
+        """
+        Helper function to verify that all logged-in cookies are available
+        and have valid values (not empty strings)
+        """
+        all_cookies = ALL_LOGGED_IN_COOKIE_NAMES + (settings.SESSION_COOKIE_NAME,)
+
+        for cookie in all_cookies:
+            # Check if the cookie is present in response.cookies.keys()
+            self.assertIn(cookie, response.cookies.keys())
+
+            # Assert that the value is not an empty string
+            self.assertNotEqual(response.cookies[cookie].value, "")
+
+    def _assert_logged_in_cookies_not_present(self, response):
+        """
+        Helper function to verify that all logged-in cookies are cleared
+        and have empty values
+        """
+        all_cookies = ALL_LOGGED_IN_COOKIE_NAMES + (settings.SESSION_COOKIE_NAME,)
+
+        for cookie in all_cookies:
+            # Check if the cookie is present in response.cookies.keys()
+            self.assertIn(cookie, response.cookies.keys())
+
+            # Assert that the value is not an empty string
+            self.assertEqual(response.cookies[cookie].value, "")

openedx/core/djangoapps/user_api/accounts/tests/test_views.py

@@ -232,7 +232,7 @@ class TestOwnUsernameAPI(FilteredQueryCountMixin, CacheIsolationTestCase, UserAP
         Test that a client (logged in) can get her own username.
         """
         self.client.login(username=self.user.username, password=TEST_PASSWORD)
-        self._verify_get_own_username(16)
+        self._verify_get_own_username(19)
 
     def test_get_username_inactive(self):
         """
@@ -242,7 +242,7 @@ class TestOwnUsernameAPI(FilteredQueryCountMixin, CacheIsolationTestCase, UserAP
         self.client.login(username=self.user.username, password=TEST_PASSWORD)
         self.user.is_active = False
         self.user.save()
-        self._verify_get_own_username(16)
+        self._verify_get_own_username(19)
 
     def test_get_username_not_logged_in(self):
         """
@@ -358,7 +358,7 @@ class TestAccountsAPI(FilteredQueryCountMixin, CacheIsolationTestCase, UserAPITe
     """
 
     ENABLED_CACHES = ['default']
-    TOTAL_QUERY_COUNT = 24
+    TOTAL_QUERY_COUNT = 27
     FULL_RESPONSE_FIELD_COUNT = 29
 
     def setUp(self):
@@ -811,7 +811,7 @@ class TestAccountsAPI(FilteredQueryCountMixin, CacheIsolationTestCase, UserAPITe
             assert data['time_zone'] is None
 
         self.client.login(username=self.user.username, password=TEST_PASSWORD)
-        verify_get_own_information(self._get_num_queries(22))
+        verify_get_own_information(self._get_num_queries(25))
 
         # Now make sure that the user can get the same information, even if not active
         self.user.is_active = False
@@ -831,7 +831,7 @@ class TestAccountsAPI(FilteredQueryCountMixin, CacheIsolationTestCase, UserAPITe
         legacy_profile.save()
 
         self.client.login(username=self.user.username, password=TEST_PASSWORD)
-        with self.assertNumQueries(self._get_num_queries(22), table_ignorelist=WAFFLE_TABLES):
+        with self.assertNumQueries(self._get_num_queries(25), table_ignorelist=WAFFLE_TABLES):
             response = self.send_get(self.client)
         for empty_field in ("level_of_education", "gender", "country", "state", "bio",):
             assert response.data[empty_field] is None