diff --git a/cms/djangoapps/contentstore/config/waffle.py b/cms/djangoapps/contentstore/config/waffle.py
index 16719e61bf29..c1970b35577e 100644
--- a/cms/djangoapps/contentstore/config/waffle.py
+++ b/cms/djangoapps/contentstore/config/waffle.py
@@ -40,20 +40,6 @@
     f'{WAFFLE_NAMESPACE}.library_authoring_mfe', __name__, LOG_PREFIX
 )
 
-
-# .. toggle_name: studio.custom_relative_dates
-# .. toggle_implementation: CourseWaffleFlag
-# .. toggle_default: False
-# .. toggle_description: Waffle flag to enable custom pacing input for Personalized Learner Schedule (PLS).
-# ..    This flag guards an input in Studio for a self paced course, where the user can enter date offsets
-# ..    for a subsection.
-# .. toggle_use_cases: temporary
-# .. toggle_creation_date: 2021-07-12
-# .. toggle_target_removal_date: 2021-12-31
-# .. toggle_warning: Flag course_experience.relative_dates should also be active for relative dates functionalities to work.
-# .. toggle_tickets: https://openedx.atlassian.net/browse/AA-844
-CUSTOM_RELATIVE_DATES = CourseWaffleFlag(f'{WAFFLE_NAMESPACE}.custom_relative_dates', __name__)
-
 # .. toggle_name: studio.prevent_staff_structure_deletion
 # .. toggle_implementation: WaffleFlag
 # .. toggle_default: False
diff --git a/cms/templates/base.html b/cms/templates/base.html
index 1e88505c7fde..322a40055e33 100644
--- a/cms/templates/base.html
+++ b/cms/templates/base.html
@@ -10,8 +10,8 @@
 <%!
 from django.utils.translation import gettext as _
 
-from cms.djangoapps.contentstore.config.waffle import CUSTOM_RELATIVE_DATES
 from lms.djangoapps.branding import api as branding_api
+from openedx.core.djangoapps.course_date_signals.waffle import CUSTOM_RELATIVE_DATES
 from openedx.core.djangoapps.util.user_messages import PageLevelMessages
 from openedx.core.djangolib.js_utils import (
     dump_js_escaped_json, js_escaped_string
diff --git a/lms/djangoapps/learner_recommendations/tests/test_views.py b/lms/djangoapps/learner_recommendations/tests/test_views.py
index d582eaf334a8..70f19fedb6ac 100644
--- a/lms/djangoapps/learner_recommendations/tests/test_views.py
+++ b/lms/djangoapps/learner_recommendations/tests/test_views.py
@@ -154,6 +154,7 @@ def test_successful_response(
         assert segment_mock.call_args[0][1] == "edx.bi.user.recommendations.viewed"
 
 
+@ddt.ddt
 class TestCrossProductRecommendationsView(APITestCase):
     """Unit tests for the Cross Product Recommendations View"""
 
@@ -218,13 +219,12 @@ def _get_recommended_courses(self, num_of_courses_with_restriction=0):
     @mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
     @mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
     @mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
-    def test_successful_response(
-        self, country_code_from_ip_mock, get_course_data_mock,
-    ):
+    @ddt.data("za", "")  # Ensure that the empty string is handled correctly.
+    def test_successful_response(self, country_code, country_code_from_ip_mock, get_course_data_mock):
         """
         Verify 2 cross product course recommendations are returned.
         """
-        country_code_from_ip_mock.return_value = "za"
+        country_code_from_ip_mock.return_value = country_code
         mock_course_data = self._get_recommended_courses()
         get_course_data_mock.side_effect = [mock_course_data[0], mock_course_data[1]]
 
diff --git a/lms/djangoapps/user_tours/toggles.py b/lms/djangoapps/user_tours/toggles.py
new file mode 100644
index 000000000000..9b97c724d390
--- /dev/null
+++ b/lms/djangoapps/user_tours/toggles.py
@@ -0,0 +1,15 @@
+"""
+Toggles for the User Tours Experience.
+"""
+
+from edx_toggles.toggles import WaffleFlag
+
+# .. toggle_name: user_tours.tours_disabled
+# .. toggle_implementation: WaffleFlag
+# .. toggle_default: False
+# .. toggle_description: This flag disables user tours in LMS.
+# .. toggle_warnings: None
+# .. toggle_use_cases: temporary
+# .. toggle_creation_date: 2021-12-13
+# .. toggle_target_removal_date: None
+USER_TOURS_DISABLED = WaffleFlag('user_tours.tours_disabled', module_name=__name__, log_prefix='user_tours')
diff --git a/lms/djangoapps/user_tours/v1/tests/test_views.py b/lms/djangoapps/user_tours/v1/tests/test_views.py
index 57aa7ca5dcc6..f78586b78dec 100644
--- a/lms/djangoapps/user_tours/v1/tests/test_views.py
+++ b/lms/djangoapps/user_tours/v1/tests/test_views.py
@@ -5,11 +5,13 @@
 from django.db.models.signals import post_save
 from django.test import TestCase, override_settings
 from django.urls import reverse
+from edx_toggles.toggles.testutils import override_waffle_flag
 from rest_framework import status
 
 from common.djangoapps.student.tests.factories import UserFactory
 from lms.djangoapps.user_tours.handlers import init_user_tour
 from lms.djangoapps.user_tours.models import UserTour, UserDiscussionsTours
+from lms.djangoapps.user_tours.toggles import USER_TOURS_DISABLED
 from openedx.core.djangoapps.oauth_dispatch.jwt import create_jwt_for_user
 
 User = get_user_model()
@@ -47,6 +49,13 @@ def send_request(self, jwt_user, request_user, method, data=None):
         elif method == 'PATCH':
             return self.client.patch(url, data, content_type='application/json', **headers)
 
+    @ddt.data('GET', 'PATCH')
+    @override_waffle_flag(USER_TOURS_DISABLED, active=True)
+    def test_tours_disabled(self, method):
+        """ Test that the tours can be turned off with a waffle flag. """
+        response = self.send_request(self.staff_user, self.user, method)
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
     @ddt.data('GET', 'PATCH')
     def test_unauthorized_user(self, method):
         """ Test all endpoints if request does not have jwt auth. """
@@ -188,6 +197,11 @@ def test_get_tours(self):
         self.assertEqual(response.data[1]['tour_name'], 'not_responded_filter')
         self.assertTrue(response.data[1]['show_tour'])
 
+        # Test that the view can be disabled by a waffle flag.
+        with override_waffle_flag(USER_TOURS_DISABLED, active=True):
+            response = self.client.get(self.url, **headers)
+            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
     def test_get_tours_unauthenticated(self):
         """
         Test that an unauthenticated user cannot access the discussion tours endpoint.
@@ -215,3 +229,8 @@ def test_update_tour(self):
         # Check that the tour was updated in the database
         updated_tour = UserDiscussionsTours.objects.get(id=self.tour.id)
         self.assertEqual(updated_tour.show_tour, False)
+
+        # Test that the view can be disabled by a waffle flag.
+        with override_waffle_flag(USER_TOURS_DISABLED, active=True):
+            response = self.client.put(url, updated_data, content_type='application/json', **headers)
+            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
diff --git a/lms/djangoapps/user_tours/v1/views.py b/lms/djangoapps/user_tours/v1/views.py
index dca1964b64db..ce4c354e5dc2 100644
--- a/lms/djangoapps/user_tours/v1/views.py
+++ b/lms/djangoapps/user_tours/v1/views.py
@@ -10,6 +10,7 @@
 from rest_framework import status
 
 from lms.djangoapps.user_tours.models import UserTour, UserDiscussionsTours
+from lms.djangoapps.user_tours.toggles import USER_TOURS_DISABLED
 from lms.djangoapps.user_tours.v1.serializers import UserTourSerializer, UserDiscussionsToursSerializer
 
 from rest_framework.views import APIView
@@ -41,9 +42,12 @@ def get(self, request, username):  # pylint: disable=arguments-differ
 
             400 if there is a not allowed request (requesting a user you don't have access to)
             401 if unauthorized request
-            403 if waffle flag is not enabled
+            403 if tours are disabled
             404 if the UserTour does not exist (shouldn't happen, but safety first)
         """
+        if USER_TOURS_DISABLED.is_enabled():
+            return Response(status=status.HTTP_403_FORBIDDEN)
+
         if request.user.username != username and not request.user.is_staff:
             return Response(status=status.HTTP_400_BAD_REQUEST)
 
@@ -66,8 +70,11 @@ def patch(self, request, username):  # pylint: disable=arguments-differ
 
             400 if update was unsuccessful or there was nothing to update
             401 if unauthorized request
-            403 if waffle flag is not enabled
+            403 if tours are disabled
         """
+        if USER_TOURS_DISABLED.is_enabled():
+            return Response(status=status.HTTP_403_FORBIDDEN)
+
         if request.user.username != username:
             return Response(status=status.HTTP_400_BAD_REQUEST)
 
@@ -125,8 +132,11 @@ def get(self, request, tour_id=None):
                     "user": 1
                 }
              ]
+            403 if the tours are disabled
 
         """
+        if USER_TOURS_DISABLED.is_enabled():
+            return Response(status=status.HTTP_403_FORBIDDEN)
         try:
             with transaction.atomic():
                 tours = UserDiscussionsTours.objects.filter(user=request.user)
@@ -158,9 +168,11 @@ def put(self, request, tour_id):
         Returns:
             200: The updated tour, serialized using the UserDiscussionsToursSerializer
             404: If the tour does not exist
-            403: If the user does not have permission to update the tour
+            403: If the user does not have permission to update the tour or the tours are disabled
             400: Validation error
         """
+        if USER_TOURS_DISABLED.is_enabled():
+            return Response(status=status.HTTP_403_FORBIDDEN)
         tour = get_object_or_404(UserDiscussionsTours, pk=tour_id)
         if tour.user != request.user:
             return Response(status=status.HTTP_403_FORBIDDEN)
diff --git a/openedx/core/djangoapps/content/course_overviews/models.py b/openedx/core/djangoapps/content/course_overviews/models.py
index 149159826ccf..edfe240e294b 100644
--- a/openedx/core/djangoapps/content/course_overviews/models.py
+++ b/openedx/core/djangoapps/content/course_overviews/models.py
@@ -2,10 +2,12 @@
 Declaration of CourseOverview model
 """
 
+from __future__ import annotations
 
 import json
 import logging
 from datetime import datetime
+from typing import List
 from urllib.parse import urlparse, urlunparse
 
 import pytz
@@ -17,6 +19,7 @@
 from django.db.models.signals import post_save, post_delete
 from django.db.utils import IntegrityError
 from django.template import defaultfilters
+from opaque_keys.edx.keys import CourseKey
 
 from django.utils.functional import cached_property
 from model_utils.models import TimeStampedModel
@@ -696,10 +699,16 @@ def get_all_courses(cls, orgs=None, filter_=None, active_only=False, course_keys
         return course_overviews
 
     @classmethod
-    def get_all_course_keys(cls):
+    def get_all_course_keys(cls, self_paced: bool | None = None) -> List[CourseKey]:
         """
-        Returns all course keys from course overviews.
+        Returns all course keys from course overviews, optionally filter by pacing.
+        The filter is only used when a boolean is passed as argument and it is disabled when this value is `None`.
+
+        Args:
+            self_paced: Optionally filter by pacing
         """
+        if self_paced is not None:
+            return CourseOverview.objects.filter(self_paced=self_paced).values_list('id', flat=True)
         return CourseOverview.objects.values_list('id', flat=True)
 
     def is_discussion_tab_enabled(self):
diff --git a/openedx/core/djangoapps/course_date_signals/handlers.py b/openedx/core/djangoapps/course_date_signals/handlers.py
index 6f3f4ed9a713..46a0712475fd 100644
--- a/openedx/core/djangoapps/course_date_signals/handlers.py
+++ b/openedx/core/djangoapps/course_date_signals/handlers.py
@@ -8,7 +8,7 @@
 from edx_when.api import FIELDS_TO_EXTRACT, set_dates_for_course
 from xblock.fields import Scope
 
-from cms.djangoapps.contentstore.config.waffle import CUSTOM_RELATIVE_DATES
+from openedx.core.djangoapps.course_date_signals.waffle import CUSTOM_RELATIVE_DATES
 from openedx.core.lib.graph_traversals import get_children, leaf_filter, traverse_pre_order
 from xmodule.modulestore import ModuleStoreEnum  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.django import SignalHandler, modulestore  # lint-amnesty, pylint: disable=wrong-import-order
diff --git a/openedx/core/djangoapps/course_date_signals/tests.py b/openedx/core/djangoapps/course_date_signals/tests.py
index ee1e95b7b5a2..288ce11df8e9 100644
--- a/openedx/core/djangoapps/course_date_signals/tests.py
+++ b/openedx/core/djangoapps/course_date_signals/tests.py
@@ -6,13 +6,13 @@
 from xmodule.modulestore.tests.django_utils import TEST_DATA_SPLIT_MODULESTORE, ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory, BlockFactory
 
-from cms.djangoapps.contentstore.config.waffle import CUSTOM_RELATIVE_DATES
 from openedx.core.djangoapps.course_date_signals.handlers import (
     _gather_graded_items,
     _get_custom_pacing_children,
     _has_assignment_blocks,
     extract_dates_from_course
 )
+from openedx.core.djangoapps.course_date_signals.waffle import CUSTOM_RELATIVE_DATES
 from openedx.core.djangoapps.course_date_signals.models import SelfPacedRelativeDatesConfig
 
 from . import utils
diff --git a/openedx/core/djangoapps/course_date_signals/utils.py b/openedx/core/djangoapps/course_date_signals/utils.py
index 78447f32610c..15b4509e8a06 100644
--- a/openedx/core/djangoapps/course_date_signals/utils.py
+++ b/openedx/core/djangoapps/course_date_signals/utils.py
@@ -5,8 +5,10 @@
 """
 
 from datetime import timedelta
+from typing import Optional
 
 from openedx.core.djangoapps.catalog.utils import get_course_run_details
+from openedx.core.djangoapps.course_date_signals.waffle import CUSTOM_RELATIVE_DATES
 
 
 MIN_DURATION = timedelta(weeks=4)
@@ -33,7 +35,24 @@ def get_expected_duration(course_id):
     return access_duration
 
 
-def spaced_out_sections(course):
+def get_expected_duration_based_on_relative_due_dates(course) -> timedelta:
+    """
+    Calculate duration based on custom relative due dates.
+    Returns the longest relative due date if set else a minimum duration of 1 week.
+    """
+    duration_in_weeks = 1
+    if CUSTOM_RELATIVE_DATES.is_enabled(course.id):
+        for section in course.get_children():
+            if section.visible_to_staff_only:
+                continue
+            for subsection in section.get_children():
+                relative_weeks_due = subsection.fields['relative_weeks_due'].read_from(subsection)
+                if relative_weeks_due and relative_weeks_due > duration_in_weeks:
+                    duration_in_weeks = relative_weeks_due
+    return timedelta(weeks=duration_in_weeks)
+
+
+def spaced_out_sections(course, duration: Optional[timedelta] = None):
     """
     Generator that returns sections of the course block with a suggested time to complete for each
 
@@ -42,13 +61,14 @@ def spaced_out_sections(course):
         section (block): a section block of the course
         relative time (timedelta): the amount of weeks to complete the section, since start of course
     """
-    duration = get_expected_duration(course.id)
+    if not duration:
+        duration = get_expected_duration(course.id)
     sections = [
         section
         for section
         in course.get_children()
         if not section.visible_to_staff_only
     ]
-    weeks_per_section = duration / len(sections)
+    weeks_per_section = duration / (len(sections) or 1)  # if course has zero sections
     for idx, section in enumerate(sections):
         yield idx, section, weeks_per_section * (idx + 1)
diff --git a/openedx/core/djangoapps/course_date_signals/waffle.py b/openedx/core/djangoapps/course_date_signals/waffle.py
new file mode 100644
index 000000000000..b3073664c1ac
--- /dev/null
+++ b/openedx/core/djangoapps/course_date_signals/waffle.py
@@ -0,0 +1,20 @@
+"""
+This module contains various configuration settings via
+waffle switches for the course_date_signals app.
+"""
+
+
+from openedx.core.djangoapps.waffle_utils import CourseWaffleFlag
+
+# .. toggle_name: studio.custom_relative_dates
+# .. toggle_implementation: CourseWaffleFlag
+# .. toggle_default: False
+# .. toggle_description: Waffle flag to enable custom pacing input for Personalized Learner Schedule (PLS).
+# ..    This flag guards an input in Studio for a self paced course, where the user can enter date offsets
+# ..    for a subsection.
+# .. toggle_use_cases: temporary
+# .. toggle_creation_date: 2021-07-12
+# .. toggle_target_removal_date: 2021-12-31
+# .. toggle_warning: Flag course_experience.relative_dates should also be active for relative dates functionalities to work.
+# .. toggle_tickets: https://openedx.atlassian.net/browse/AA-844
+CUSTOM_RELATIVE_DATES = CourseWaffleFlag('studio.custom_relative_dates', __name__)
diff --git a/openedx/core/djangoapps/geoinfo/api.py b/openedx/core/djangoapps/geoinfo/api.py
index 9445b56fb998..76ca9f3f2fad 100644
--- a/openedx/core/djangoapps/geoinfo/api.py
+++ b/openedx/core/djangoapps/geoinfo/api.py
@@ -22,7 +22,7 @@ def country_code_from_ip(ip_addr: str) -> str:
     try:
         response = reader.country(ip_addr)
         # pylint: disable=no-member
-        country_code = response.country.iso_code
+        country_code = response.country.iso_code or ""
     except geoip2.errors.AddressNotFoundError:
         country_code = ""
     reader.close()
diff --git a/openedx/core/djangoapps/schedules/admin.py b/openedx/core/djangoapps/schedules/admin.py
index 4f5a9cad7b83..31915a4fd408 100644
--- a/openedx/core/djangoapps/schedules/admin.py
+++ b/openedx/core/djangoapps/schedules/admin.py
@@ -176,5 +176,6 @@ class ScheduleConfigAdmin(admin.ModelAdmin):  # lint-amnesty, pylint: disable=mi
         'enqueue_recurring_nudge', 'deliver_recurring_nudge',
         'enqueue_upgrade_reminder', 'deliver_upgrade_reminder',
         'enqueue_course_update', 'deliver_course_update',
+        'enqueue_course_due_date_reminder', 'deliver_course_due_date_reminder',
     )
     form = ScheduleConfigAdminForm
diff --git a/openedx/core/djangoapps/schedules/content_highlights.py b/openedx/core/djangoapps/schedules/content_highlights.py
index c50d00484bc6..9ed035d66727 100644
--- a/openedx/core/djangoapps/schedules/content_highlights.py
+++ b/openedx/core/djangoapps/schedules/content_highlights.py
@@ -1,17 +1,21 @@
 """
-Contains methods for accessing course highlights. Course highlights is a
-schedule experience built on the Schedules app.
+Contains methods for accessing course highlights and course due dates.
+Course highlights is a schedule experience built on the Schedules app.
 """
 
 
+from datetime import timedelta
 import logging
 
+from lms.djangoapps.courseware.courses import get_course
+from lms.djangoapps.courseware.exceptions import CourseRunNotFound
 from openedx.core.djangoapps.course_date_signals.utils import spaced_out_sections
+from openedx.core.djangoapps.course_date_signals.waffle import CUSTOM_RELATIVE_DATES
 from openedx.core.djangoapps.schedules.exceptions import CourseUpdateDoesNotExist
 from openedx.core.lib.request_utils import get_request_or_stub
-from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
 
 log = logging.getLogger(__name__)
+DUE_DATE_FORMAT = "%b %d, %Y at %H:%M %Z"
 
 
 def get_all_course_highlights(course_key):
@@ -67,8 +71,8 @@ def course_has_highlights_from_store(course_key):
         course_key (CourseKey): course to lookup from the modulestore
     """
     try:
-        course = _get_course_descriptor(course_key)
-    except CourseUpdateDoesNotExist:
+        course = get_course(course_key, depth=1)
+    except CourseRunNotFound:
         return False
     return course_has_highlights(course)
 
@@ -93,7 +97,16 @@ def get_week_highlights(user, course_key, week_num):
     return highlights
 
 
-def get_next_section_highlights(user, course_key, start_date, target_date):
+def get_upcoming_subsection_due_dates(user, course_key, start_date, target_date, current_date, duration=None):
+    """
+    Get section due dates, based upon the current date.
+    """
+    course_descriptor = get_course(course_key, depth=2)
+    course_block = _get_course_block(course_descriptor, user)
+    return _get_upcoming_due_dates(course_block, start_date, target_date, current_date, duration)
+
+
+def get_next_section_highlights(user, course_key, start_date, target_date, duration=None):
     """
     Get highlights (list of unicode strings) for a week, based upon the current date.
 
@@ -102,12 +115,12 @@ def get_next_section_highlights(user, course_key, start_date, target_date):
     """
     course_descriptor = _get_course_with_highlights(course_key)
     course_block = _get_course_block(course_descriptor, user)
-    return _get_highlights_for_next_section(course_block, start_date, target_date)
+    return _get_highlights_for_next_section(course_block, start_date, target_date, duration)
 
 
 def _get_course_with_highlights(course_key):
-    """ Gets Course descriptor iff highlights are enabled for the course """
-    course_descriptor = _get_course_descriptor(course_key)
+    """ Gets Course descriptor if highlights are enabled for the course """
+    course_descriptor = get_course(course_key, depth=1)
     if not course_descriptor.highlights_enabled_for_messaging:
         raise CourseUpdateDoesNotExist(
             f'{course_key} Course Update Messages are disabled.'
@@ -116,16 +129,6 @@ def _get_course_with_highlights(course_key):
     return course_descriptor
 
 
-def _get_course_descriptor(course_key):
-    """ Gets course descriptor from modulestore """
-    course_descriptor = modulestore().get_course(course_key, depth=1)
-    if course_descriptor is None:
-        raise CourseUpdateDoesNotExist(
-            f'Course {course_key} not found.'
-        )
-    return course_descriptor
-
-
 def _get_course_block(course_descriptor, user):
     """ Gets course block that takes into account user state and permissions """
     # Adding courseware imports here to insulate other apps (e.g. schedules) to
@@ -146,7 +149,7 @@ def _get_course_block(course_descriptor, user):
         user, request, course_descriptor, field_data_cache, course_descriptor.id, course=course_descriptor,
     )
     if not course_block:
-        raise CourseUpdateDoesNotExist(f'Course block {course_descriptor.id} not found')
+        raise CourseRunNotFound(course_descriptor.id)
     return course_block
 
 
@@ -175,10 +178,10 @@ def _get_highlights_for_week(sections, week_num, course_key):
     return section.highlights
 
 
-def _get_highlights_for_next_section(course, start_date, target_date):
+def _get_highlights_for_next_section(course, start_date, target_date, duration=None):
     """ Using the target date, retrieves highlights for the next section. """
     use_next_sections_highlights = False
-    for index, section, weeks_to_complete in spaced_out_sections(course):
+    for index, section, weeks_to_complete in spaced_out_sections(course, duration):
         # We calculate section due date ourselves (rather than grabbing the due attribute),
         # since not every section has a real due date (i.e. not all are graded), but we still
         # want to know when this section should have been completed by the learner.
@@ -199,3 +202,26 @@ def _get_highlights_for_next_section(course, start_date, target_date):
         )
 
     return None, None
+
+
+def _get_upcoming_due_dates(course, start_date, target_date, current_date, duration=None):
+    """ Retrieves section names and due dates within the provided target_date. """
+    date_items = []
+    # Apply the same relative due date to all content inside a section,
+    # unless that item already has a relative date set
+    for _, section, days_to_complete in spaced_out_sections(course, duration):
+        # Default to Personalized Learner Schedules (PLS) logic for self paced courses.
+        section_due_date = start_date + days_to_complete
+        section_date_items = []
+
+        for subsection in section.get_children():
+            # Get custom due date for subsection if it is set
+            relative_weeks_due = subsection.fields['relative_weeks_due'].read_from(subsection)
+            if CUSTOM_RELATIVE_DATES.is_enabled(course.id) and relative_weeks_due:
+                section_due_date = start_date + timedelta(weeks=relative_weeks_due)
+
+            # If the section_due_date is within current date and the target date range, include it in reminder list.
+            if current_date <= section_due_date <= target_date:
+                section_date_items.append((subsection.display_name, section_due_date.strftime(DUE_DATE_FORMAT)))
+        date_items.extend(section_date_items)
+    return date_items
diff --git a/openedx/core/djangoapps/schedules/management/commands/send_course_due_date_reminders.py b/openedx/core/djangoapps/schedules/management/commands/send_course_due_date_reminders.py
new file mode 100644
index 000000000000..95698fad344a
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/management/commands/send_course_due_date_reminders.py
@@ -0,0 +1,54 @@
+"""
+Management command to send Schedule course due date reminders
+"""
+
+import datetime
+import pytz
+from textwrap import dedent  # lint-amnesty, pylint: disable=wrong-import-order
+
+from django.contrib.sites.models import Site
+
+from openedx.core.djangoapps.schedules.management.commands import SendEmailBaseCommand
+from openedx.core.djangoapps.schedules.tasks import COURSE_DUE_DATE_REMINDER_LOG_PREFIX, ScheduleCourseDueDateReminders
+
+
+class Command(SendEmailBaseCommand):
+    """
+    Command to send due date reminders for subsections in Self paced courses.
+
+    Note: this feature does not support reminders for INDIVIDUAL_DUE_DATES as the applicable schedule
+    objects are fetched based on course relative due dates.
+
+    Usage:
+        ./manage.py lms send_course_due_date_reminders localhost:18000 --due 7 --date 2023-06-07
+
+    Positional required args:
+        - site: Django site domain name, for example: localhost:18000
+    Keyword Required args
+        - due-in: Remind subsections due in given days
+    Optional args:
+        - date: The date to compute weekly messages relative to, in YYYY-MM-DD format.
+        - override-recipient-email: Send all emails to this address instead of the actual recipient
+    """
+    help = dedent(__doc__).strip()
+    async_send_task = ScheduleCourseDueDateReminders
+    log_prefix = COURSE_DUE_DATE_REMINDER_LOG_PREFIX
+
+    def add_arguments(self, parser):
+        super().add_arguments(parser)
+        parser.add_argument(
+            '--due-in',
+            type=int,
+            help='Remind subsections due in given days',
+        )
+
+    def handle(self, *args, **options):
+        current_date = datetime.datetime(
+            *[int(x) for x in options['date'].split('-')],
+            tzinfo=pytz.UTC
+        )
+
+        site = Site.objects.get(domain__iexact=options['site_domain_name'])
+        override_recipient_email = options.get('override_recipient_email')
+
+        self.async_send_task.enqueue(site, current_date, options['due_in'], override_recipient_email)
diff --git a/openedx/core/djangoapps/schedules/management/commands/send_course_next_section_update.py b/openedx/core/djangoapps/schedules/management/commands/send_course_next_section_update.py
index 53e2100649a1..4f95cc1e76bd 100644
--- a/openedx/core/djangoapps/schedules/management/commands/send_course_next_section_update.py
+++ b/openedx/core/djangoapps/schedules/management/commands/send_course_next_section_update.py
@@ -15,6 +15,9 @@
 class Command(SendEmailBaseCommand):
     """
     Command to send Schedule course updates for Self-paced Courses
+
+    Usage:
+        ./manage.py lms send_course_next_section_update localhost:18000 --date 2023-06-08
     """
     help = dedent(__doc__).strip()
     async_send_task = ScheduleCourseNextSectionUpdate
diff --git a/openedx/core/djangoapps/schedules/message_types.py b/openedx/core/djangoapps/schedules/message_types.py
index 8239a1adcdf7..0ba27c7cbcc9 100644
--- a/openedx/core/djangoapps/schedules/message_types.py
+++ b/openedx/core/djangoapps/schedules/message_types.py
@@ -29,5 +29,9 @@ class CourseUpdate(ScheduleMessageType):
     pass
 
 
+class CourseDueDatesReminder(ScheduleMessageType):
+    pass
+
+
 class InstructorLedCourseUpdate(ScheduleMessageType):
     pass
diff --git a/openedx/core/djangoapps/schedules/migrations/0021_add_due_date_reminder_schedule_config.py b/openedx/core/djangoapps/schedules/migrations/0021_add_due_date_reminder_schedule_config.py
new file mode 100644
index 000000000000..c32996bad9df
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/migrations/0021_add_due_date_reminder_schedule_config.py
@@ -0,0 +1,53 @@
+# Generated by Django 3.2.20 on 2023-09-04 11:43
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('schedules', '0020_remove_config_rollout_fields'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='scheduleconfig',
+            name='deliver_course_due_date_reminder',
+            field=models.BooleanField(default=False, help_text='Enable sending emails for due date reminder schedule.'),
+        ),
+        migrations.AddField(
+            model_name='scheduleconfig',
+            name='enqueue_course_due_date_reminder',
+            field=models.BooleanField(default=False, help_text='Enable message queuing process for due date reminder schedule.'),
+        ),
+        migrations.AlterField(
+            model_name='scheduleconfig',
+            name='deliver_course_update',
+            field=models.BooleanField(default=False, help_text='Enable sending emails for course update schedule.'),
+        ),
+        migrations.AlterField(
+            model_name='scheduleconfig',
+            name='deliver_recurring_nudge',
+            field=models.BooleanField(default=False, help_text='Enable sending emails for recurring nudge schedule.'),
+        ),
+        migrations.AlterField(
+            model_name='scheduleconfig',
+            name='deliver_upgrade_reminder',
+            field=models.BooleanField(default=False, help_text='Enable sending emails for upgrade reminder schedule.'),
+        ),
+        migrations.AlterField(
+            model_name='scheduleconfig',
+            name='enqueue_course_update',
+            field=models.BooleanField(default=False, help_text='Enable message queuing process for course update schedule.'),
+        ),
+        migrations.AlterField(
+            model_name='scheduleconfig',
+            name='enqueue_recurring_nudge',
+            field=models.BooleanField(default=False, help_text='Enable message queuing process for recurring nudge schedule.'),
+        ),
+        migrations.AlterField(
+            model_name='scheduleconfig',
+            name='enqueue_upgrade_reminder',
+            field=models.BooleanField(default=False, help_text='Enable message queuing process for upgrade reminder schedule.'),
+        ),
+    ]
diff --git a/openedx/core/djangoapps/schedules/models.py b/openedx/core/djangoapps/schedules/models.py
index 2db574cf0e6c..fe41d8aff0bb 100644
--- a/openedx/core/djangoapps/schedules/models.py
+++ b/openedx/core/djangoapps/schedules/models.py
@@ -54,12 +54,38 @@ class ScheduleConfig(ConfigurationModel):
     KEY_FIELDS = ('site',)
 
     site = models.ForeignKey(Site, on_delete=models.CASCADE)
-    enqueue_recurring_nudge = models.BooleanField(default=False)
-    deliver_recurring_nudge = models.BooleanField(default=False)
-    enqueue_upgrade_reminder = models.BooleanField(default=False)
-    deliver_upgrade_reminder = models.BooleanField(default=False)
-    enqueue_course_update = models.BooleanField(default=False)
-    deliver_course_update = models.BooleanField(default=False)
+    enqueue_recurring_nudge = models.BooleanField(
+        default=False,
+        help_text=_('Enable message queuing process for recurring nudge schedule.'),
+    )
+    deliver_recurring_nudge = models.BooleanField(
+        default=False,
+        help_text=_('Enable sending emails for recurring nudge schedule.'),
+    )
+    enqueue_upgrade_reminder = models.BooleanField(
+        default=False,
+        help_text=_('Enable message queuing process for upgrade reminder schedule.'),
+    )
+    deliver_upgrade_reminder = models.BooleanField(
+        default=False,
+        help_text=_('Enable sending emails for upgrade reminder schedule.'),
+    )
+    enqueue_course_update = models.BooleanField(
+        default=False,
+        help_text=_('Enable message queuing process for course update schedule.'),
+    )
+    deliver_course_update = models.BooleanField(
+        default=False,
+        help_text=_('Enable sending emails for course update schedule.'),
+    )
+    enqueue_course_due_date_reminder = models.BooleanField(
+        default=False,
+        help_text=_('Enable message queuing process for due date reminder schedule.'),
+    )
+    deliver_course_due_date_reminder = models.BooleanField(
+        default=False,
+        help_text=_('Enable sending emails for due date reminder schedule.'),
+    )
 
 
 class ScheduleExperience(models.Model):
diff --git a/openedx/core/djangoapps/schedules/resolvers.py b/openedx/core/djangoapps/schedules/resolvers.py
index 3498042415ff..625e67b74981 100644
--- a/openedx/core/djangoapps/schedules/resolvers.py
+++ b/openedx/core/djangoapps/schedules/resolvers.py
@@ -11,20 +11,27 @@
 from django.templatetags.static import static
 from django.db.models import Exists, F, OuterRef, Q
 from django.urls import reverse
+from opaque_keys.edx.keys import CourseKey
 from edx_ace.recipient import Recipient
 from edx_ace.recipient_resolver import RecipientResolver
 from edx_django_utils.monitoring import function_trace, set_custom_attribute
+from lms.djangoapps.courseware.courses import get_course
+from lms.djangoapps.courseware.exceptions import CourseRunNotFound
 
 from lms.djangoapps.courseware.utils import verified_upgrade_deadline_link, can_show_verified_upgrade
 from lms.djangoapps.discussion.notification_prefs.views import UsernameCipher
 from openedx.core.djangoapps.ace_common.template_context import get_base_template_context
-from openedx.core.djangoapps.course_date_signals.utils import get_expected_duration
+from openedx.core.djangoapps.course_date_signals.utils import (
+    get_expected_duration, get_expected_duration_based_on_relative_due_dates
+)
 from openedx.core.djangoapps.schedules.config import (
     COURSE_UPDATE_SHOW_UNSUBSCRIBE_WAFFLE_SWITCH, query_external_updates
 )
-from openedx.core.djangoapps.schedules.content_highlights import get_week_highlights, get_next_section_highlights
+from openedx.core.djangoapps.schedules.content_highlights import (
+    get_upcoming_subsection_due_dates, get_week_highlights, get_next_section_highlights
+)
 from openedx.core.djangoapps.schedules.exceptions import CourseUpdateDoesNotExist
-from openedx.core.djangoapps.schedules.message_types import CourseUpdate, InstructorLedCourseUpdate
+from openedx.core.djangoapps.schedules.message_types import InstructorLedCourseUpdate
 from openedx.core.djangoapps.schedules.models import Schedule, ScheduleExperience
 from openedx.core.djangoapps.schedules.utils import PrefixedDebugLoggerMixin
 from openedx.core.djangoapps.site_configuration.models import SiteConfiguration
@@ -399,7 +406,7 @@ def schedules_for_bin(self):
 
             try:
                 week_highlights = get_week_highlights(user, enrollment.course_id, week_num)
-            except CourseUpdateDoesNotExist:
+            except (CourseRunNotFound, CourseUpdateDoesNotExist):
                 LOG.warning(
                     'Weekly highlights for user {} in week {} of course {} does not exist or is disabled'.format(
                         user, week_num, enrollment.course_id
@@ -432,86 +439,89 @@ def schedules_for_bin(self):
 
 
 @attr.s
-class CourseNextSectionUpdate(PrefixedDebugLoggerMixin, RecipientResolver):
+class SelfPacedResolverBase(PrefixedDebugLoggerMixin, RecipientResolver):
     """
-    Send a message to all users whose schedule gives them a due date of yesterday.
+    Resolver base class for self paced courses.
 
-    Only used for Self-paced Courses
+    Arguments:
+        async_send_task -- celery task function that sends the message
+        site -- Site object that filtered Schedules will be a part of
+        effective_datetime -- effective date to get active courses.
+        target_datetime -- datetime that the User's Schedule's schedule_date_field value should fall under
+        course_id -- course id
+        override_recipient_email -- string email address that should receive all emails instead of the normal
+                                    recipient. (default: None)
+
+    Static attributes:
+        log_prefix -- a string to indentify this queue in logs
+        experience_filter -- a queryset filter used to select only the users who should be getting this message as part
+                             of their experience. This defaults to users without a specified experience type and those
+                             in the "recurring nudges and upgrade reminder" experience.
     """
     async_send_task = attr.ib()
     site = attr.ib()
+    effective_datetime = attr.ib()
     target_datetime = attr.ib()
     course_id = attr.ib()
     override_recipient_email = attr.ib(default=None)
+    experience_filter = None
+    log_prefix = ''
 
-    log_prefix = 'Next Section Course Update'
-    experience_filter = Q(experience__experience_type=ScheduleExperience.EXPERIENCES.course_updates)
-
-    def send(self):  # lint-amnesty, pylint: disable=arguments-differ
-        schedules = self.get_schedules()
-        for (user, language, context) in schedules:
-            msg = CourseUpdate().personalize(
-                Recipient(
-                    user.id,
-                    self.override_recipient_email or user.email,
-                ),
-                language,
-                context,
-            )
-            LOG.info(
-                'Sending email to user: {} for course-key: {}'.format(
-                    user.username,
-                    self.course_id
-                )
-            )
-            with function_trace('enqueue_send_task'):
-                self.async_send_task.apply_async((self.site.id, str(msg)), retry=False)
+    def get_expected_course_duration(self):
+        """
+        Get expected course duration.
+        """
+        return get_expected_duration(self.course_id)
 
-    def get_schedules(self):
+    def get_schedules(self, duration):
         """
-        Grabs possible schedules that could receive a Course Next Section Update and if a
-        next section highlight is applicable for the user, yields information needed to
-        send the next section highlight email.
+        Get applicable schedule objects for the effective dates and course.
         """
-        target_date = self.target_datetime.date()
-        course_duration = get_expected_duration(self.course_id)
         schedules = Schedule.objects.select_related('enrollment').filter(
-            self.experience_filter,
             enrollment__is_active=True,
             enrollment__course_id=self.course_id,
             enrollment__user__is_active=True,
-            start_date__gte=target_date - course_duration,
-            start_date__lt=target_date,
+            start_date__gte=self.effective_datetime - duration,
+            start_date__lt=self.effective_datetime,
         )
+        if self.experience_filter is not None:
+            schedules = schedules.filter(self.experience_filter)
+        return schedules
+
+    def get_line_item_context(self, user, course_id, start_date, duration):  # pylint: disable=unused-argument
+        """
+        To be overridden by child class to fetch main line items in the email like due dates or highlights.
+        """
+        raise NotImplementedError
+
+    def get_context(self):
+        """
+        Build and return context for schedule message email.
+        """
+        try:
+            course_duration = self.get_expected_course_duration()
+        except CourseRunNotFound as e:
+            log_message = self.log_prefix + ': ' + str(e)
+            LOG.warning(log_message)
+            return
 
+        schedules = self.get_schedules(course_duration)
         template_context = get_base_template_context(self.site)
         for schedule in schedules:
             course = schedule.enrollment.course
             # We don't want to show any updates if the course has ended so we short circuit here.
-            if course.end and course.end.date() <= target_date:
+            if course.end and course.end <= self.effective_datetime:
                 return
-
-            # Next Section Updates are only for Self-paced courses since it uses Personalized
-            # Learner Schedule logic. See CourseUpdateResolver for Instructor-paced updates
-            if not course.self_paced:
-                continue
-
             user = schedule.enrollment.user
             start_date = max(filter(None, (schedule.start_date, course.start)))
             LOG.info('Received a schedule for user {} in course {} for date {}'.format(
-                user.username, self.course_id, target_date,
+                user.username, self.course_id, self.target_datetime,
             ))
 
-            try:
-                week_highlights, week_num = get_next_section_highlights(user, course.id, start_date, target_date)
-                # (None, None) is returned when there is no section with a due date of the target_date
-                if week_highlights is None:
-                    continue
-            except CourseUpdateDoesNotExist as e:
-                log_message = self.log_prefix + ': ' + str(e)
-                LOG.warning(log_message)
-                # continue to the next schedule, don't yield an email for this one
+            line_item_context = self.get_line_item_context(user, course.id, start_date, course_duration)
+            if not line_item_context:
                 continue
+
             unsubscribe_url = None
             if (COURSE_UPDATE_SHOW_UNSUBSCRIBE_WAFFLE_SWITCH.is_enabled() and
                     'bulk_email_optout' in settings.ACE_ENABLED_POLICIES):
@@ -523,16 +533,100 @@ def get_schedules(self):
             template_context.update({
                 'course_name': course.display_name,
                 'course_url': _get_trackable_course_home_url(course.id),
-                'week_num': week_num,
-                'week_highlights': week_highlights,
                 # This is used by the bulk email optout policy
                 'course_ids': [str(course.id)],
                 'unsubscribe_url': unsubscribe_url,
             })
+            template_context.update(line_item_context)
             template_context.update(_get_upsell_information_for_schedule(user, schedule))
 
             yield (user, course.closest_released_language, template_context)
 
+    def send(self, msg_type):  # lint-amnesty, pylint: disable=arguments-differ
+        schedules = self.get_context()
+        for (user, language, context) in schedules:
+            msg = msg_type.personalize(
+                Recipient(
+                    user.id,
+                    self.override_recipient_email or user.email,
+                ),
+                language,
+                context,
+            )
+            LOG.info(
+                'Sending email to user: {} for course-key: {}'.format(
+                    user.username,
+                    self.course_id
+                )
+            )
+            with function_trace('enqueue_send_task'):
+                self.async_send_task.apply_async((self.site.id, str(msg)), retry=False)
+
+
+@attr.s
+class CourseNextSectionUpdate(SelfPacedResolverBase):
+    """
+    Send a message to all users whose schedule gives them a due date of yesterday.
+
+    Only used for Self-paced Courses
+    """
+    log_prefix = 'Next Section Course Update'
+    experience_filter = Q(experience__experience_type=ScheduleExperience.EXPERIENCES.course_updates)
+
+    def get_line_item_context(self, user, course_id, start_date, duration):
+        context = {}
+        try:
+            week_highlights, week_num = get_next_section_highlights(
+                user,
+                course_id,
+                start_date,
+                self.target_datetime.date(),
+                duration,
+            )
+            # (None, None) is returned when there is no section with a due date of the target_date
+            if week_highlights:
+                context = {"week_highlights": week_highlights, "week_num": week_num}
+        except (CourseRunNotFound, CourseUpdateDoesNotExist) as e:
+            log_message = self.log_prefix + ': ' + str(e)
+            LOG.warning(log_message)
+        return context
+
+
+@attr.s
+class CourseSectionDueDateReminder(SelfPacedResolverBase):
+    """
+    Send a reminder with subsection due dates with specified target date to all users.
+    """
+    log_prefix = 'Course Section Due Date'
+    experience_filter = None
+
+    def get_expected_course_duration(self):
+        """
+        Return duration based on relative due dates.
+        Note: The limitation of getting course duration based on common relative due dates is that it does
+        not consider INDIVIDUAL_DUE_DATES if set.
+        """
+        course = get_course(CourseKey.from_string(self.course_id), 2)
+        return get_expected_duration_based_on_relative_due_dates(course)
+
+    def get_line_item_context(self, user, course_id, start_date, duration):
+        context = {}
+        try:
+            date_items = get_upcoming_subsection_due_dates(
+                user,
+                course_id,
+                start_date,
+                self.target_datetime,
+                self.effective_datetime,
+                duration,
+            )
+            if date_items:
+                context["date_items"] = date_items
+        except CourseRunNotFound as e:
+            log_message = self.log_prefix + ': ' + str(e)
+            LOG.warning(log_message)
+        return context
+
 
 def _get_trackable_course_home_url(course_id):
     """
@@ -543,7 +637,7 @@ def _get_trackable_course_home_url(course_id):
 
     Args:
         course_id (CourseKey): The course to get the home page URL for.
-U
+
     Returns:
         A URL to the course home page.
     """
diff --git a/openedx/core/djangoapps/schedules/tasks.py b/openedx/core/djangoapps/schedules/tasks.py
index 55288d63f1a8..3cec11a8ea70 100644
--- a/openedx/core/djangoapps/schedules/tasks.py
+++ b/openedx/core/djangoapps/schedules/tasks.py
@@ -2,6 +2,7 @@
 
 import datetime
 import logging
+from typing import Tuple
 
 from celery import shared_task, current_app
 from celery_utils.logged_task import LoggedTask
@@ -23,6 +24,7 @@
 from opaque_keys.edx.keys import CourseKey
 
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
+from openedx.core.djangoapps.course_date_signals.models import SelfPacedRelativeDatesConfig
 from openedx.core.djangoapps.schedules import message_types, resolvers
 from openedx.core.djangoapps.schedules.models import Schedule, ScheduleConfig
 from openedx.core.lib.celery.task_utils import emulate_http_request
@@ -42,6 +44,7 @@
 UPGRADE_REMINDER_LOG_PREFIX = 'Upgrade Reminder'
 COURSE_UPDATE_LOG_PREFIX = 'Course Update'
 COURSE_NEXT_SECTION_UPDATE_LOG_PREFIX = 'Course Next Section Update'
+COURSE_DUE_DATE_REMINDER_LOG_PREFIX = 'Course Subsection Due Date Reminder'
 
 
 @shared_task(base=LoggedPersistOnFailureTask, bind=True, default_retry_delay=30)
@@ -181,6 +184,17 @@ def _course_update_schedule_send(site_id, msg_str):
     )
 
 
+@shared_task(base=LoggedTask, ignore_result=True)
+@set_code_owner_attribute
+def _course_due_date_reminder_schedule_send(site_id, msg_str):
+    _schedule_send(
+        msg_str,
+        site_id,
+        'deliver_course_due_date_reminder',
+        COURSE_DUE_DATE_REMINDER_LOG_PREFIX,
+    )
+
+
 class ScheduleRecurringNudge(BinnedScheduleMessageBaseTask):  # lint-amnesty, pylint: disable=missing-class-docstring
     num_bins = resolvers.RECURRING_NUDGE_NUM_BINS
     enqueue_config_var = 'enqueue_recurring_nudge'
@@ -223,53 +237,137 @@ def make_message_type(self, day_offset):
 ScheduleCourseUpdate = ScheduleCourseUpdate.task_instance
 
 
-class ScheduleCourseNextSectionUpdate(ScheduleMessageBaseTask):  # lint-amnesty, pylint: disable=missing-class-docstring
-    enqueue_config_var = 'enqueue_course_update'
-    log_prefix = COURSE_NEXT_SECTION_UPDATE_LOG_PREFIX
-    resolver = resolvers.CourseNextSectionUpdate
-    async_send_task = _course_update_schedule_send
+class SelfPacedCourseMessageBaseTask(ScheduleMessageBaseTask):
+    """
+    Base class for self-paced course schedule tasks that create subtasks.
+    """
+    enqueue_config_var = ''
+    log_prefix = ''
+    resolver = None  # define in subclass
+    async_send_task = None
     task_instance = None
 
     @classmethod
-    def enqueue(cls, site, current_date, day_offset, override_recipient_email=None):  # lint-amnesty, pylint: disable=missing-function-docstring
+    def should_process_course(cls, course_key, site):  # pylint: disable=unused-argument
+        """
+        Check if course should be processed for scheduled messages.
+        Override in child class if this check is required.
+        """
+        return True
+
+    @classmethod
+    def calculate_dates(cls, effective_date, day_offset) -> Tuple[datetime.datetime, datetime.datetime]:  # pylint: disable=unused-argument
+        """
+        Get target and effective/current datetime for given effective_date and day_offset.
+        """
+        raise NotImplementedError
+
+    def make_message_type(self):
+        """ Make schedule message type. """
+        raise NotImplementedError
+
+    @classmethod
+    def enqueue(cls, site, effective_date, day_offset, override_recipient_email=None):
+        """ Enqueue this subtasks for getting section due dates and sending emails. """
         set_code_owner_attribute_from_module(__name__)
-        target_datetime = (current_date - datetime.timedelta(days=day_offset))
+        target_datetime, current_datetime = cls.calculate_dates(effective_date, day_offset)
 
         if not cls.is_enqueue_enabled(site):
             cls.log_info('Message queuing disabled for site %s', site.domain)
             return
 
         cls.log_info('Target date = %s', target_datetime.date().isoformat())
-        for course_key in CourseOverview.get_all_course_keys():
-            task_args = (
-                site.id,
-                serialize(target_datetime),  # Need to leave as a datetime for serialization purposes here
-                str(course_key),  # Needs to be a string for celery to properly process
-                override_recipient_email,
-            )
-            cls.log_info('Launching task with args = %r', task_args)
-            cls.task_instance.apply_async(
-                task_args,
-                retry=False,
-            )
+        for course_key in CourseOverview.get_all_course_keys(self_paced=True):
+            if cls.should_process_course(course_key, site):
+                task_args = (
+                    site.id,
+                    serialize(current_datetime),
+                    serialize(target_datetime),
+                    str(course_key),  # Needs to be a string for celery to properly process
+                    override_recipient_email,
+                )
+                cls.log_info('Launching task with args = %r', task_args)
+                cls.task_instance.apply_async(
+                    task_args,
+                    retry=False,
+                )
 
-    def run(self, site_id, target_day_str, course_key, override_recipient_email=None):  # lint-amnesty, pylint: disable=arguments-differ
+    def run(  # lint-amnesty, pylint: disable=arguments-differ
+        self, site_id, current_date_str, target_date_str, course_key, override_recipient_email=None
+    ):
+        """ Run method for celery which calls resolver to get due dates and send emails. """
         set_code_owner_attribute_from_module(__name__)
         site = Site.objects.select_related('configuration').get(id=site_id)
         with emulate_http_request(site=site):
-            _annotate_for_monitoring(message_types.CourseUpdate(), site, 0, target_day_str, -1)
+            _annotate_for_monitoring(self.make_message_type(), site, 0, target_date_str, -1)
             return self.resolver(
                 self.async_send_task,
                 site,
-                deserialize(target_day_str),
+                deserialize(current_date_str),
+                deserialize(target_date_str),
                 str(course_key),
                 override_recipient_email,
-            ).send()
+            ).send(self.make_message_type())
+
+
+class ScheduleCourseNextSectionUpdate(SelfPacedCourseMessageBaseTask):
+    """ Scheduler class for course updates which creates sub tasks. """
+    enqueue_config_var = 'enqueue_course_update'
+    log_prefix = COURSE_NEXT_SECTION_UPDATE_LOG_PREFIX
+    resolver = resolvers.CourseNextSectionUpdate
+    async_send_task = _course_update_schedule_send
+    task_instance = None
+
+    def make_message_type(self):
+        """ CourseUpdate message type """
+        return message_types.CourseUpdate()
+
+    @classmethod
+    def calculate_dates(cls, effective_date, day_offset):
+        """
+        Target datetime should be less number day_offset as we want to get the highlights for next section.
+        """
+        target_datetime = effective_date - datetime.timedelta(days=day_offset)
+        # effective_datetime/current_datetime and target_datetime should be same for this message.
+        return target_datetime, target_datetime
 # Save the task instance on the class object so that it's accessible via the cls argument to enqueue
 ScheduleCourseNextSectionUpdate.task_instance = current_app.register_task(ScheduleCourseNextSectionUpdate())
 ScheduleCourseNextSectionUpdate = ScheduleCourseNextSectionUpdate.task_instance
 
 
+class ScheduleCourseDueDateReminders(SelfPacedCourseMessageBaseTask):
+    """ Scheduler class for course due date reminders which creates sub tasks. """
+    enqueue_config_var = 'enqueue_course_due_date_reminder'
+    log_prefix = COURSE_DUE_DATE_REMINDER_LOG_PREFIX
+    resolver = resolvers.CourseSectionDueDateReminder
+    async_send_task = _course_due_date_reminder_schedule_send
+    task_instance = None
+
+    def make_message_type(self):
+        """ CourseDueDatesReminder message type. """
+        return message_types.CourseDueDatesReminder()
+
+    @classmethod
+    def calculate_dates(cls, effective_date, day_offset):
+        """
+        Target datetime should be plus day_offset as we want to get due dates for future sections.
+        """
+        return effective_date + datetime.timedelta(days=day_offset), effective_date
+
+    @classmethod
+    def should_process_course(cls, course_key, site):
+        """
+        Only process courses with relative due dates enabled.
+        """
+        enabled_at_course_level = SelfPacedRelativeDatesConfig.current(course_key=course_key).enabled
+        if enabled_at_course_level is not None:
+            return enabled_at_course_level
+        return SelfPacedRelativeDatesConfig.current(site=site).enabled
+# Save the task instance on the class object so that it's accessible via the cls argument to enqueue
+ScheduleCourseDueDateReminders.task_instance = current_app.register_task(ScheduleCourseDueDateReminders())
+ScheduleCourseDueDateReminders = ScheduleCourseDueDateReminders.task_instance
+
+
 def _schedule_send(msg_str, site_id, delivery_config_var, log_prefix):  # lint-amnesty, pylint: disable=missing-function-docstring
     site = Site.objects.select_related('configuration').get(pk=site_id)
     if _is_delivery_enabled(site, delivery_config_var, log_prefix):
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/body.html b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/body.html
new file mode 100644
index 000000000000..6d83c58424f3
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/body.html
@@ -0,0 +1,45 @@
+{% extends 'ace_common/edx_ace/common/base_body.html' %}
+{% load i18n %}
+{% load django_markup %}
+
+{% block preview_text %}
+    {% filter force_escape %}
+    {% blocktrans trimmed %}
+        Important course due dates coming up!
+    {% endblocktrans %}
+    {% endfilter %}
+{% endblock %}
+
+{% block content %}
+<table width="100%" align="left" border="0" cellpadding="0" cellspacing="0" role="presentation">
+    <tr>
+        <td>
+            <p>
+                {% blocktrans trimmed asvar tmsg %}
+                    Hi there! We wanted to remind you of some important due dates coming up for your course, {start_strong}{course_name}{end_strong}, that you need to be aware of. The due dates are coming up in the following subsections:
+                {% endblocktrans %}
+                {% interpolate_html tmsg start_strong='<strong>'|safe end_strong='</strong>'|safe course_name=course_name|force_escape|safe %}
+                <ul>
+                    {% for subsection_name, date_item in date_items %}
+                        <li>{{ subsection_name }} ({{ date_item }})</li>
+                    {% endfor %}
+                </ul>
+            </p>
+            <p>
+                {% filter force_escape %}
+                {% blocktrans trimmed %}
+                    Happy learning!
+                {% endblocktrans %}
+                {% endfilter %}
+            </p>
+
+            {% filter force_escape %}
+                {% blocktrans asvar course_cta_text %}Resume your course now{% endblocktrans %}
+            {% endfilter %}
+            {% include "ace_common/edx_ace/common/return_to_course_cta.html" with course_cta_text=course_cta_text%}
+
+            {% include "ace_common/edx_ace/common/upsell_cta.html"%}
+        </td>
+    </tr>
+</table>
+{% endblock %}
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/body.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/body.txt
new file mode 100644
index 000000000000..03960ff43608
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/body.txt
@@ -0,0 +1,16 @@
+{% autoescape off %}
+{% load i18n %}
+
+{% blocktrans trimmed %}
+Hi there! We wanted to remind you of some important due dates coming up for your course, {{ course_name }}, that you need to be aware of. The due dates are coming up in the following subsections:
+{% endblocktrans %}
+
+{% for subsection_name, date_item in date_items %}
+  * {{ subsection_name }} ({{ date_item }})
+{% endfor %}
+
+{% blocktrans trimmed %}
+Happy learning!
+{% endblocktrans %}
+{% include "ace_common/edx_ace/common/upsell_cta.txt"%}
+{% endautoescape %}
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/from_name.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/from_name.txt
new file mode 100644
index 000000000000..fd8b9b7554d6
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/from_name.txt
@@ -0,0 +1,3 @@
+{% autoescape off %}
+{{ course_name }}
+{% endautoescape %}
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/head.html b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/head.html
new file mode 100644
index 000000000000..366ada7ad92e
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/head.html
@@ -0,0 +1 @@
+{% extends 'ace_common/edx_ace/common/base_head.html' %}
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/subject.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/subject.txt
new file mode 100644
index 000000000000..7e8f6d4fce3a
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/courseduedatesreminder/email/subject.txt
@@ -0,0 +1,5 @@
+{% autoescape off %}
+{% load i18n %}
+
+{% blocktrans trimmed %}Important course due dates coming up{% endblocktrans %}
+{% endautoescape %}
diff --git a/openedx/core/djangoapps/schedules/tests/test_content_highlights.py b/openedx/core/djangoapps/schedules/tests/test_content_highlights.py
index ece49fc50f0d..5a4271113835 100644
--- a/openedx/core/djangoapps/schedules/tests/test_content_highlights.py
+++ b/openedx/core/djangoapps/schedules/tests/test_content_highlights.py
@@ -6,16 +6,19 @@
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory, BlockFactory
 
+from common.djangoapps.student.models import CourseEnrollment
+from common.djangoapps.student.tests.factories import UserFactory
+from lms.djangoapps.courseware.exceptions import CourseRunNotFound
 from openedx.core.djangoapps.schedules.content_highlights import (
+    DUE_DATE_FORMAT,
     course_has_highlights_from_store,
     get_all_course_highlights,
     get_next_section_highlights,
+    get_upcoming_subsection_due_dates,
     get_week_highlights
 )
 from openedx.core.djangoapps.schedules.exceptions import CourseUpdateDoesNotExist
 from openedx.core.djangolib.testing.utils import skip_unless_lms
-from common.djangoapps.student.models import CourseEnrollment
-from common.djangoapps.student.tests.factories import UserFactory
 
 
 @skip_unless_lms
@@ -36,7 +39,7 @@ def _setup_user(self):
         CourseEnrollment.enroll(self.user, self.course_key)
 
     def _create_chapter(self, **kwargs):
-        BlockFactory.create(
+        return BlockFactory.create(
             parent=self.course,
             category='chapter',
             **kwargs
@@ -44,7 +47,7 @@ def _create_chapter(self, **kwargs):
 
     def test_non_existent_course_raises_exception(self):
         nonexistent_course_key = self.course_key.replace(run='no_such_run')
-        with pytest.raises(CourseUpdateDoesNotExist):
+        with pytest.raises(CourseRunNotFound):
             get_week_highlights(self.user, nonexistent_course_key, week_num=1)
 
     def test_empty_course_raises_exception(self):
@@ -168,10 +171,71 @@ def test_get_highlights_without_block(self, mock_get_block):
         with self.store.bulk_operations(self.course_key):
             self._create_chapter(highlights=['Test highlight'])
 
-        with self.assertRaisesRegex(CourseUpdateDoesNotExist, 'Course block .* not found'):
+        with self.assertRaisesRegex(CourseRunNotFound, 'Course run not found'):
             get_week_highlights(self.user, self.course_key, 1)
 
         yesterday = datetime.datetime.utcnow() - datetime.timedelta(days=1)
         today = datetime.datetime.utcnow()
-        with self.assertRaisesRegex(CourseUpdateDoesNotExist, 'Course block .* not found'):
+        with self.assertRaisesRegex(CourseRunNotFound, 'Course run not found'):
             get_next_section_highlights(self.user, self.course_key, yesterday, today.date())
+
+    @patch('openedx.core.djangoapps.course_date_signals.utils.get_expected_duration')
+    def test_get_upcoming_subsection_due_dates(self, mock_duration):
+        # All of the dates chosen here are to make things easy and clean to calculate with date offsets
+        # It only goes up to 6 days because we are using two_days_ago as our reference point
+        # so 6 + 2 = 8 days for the duration of the course
+        mock_duration.return_value = datetime.timedelta(days=8)
+        today = datetime.datetime.utcnow()
+        tomorrow = today + datetime.timedelta(days=1)
+        six_days_ago = today - datetime.timedelta(days=6)
+        two_days_ago = today - datetime.timedelta(days=2)
+        two_days = today + datetime.timedelta(days=2)
+        six_days = today + datetime.timedelta(days=6)
+        ten_days = today + datetime.timedelta(days=10)
+        for chapter_num in range(4):
+            with self.store.bulk_operations(self.course_key):
+                chapter = self._create_chapter(display_name=f"week {chapter_num}")
+                BlockFactory.create(
+                    parent_location=chapter.location,
+                    category='sequential',
+                    display_name=f"subsection_1_week_{chapter_num}"
+                )
+                BlockFactory.create(
+                    parent_location=chapter.location,
+                    category='sequential',
+                    display_name=f"subsection_2_week_{chapter_num}"
+                )
+
+        assert get_upcoming_subsection_due_dates(
+            self.user,
+            self.course_key,
+            two_days_ago,
+            two_days,
+            today,
+        ) == [
+            ('subsection_1_week_0', today.strftime(DUE_DATE_FORMAT)),
+            ('subsection_2_week_0', today.strftime(DUE_DATE_FORMAT)),
+            ('subsection_1_week_1', two_days.strftime(DUE_DATE_FORMAT)),
+            ('subsection_2_week_1', two_days.strftime(DUE_DATE_FORMAT)),
+        ]
+        # Returns [] if no due dates come within the target date. This is caused by
+        # making the mock_duration 8 days, there being only 4 chapters hence duration
+        # for each chapter is 2 days, so a target date for tomorrow will not have any due dates
+        assert not get_upcoming_subsection_due_dates(self.user, self.course_key, today, tomorrow, today)
+        # So if we set target date on the first due date, we should receive alert for it.
+        assert get_upcoming_subsection_due_dates(self.user, self.course_key, today, two_days, tomorrow) == [
+            ('subsection_1_week_0', two_days.strftime(DUE_DATE_FORMAT)),
+            ('subsection_2_week_0', two_days.strftime(DUE_DATE_FORMAT))
+        ]
+        assert get_upcoming_subsection_due_dates(
+            self.user,
+            self.course_key,
+            two_days_ago,
+            six_days,
+            six_days,
+        ) == [
+            ('subsection_1_week_3', six_days.strftime(DUE_DATE_FORMAT)),
+            ('subsection_2_week_3', six_days.strftime(DUE_DATE_FORMAT))
+        ]
+        # Returns [] if we are past all due dates
+        assert not get_upcoming_subsection_due_dates(self.user, self.course_key, six_days_ago, ten_days, six_days)
diff --git a/openedx/core/djangoapps/schedules/tests/test_resolvers.py b/openedx/core/djangoapps/schedules/tests/test_resolvers.py
index 36bc8beacd32..848a5e447bcd 100644
--- a/openedx/core/djangoapps/schedules/tests/test_resolvers.py
+++ b/openedx/core/djangoapps/schedules/tests/test_resolvers.py
@@ -8,11 +8,12 @@
 
 import crum
 import ddt
+import pytest
 import pytz
 from django.test import TestCase
 from django.test.client import RequestFactory
 from django.test.utils import override_settings
-from edx_toggles.toggles.testutils import override_waffle_switch
+from edx_toggles.toggles.testutils import override_waffle_flag, override_waffle_switch
 from testfixtures import LogCapture
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory, BlockFactory
@@ -21,15 +22,19 @@
 from common.djangoapps.student.tests.factories import CourseEnrollmentFactory, UserFactory
 from lms.djangoapps.experiments.testutils import override_experiment_waffle_flag
 from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
+from openedx.core.djangoapps.course_date_signals.models import SelfPacedRelativeDatesConfig
+from openedx.core.djangoapps.course_date_signals.waffle import CUSTOM_RELATIVE_DATES
 from openedx.core.djangoapps.schedules.config import (
     _EXTERNAL_COURSE_UPDATES_FLAG,
     COURSE_UPDATE_SHOW_UNSUBSCRIBE_WAFFLE_SWITCH,
 )
+from openedx.core.djangoapps.schedules.content_highlights import DUE_DATE_FORMAT
 from openedx.core.djangoapps.schedules.models import Schedule
 from openedx.core.djangoapps.schedules.resolvers import (
     LOG,
     BinnedSchedulesBaseResolver,
     CourseNextSectionUpdate,
+    CourseSectionDueDateReminder,
     CourseUpdateResolver,
 )
 from openedx.core.djangoapps.schedules.tests.factories import ScheduleConfigFactory
@@ -235,7 +240,7 @@ class TestCourseNextSectionUpdateResolver(SchedulesResolverTestMixin, ModuleStor
 
     def setUp(self):
         super().setUp()
-        self.today = datetime.datetime.utcnow()
+        self.today = datetime.datetime.now(pytz.UTC)
         self.yesterday = self.today - datetime.timedelta(days=1)
         self.course = CourseFactory.create(
             highlights_enabled_for_messaging=True, self_paced=True,
@@ -265,6 +270,7 @@ def create_resolver(self, user_start_date_offset=8):
         return CourseNextSectionUpdate(
             async_send_task=Mock(name='async_send_task'),
             site=self.site_config.site,
+            effective_datetime=self.yesterday,
             target_datetime=self.yesterday,
             course_id=self.course.id,
         )
@@ -274,8 +280,8 @@ def create_resolver(self, user_start_date_offset=8):
     def test_schedule_context(self):
         resolver = self.create_resolver()
         # using this to make sure the select_related stays intact
-        with self.assertNumQueries(38):
-            sc = resolver.get_schedules()
+        with self.assertNumQueries(30):
+            sc = resolver.get_context()
             schedules = list(sc)
         apple_logo_url = 'http://email-media.s3.amazonaws.com/edX/2021/store_apple_229x78.jpg'
         google_logo_url = 'http://email-media.s3.amazonaws.com/edX/2021/store_google_253x78.jpg'
@@ -323,13 +329,13 @@ def test_schedule_context(self):
     @override_waffle_switch(COURSE_UPDATE_SHOW_UNSUBSCRIBE_WAFFLE_SWITCH, True)
     def test_schedule_context_show_unsubscribe(self):
         resolver = self.create_resolver()
-        schedules = list(resolver.get_schedules())
+        schedules = list(resolver.get_context())
         assert 'optout' in schedules[0][2]['unsubscribe_url']
 
     def test_schedule_context_error(self):
         resolver = self.create_resolver(user_start_date_offset=29)
         with LogCapture(LOG.name) as log_capture:
-            list(resolver.get_schedules())
+            list(resolver.get_context())
             log_message = ('Next Section Course Update: Last section was reached. '
                            'There are no more highlights for {}'.format(self.course.id))
             log_capture.check_present((LOG.name, 'WARNING', log_message))
@@ -338,5 +344,155 @@ def test_no_updates_if_course_ended(self):
         self.course.end = self.yesterday
         self.course = self.update_course(self.course, self.user.id)
         resolver = self.create_resolver()
-        schedules = list(resolver.get_schedules())
+        schedules = list(resolver.get_context())
         self.assertListEqual(schedules, [])
+
+
+@skip_unless_lms
+class TestCourseDueDateReminders(SchedulesResolverTestMixin, ModuleStoreTestCase):
+    """
+    Tests ScheduleCourseDueDateReminders resolver.
+    """
+    ENABLED_SIGNALS = ['course_published']
+
+    def setUp(self):
+        super().setUp()
+        self.today = datetime.datetime.now(pytz.UTC)
+        self.target_datetime = self.today + datetime.timedelta(days=7)
+        SelfPacedRelativeDatesConfig.objects.create(enabled=True)
+        self.course = CourseFactory.create(
+            self_paced=True,
+            # putting it in the past so the schedule can be later than the start
+            start=self.today - datetime.timedelta(days=30)
+        )
+
+        for chapter_num in range(4):
+            with self.store.bulk_operations(self.course.id):
+                chapter = BlockFactory.create(
+                    parent=self.course,
+                    category='chapter',
+                    display_name=f"section_{chapter_num}"
+                )
+                BlockFactory.create(
+                    parent_location=chapter.location,
+                    category='sequential',
+                    display_name=f"sub_section_{chapter_num}1",
+                )
+                BlockFactory.create(
+                    parent_location=chapter.location,
+                    category='sequential',
+                    display_name=f"sub_section_{chapter_num}2",
+                    relative_weeks_due=4 if chapter_num == 3 else None
+                )
+
+    def create_resolver(self, user_start_date_offset=8):
+        """
+        Creates a ScheduleCourseDueDateReminders with an enrollment to schedule.
+        """
+        CourseEnrollmentFactory(course_id=self.course.id, user=self.user, mode='audit')
+
+        # Need to update the user's schedule so the due date for the chapter we want
+        # matches with the user's schedule and the target date. The numbers are based on the
+        # course having the default course duration of 28 days.
+        user_schedule = Schedule.objects.first()
+        user_schedule.start_date = self.today - datetime.timedelta(days=user_start_date_offset)
+        user_schedule.save()
+
+        return CourseSectionDueDateReminder(
+            async_send_task=Mock(name='async_send_task'),
+            site=self.site_config.site,
+            effective_datetime=self.today,
+            target_datetime=self.target_datetime,
+            course_id=str(self.course.id),
+        )
+
+    @override_settings(CONTACT_MAILING_ADDRESS='123 Sesame Street')
+    @override_settings(LOGO_URL_PNG='https://www.logo.png')
+    @override_waffle_flag(CUSTOM_RELATIVE_DATES, active=True)
+    def test_reminder_schedule_context(self):
+        resolver = self.create_resolver(7)
+        # using this to make sure the select_related stays intact
+        sc = resolver.get_context()
+        schedules = list(sc)
+        apple_logo_url = 'http://email-media.s3.amazonaws.com/edX/2021/store_apple_229x78.jpg'
+        google_logo_url = 'http://email-media.s3.amazonaws.com/edX/2021/store_google_253x78.jpg'
+        apple_store_url = 'https://itunes.apple.com/us/app/edx/id945480667?mt=8'
+        google_store_url = 'https://play.google.com/store/apps/details?id=org.edx.mobile'
+        facebook_url = 'http://www.facebook.com/EdxOnline'
+        linkedin_url = 'http://www.linkedin.com/company/edx'
+        twitter_url = 'https://twitter.com/edXOnline'
+        reddit_url = 'http://www.reddit.com/r/edx'
+        facebook_logo_url = 'http://email-media.s3.amazonaws.com/edX/2021/social_1_fb.png'
+        linkedin_logo_url = 'http://email-media.s3.amazonaws.com/edX/2021/social_3_linkedin.png'
+        twitter_logo_url = 'http://email-media.s3.amazonaws.com/edX/2021/social_2_twitter.png'
+        reddit_logo_url = 'http://email-media.s3.amazonaws.com/edX/2021/social_5_reddit.png'
+        due_date = (self.today + datetime.timedelta(weeks=1)).strftime(DUE_DATE_FORMAT)
+        expected_context = {
+            'contact_email': 'info@example.com',
+            'contact_mailing_address': '123 Sesame Street',
+            'course_ids': [str(self.course.id)],
+            'course_name': self.course.display_name,
+            'course_url': f'http://learning-mfe/course/{self.course.id}/home',
+            'dashboard_url': '/dashboard',
+            'homepage_url': '/',
+            'mobile_store_logo_urls': {'apple': apple_logo_url,
+                                       'google': google_logo_url},
+            'mobile_store_urls': {'apple': apple_store_url,
+                                  'google': google_store_url},
+            'logo_url': 'https://www.logo.png',
+            'platform_name': '\xe9dX',
+            'show_upsell': False,
+            'site_configuration_values': {},
+            'social_media_logo_urls': {'facebook': facebook_logo_url,
+                                       'linkedin': linkedin_logo_url,
+                                       'reddit': reddit_logo_url,
+                                       'twitter': twitter_logo_url},
+            'social_media_urls': {'facebook': facebook_url,
+                                  'linkedin': linkedin_url,
+                                  'reddit': reddit_url,
+                                  'twitter': twitter_url},
+            'template_revision': 'release',
+            'unsubscribe_url': None,
+            'date_items': [
+                ('sub_section_01', self.today.strftime(DUE_DATE_FORMAT)),
+                ('sub_section_02', self.today.strftime(DUE_DATE_FORMAT)),
+                ('sub_section_11', due_date),
+                ('sub_section_12', due_date)
+            ]
+        }
+        assert schedules == [(self.user, None, expected_context)]
+
+    def test_reminder_no_due_dates_if_course_ended(self):
+        self.course.end = self.today - datetime.timedelta(days=1)
+        self.course = self.update_course(self.course, self.user.id)
+        resolver = self.create_resolver()
+        schedules = list(resolver.get_context())
+        self.assertListEqual(schedules, [])
+
+    def test_course_not_found_error(self):
+        with pytest.raises(StopIteration) as cm:
+            resolver = CourseSectionDueDateReminder(
+                async_send_task=Mock(name='async_send_task'),
+                site=self.site_config.site,
+                effective_datetime=self.today,
+                target_datetime=self.target_datetime,
+                course_id="course-v1:edX+DemoX+Demo_Course",
+            )
+            next(resolver.get_context())
+
+    @override_waffle_flag(CUSTOM_RELATIVE_DATES, active=True)
+    def test_reminder_course_duration_based_on_relative_dates(self):
+        with self.store.bulk_operations(self.course.id):
+            chapter = BlockFactory.create(
+                parent=self.course,
+                category='chapter',
+                display_name="section_4"
+            )
+            BlockFactory.create(
+                parent_location=chapter.location,
+                category='sequential',
+                display_name="sub_section_41",
+                relative_weeks_due=5,
+            )
+        resolver = self.create_resolver()
+        self.assertEqual(resolver.get_expected_course_duration(), datetime.timedelta(weeks=5))