From 6a36eb0183d6ad52fb55b5fc9f6c3a2916ee9dd9 Mon Sep 17 00:00:00 2001
From: Gabe Mulley <gabe@edx.org>
Date: Sun, 30 Jul 2017 18:04:33 -0400
Subject: [PATCH] Use ACE to 'send' Recurring Nudge emails

---
 cms/envs/common.py                            |   3 +
 common/djangoapps/student/models.py           |   9 +-
 common/djangoapps/student/tests/factories.py  |   4 +-
 .../djangoapps/student/tests/test_models.py   |   2 +
 lms/djangoapps/bulk_email/policies.py         |  20 ++
 .../bulk_email/tests/test_course_optout.py    |  81 +++++-
 .../tests/test_field_override_performance.py  |  54 ++--
 .../migrations/0003_auto_20170825_0935.py     |  19 ++
 lms/djangoapps/courseware/models.py           |   2 +-
 .../courseware/tests/test_date_summary.py     | 234 +++++++++++-------
 lms/djangoapps/courseware/tests/test_views.py |   6 +-
 lms/envs/aws.py                               |   9 +
 lms/envs/common.py                            |  17 +-
 .../course_overviews/tests/__init__.py        |   0
 .../course_overviews/tests/factories.py       |  30 +++
 .../test_course_overviews.py}                 |   2 +-
 openedx/core/djangoapps/schedules/admin.py    |   6 +
 openedx/core/djangoapps/schedules/apps.py     |   2 +-
 .../schedules/management/__init__.py          |   0
 .../schedules/management/commands/__init__.py |   0
 .../commands/send_recurring_nudge.py          |  80 ++++++
 ...send_verified_upgrade_deadline_reminder.py | 118 +++++++++
 .../management/commands/tests/__init__.py     |   0
 .../tests/test_send_recurring_nudge.py        | 169 +++++++++++++
 .../migrations/0003_scheduleconfig.py         |  35 +++
 openedx/core/djangoapps/schedules/models.py   |  14 ++
 openedx/core/djangoapps/schedules/signals.py  | 122 +++++----
 openedx/core/djangoapps/schedules/tasks.py    |  94 +++++++
 .../schedules/edx_ace/common/base_body.html   | 179 ++++++++++++++
 .../schedules/edx_ace/common/base_head.html   |  29 +++
 .../recurringnudge_day10/email/body.html      |  40 +++
 .../recurringnudge_day10/email/body.txt       |   4 +
 .../recurringnudge_day10/email/from_name.txt  |   1 +
 .../recurringnudge_day10/email/head.html      |   1 +
 .../recurringnudge_day10/email/subject.txt    |   2 +
 .../recurringnudge_day3/email/body.html       |  37 +++
 .../recurringnudge_day3/email/body.txt        |   4 +
 .../recurringnudge_day3/email/from_name.txt   |   1 +
 .../recurringnudge_day3/email/head.html       |   1 +
 .../recurringnudge_day3/email/subject.txt     |   3 +
 .../email/body.html                           |   7 +
 .../email/body.txt                            |   6 +
 .../email/subject.txt                         |   1 +
 .../djangoapps/schedules/tests/factories.py   |  13 +
 .../schedules/tests/test_signals.py           | 114 +++++++--
 .../tests/views/test_course_home.py           |   2 +-
 .../tests/views/test_course_updates.py        |   2 +-
 requirements/edx/github.txt                   |   1 +
 setup.py                                      |   3 +
 49 files changed, 1383 insertions(+), 200 deletions(-)
 create mode 100644 lms/djangoapps/bulk_email/policies.py
 create mode 100644 lms/djangoapps/courseware/migrations/0003_auto_20170825_0935.py
 create mode 100644 openedx/core/djangoapps/content/course_overviews/tests/__init__.py
 create mode 100644 openedx/core/djangoapps/content/course_overviews/tests/factories.py
 rename openedx/core/djangoapps/content/course_overviews/{tests.py => tests/test_course_overviews.py} (99%)
 create mode 100644 openedx/core/djangoapps/schedules/management/__init__.py
 create mode 100644 openedx/core/djangoapps/schedules/management/commands/__init__.py
 create mode 100644 openedx/core/djangoapps/schedules/management/commands/send_recurring_nudge.py
 create mode 100644 openedx/core/djangoapps/schedules/management/commands/send_verified_upgrade_deadline_reminder.py
 create mode 100644 openedx/core/djangoapps/schedules/management/commands/tests/__init__.py
 create mode 100644 openedx/core/djangoapps/schedules/management/commands/tests/test_send_recurring_nudge.py
 create mode 100644 openedx/core/djangoapps/schedules/migrations/0003_scheduleconfig.py
 create mode 100644 openedx/core/djangoapps/schedules/tasks.py
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/common/base_body.html
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/common/base_head.html
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/body.html
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/body.txt
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/from_name.txt
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/head.html
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/subject.txt
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/body.html
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/body.txt
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/from_name.txt
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/head.html
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/subject.txt
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/body.html
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/body.txt
 create mode 100644 openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/subject.txt

diff --git a/cms/envs/common.py b/cms/envs/common.py
index 685184c283..f036200ac3 100644
--- a/cms/envs/common.py
+++ b/cms/envs/common.py
@@ -1067,6 +1067,9 @@ INSTALLED_APPS = [
     # Waffle related utilities
     'openedx.core.djangoapps.waffle_utils',
 
+    # Dynamic schedules
+    'openedx.core.djangoapps.schedules.apps.SchedulesConfig',
+
     # DRF filters
     'django_filters',
 ]
diff --git a/common/djangoapps/student/models.py b/common/djangoapps/student/models.py
index dd5372d36f..aae2a3cc8c 100644
--- a/common/djangoapps/student/models.py
+++ b/common/djangoapps/student/models.py
@@ -50,9 +50,12 @@ import lms.lib.comment_client as cc
 import request_cache
 from certificates.models import GeneratedCertificate
 from course_modes.models import CourseMode
+from courseware.models import DynamicUpgradeDeadlineConfiguration, CourseDynamicUpgradeDeadlineConfiguration
 from enrollment.api import _default_course_mode
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
+from openedx.core.djangoapps.schedules.models import ScheduleConfig
 from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
+from openedx.core.djangoapps.theming.helpers import get_current_site
 from openedx.core.djangoapps.xmodule_django.models import CourseKeyField, NoneToEmptyManager
 from track import contexts
 from util.milestones_helpers import is_entrance_exams_enabled
@@ -1715,7 +1718,11 @@ class CourseEnrollment(models.Model):
             return None
 
         try:
-            if self.schedule:
+            schedule_driven_deadlines_enabled = (
+                DynamicUpgradeDeadlineConfiguration.is_enabled()
+                or CourseDynamicUpgradeDeadlineConfiguration.is_enabled(self.course_id)
+            )
+            if schedule_driven_deadlines_enabled and self.schedule and self.schedule.upgrade_deadline is not None:
                 log.debug(
                     'Schedules: Pulling upgrade deadline for CourseEnrollment %d from Schedule %d.',
                     self.id, self.schedule.id
diff --git a/common/djangoapps/student/tests/factories.py b/common/djangoapps/student/tests/factories.py
index 313098d5c4..7099da2976 100644
--- a/common/djangoapps/student/tests/factories.py
+++ b/common/djangoapps/student/tests/factories.py
@@ -126,7 +126,9 @@ class CourseEnrollmentFactory(DjangoModelFactory):
         model = CourseEnrollment
 
     user = factory.SubFactory(UserFactory)
-    course_id = CourseKey.from_string('edX/toy/2012_Fall')
+    course = factory.SubFactory(
+        'openedx.core.djangoapps.content.course_overviews.tests.factories.CourseOverviewFactory',
+    )
 
 
 class CourseAccessRoleFactory(DjangoModelFactory):
diff --git a/common/djangoapps/student/tests/test_models.py b/common/djangoapps/student/tests/test_models.py
index 927fa9117a..9b32f9f4c1 100644
--- a/common/djangoapps/student/tests/test_models.py
+++ b/common/djangoapps/student/tests/test_models.py
@@ -13,6 +13,7 @@ from django.db.models.functions import Lower
 
 from course_modes.models import CourseMode
 from course_modes.tests.factories import CourseModeFactory
+from courseware.models import DynamicUpgradeDeadlineConfiguration
 from openedx.core.djangoapps.schedules.models import Schedule
 from openedx.core.djangoapps.schedules.tests.factories import ScheduleFactory
 from openedx.core.djangolib.testing.utils import skip_unless_lms
@@ -131,6 +132,7 @@ class CourseEnrollmentTests(SharedModuleStoreTestCase):
         self.assertEqual(enrollment.upgrade_deadline, course_mode.expiration_datetime)
 
         # The schedule's upgrade deadline should be used if a schedule exists
+        DynamicUpgradeDeadlineConfiguration.objects.create(enabled=True)
         schedule = ScheduleFactory(enrollment=enrollment)
         self.assertEqual(enrollment.upgrade_deadline, schedule.upgrade_deadline)
 
diff --git a/lms/djangoapps/bulk_email/policies.py b/lms/djangoapps/bulk_email/policies.py
new file mode 100644
index 0000000000..a21f9b0068
--- /dev/null
+++ b/lms/djangoapps/bulk_email/policies.py
@@ -0,0 +1,20 @@
+
+from edx_ace.policy import Policy, PolicyResult
+from edx_ace.channel import ChannelType
+from opaque_keys.edx.keys import CourseKey
+
+from bulk_email.models import Optout
+
+
+class CourseEmailOptout(Policy):
+
+    def check(self, message):
+        course_id = message.context.get('course_id')
+        if not course_id:
+            return PolicyResult(deny=frozenset())
+
+        course_key = CourseKey.from_string(course_id)
+        if Optout.objects.filter(user__username=message.recipient.username, course_id=course_key).exists():
+            return PolicyResult(deny={ChannelType.EMAIL})
+
+        return PolicyResult(deny=frozenset())
diff --git a/lms/djangoapps/bulk_email/tests/test_course_optout.py b/lms/djangoapps/bulk_email/tests/test_course_optout.py
index ae235c0cf5..31d0e7b75e 100644
--- a/lms/djangoapps/bulk_email/tests/test_course_optout.py
+++ b/lms/djangoapps/bulk_email/tests/test_course_optout.py
@@ -11,6 +11,11 @@ from mock import Mock, patch
 from nose.plugins.attrib import attr
 
 from bulk_email.models import BulkEmailFlag
+from bulk_email.policies import CourseEmailOptout
+from edx_ace.message import Message
+from edx_ace.recipient import Recipient
+from edx_ace.policy import PolicyResult
+from edx_ace.channel import ChannelType
 from student.models import CourseEnrollment
 from student.tests.factories import AdminFactory, CourseEnrollmentFactory, UserFactory
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
@@ -27,7 +32,7 @@ class TestOptoutCourseEmails(ModuleStoreTestCase):
     def setUp(self):
         super(TestOptoutCourseEmails, self).setUp()
         course_title = u"ẗëṡẗ title ｲ乇丂ｲ ﾶ乇丂丂ﾑg乇 ｷo尺 ﾑﾚﾚ тэѕт мэѕѕаБэ"
-        self.course = CourseFactory.create(display_name=course_title)
+        self.course = CourseFactory.create(run='testcourse1', display_name=course_title)
         self.instructor = AdminFactory.create()
         self.student = UserFactory.create()
         CourseEnrollmentFactory.create(user=self.student, course_id=self.course.id)
@@ -44,10 +49,6 @@ class TestOptoutCourseEmails(ModuleStoreTestCase):
         }
         BulkEmailFlag.objects.create(enabled=True, require_course_email_auth=False)
 
-    def tearDown(self):
-        super(TestOptoutCourseEmails, self).tearDown()
-        BulkEmailFlag.objects.all().delete()
-
     def navigate_to_email_view(self):
         """Navigate to the instructor dash's email view"""
         # Pull up email view on instructor dashboard
@@ -114,3 +115,73 @@ class TestOptoutCourseEmails(ModuleStoreTestCase):
         sent_addresses = [message.to[0] for message in mail.outbox]
         self.assertIn(self.student.email, sent_addresses)
         self.assertIn(self.instructor.email, sent_addresses)
+
+
+@attr(shard=1)
+@patch('bulk_email.models.html_to_text', Mock(return_value='Mocking CourseEmail.text_message', autospec=True))
+class TestACEOptoutCourseEmails(ModuleStoreTestCase):
+    """
+    Test that optouts are referenced in sending course email.
+    """
+    def setUp(self):
+        super(TestACEOptoutCourseEmails, self).setUp()
+        course_title = u"ẗëṡẗ title ｲ乇丂ｲ ﾶ乇丂丂ﾑg乇 ｷo尺 ﾑﾚﾚ тэѕт мэѕѕаБэ"
+        self.course = CourseFactory.create(run='testcourse1', display_name=course_title)
+        self.instructor = AdminFactory.create()
+        self.student = UserFactory.create()
+        CourseEnrollmentFactory.create(user=self.student, course_id=self.course.id)
+
+        self.client.login(username=self.student.username, password="test")
+
+        self._set_email_optout(False)
+        self.policy = CourseEmailOptout()
+
+    def _set_email_optout(self, opted_out):
+        url = reverse('change_email_settings')
+        # This is a checkbox, so on the post of opting out (that is, an Un-check of the box),
+        # the Post that is sent will not contain 'receive_emails'
+        post_data = {'course_id': self.course.id.to_deprecated_string()}
+
+        if not opted_out:
+            post_data['receive_emails'] = 'on'
+
+        response = self.client.post(url, post_data)
+        self.assertEquals(json.loads(response.content), {'success': True})
+
+    def test_policy_optedout(self):
+        """
+        Make sure the policy prevents ACE emails if the user is opted-out.
+        """
+        self._set_email_optout(True)
+
+        channel_mods = self.policy.check(self.create_test_message())
+        self.assertEqual(channel_mods, PolicyResult(deny={ChannelType.EMAIL}))
+
+    def create_test_message(self):
+        return Message(
+            app_label='foo',
+            name='bar',
+            recipient=Recipient(
+                username=self.student.username,
+                email_address=self.student.email,
+            ),
+            context={
+                'course_id': str(self.course.id)
+            },
+        )
+
+    def test_policy_optedin(self):
+        """
+        Make sure the policy allows ACE emails if the user is opted-in.
+        """
+        channel_mods = self.policy.check(self.create_test_message())
+        self.assertEqual(channel_mods, PolicyResult(deny=set()))
+
+    def test_policy_no_course_id(self):
+        """
+        Make sure the policy denies ACE emails if there is no course id in the context.
+        """
+        message = self.create_test_message()
+        message.context = {}
+        channel_mods = self.policy.check(message)
+        self.assertEqual(channel_mods, PolicyResult(deny=set()))
diff --git a/lms/djangoapps/ccx/tests/test_field_override_performance.py b/lms/djangoapps/ccx/tests/test_field_override_performance.py
index 2e68180ab0..49aa217082 100644
--- a/lms/djangoapps/ccx/tests/test_field_override_performance.py
+++ b/lms/djangoapps/ccx/tests/test_field_override_performance.py
@@ -237,18 +237,18 @@ class TestFieldOverrideMongoPerformance(FieldOverridePerformanceTestCase):
         #     # of sql queries to default,
         #     # of mongo queries,
         # )
-        ('no_overrides', 1, True, False): (25, 1),
-        ('no_overrides', 2, True, False): (25, 1),
-        ('no_overrides', 3, True, False): (25, 1),
-        ('ccx', 1, True, False): (25, 1),
-        ('ccx', 2, True, False): (25, 1),
-        ('ccx', 3, True, False): (25, 1),
-        ('no_overrides', 1, False, False): (25, 1),
-        ('no_overrides', 2, False, False): (25, 1),
-        ('no_overrides', 3, False, False): (25, 1),
-        ('ccx', 1, False, False): (25, 1),
-        ('ccx', 2, False, False): (25, 1),
-        ('ccx', 3, False, False): (25, 1),
+        ('no_overrides', 1, True, False): (26, 1),
+        ('no_overrides', 2, True, False): (26, 1),
+        ('no_overrides', 3, True, False): (26, 1),
+        ('ccx', 1, True, False): (26, 1),
+        ('ccx', 2, True, False): (26, 1),
+        ('ccx', 3, True, False): (26, 1),
+        ('no_overrides', 1, False, False): (26, 1),
+        ('no_overrides', 2, False, False): (26, 1),
+        ('no_overrides', 3, False, False): (26, 1),
+        ('ccx', 1, False, False): (26, 1),
+        ('ccx', 2, False, False): (26, 1),
+        ('ccx', 3, False, False): (26, 1),
     }
 
 
@@ -260,19 +260,19 @@ class TestFieldOverrideSplitPerformance(FieldOverridePerformanceTestCase):
     __test__ = True
 
     TEST_DATA = {
-        ('no_overrides', 1, True, False): (25, 3),
-        ('no_overrides', 2, True, False): (25, 3),
-        ('no_overrides', 3, True, False): (25, 3),
-        ('ccx', 1, True, False): (25, 3),
-        ('ccx', 2, True, False): (25, 3),
-        ('ccx', 3, True, False): (25, 3),
-        ('ccx', 1, True, True): (26, 3),
-        ('ccx', 2, True, True): (26, 3),
-        ('ccx', 3, True, True): (26, 3),
-        ('no_overrides', 1, False, False): (25, 3),
-        ('no_overrides', 2, False, False): (25, 3),
-        ('no_overrides', 3, False, False): (25, 3),
-        ('ccx', 1, False, False): (25, 3),
-        ('ccx', 2, False, False): (25, 3),
-        ('ccx', 3, False, False): (25, 3),
+        ('no_overrides', 1, True, False): (26, 3),
+        ('no_overrides', 2, True, False): (26, 3),
+        ('no_overrides', 3, True, False): (26, 3),
+        ('ccx', 1, True, False): (26, 3),
+        ('ccx', 2, True, False): (26, 3),
+        ('ccx', 3, True, False): (26, 3),
+        ('ccx', 1, True, True): (27, 3),
+        ('ccx', 2, True, True): (27, 3),
+        ('ccx', 3, True, True): (27, 3),
+        ('no_overrides', 1, False, False): (26, 3),
+        ('no_overrides', 2, False, False): (26, 3),
+        ('no_overrides', 3, False, False): (26, 3),
+        ('ccx', 1, False, False): (26, 3),
+        ('ccx', 2, False, False): (26, 3),
+        ('ccx', 3, False, False): (26, 3),
     }
diff --git a/lms/djangoapps/courseware/migrations/0003_auto_20170825_0935.py b/lms/djangoapps/courseware/migrations/0003_auto_20170825_0935.py
new file mode 100644
index 0000000000..d8f367e242
--- /dev/null
+++ b/lms/djangoapps/courseware/migrations/0003_auto_20170825_0935.py
@@ -0,0 +1,19 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('courseware', '0002_coursedynamicupgradedeadlineconfiguration_dynamicupgradedeadlineconfiguration'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='coursedynamicupgradedeadlineconfiguration',
+            name='opt_out',
+            field=models.BooleanField(default=False, help_text='This does not do anything and is no longer used. Setting enabled=False has the same effect.'),
+        ),
+    ]
diff --git a/lms/djangoapps/courseware/models.py b/lms/djangoapps/courseware/models.py
index 00b58c82d8..c69e22e3eb 100644
--- a/lms/djangoapps/courseware/models.py
+++ b/lms/djangoapps/courseware/models.py
@@ -398,5 +398,5 @@ class CourseDynamicUpgradeDeadlineConfiguration(ConfigurationModel):
     )
     opt_out = models.BooleanField(
         default=False,
-        help_text=_('Disable the dynamic upgrade deadline for this course run.')
+        help_text=_('This does not do anything and is no longer used. Setting enabled=False has the same effect.')
     )
diff --git a/lms/djangoapps/courseware/tests/test_date_summary.py b/lms/djangoapps/courseware/tests/test_date_summary.py
index 449db0e5ff..4ac9228b5a 100644
--- a/lms/djangoapps/courseware/tests/test_date_summary.py
+++ b/lms/djangoapps/courseware/tests/test_date_summary.py
@@ -6,6 +6,7 @@ import ddt
 import waffle
 from django.core.urlresolvers import reverse
 from freezegun import freeze_time
+from mock import patch
 from nose.plugins.attrib import attr
 from pytz import utc
 
@@ -25,7 +26,9 @@ from courseware.models import DynamicUpgradeDeadlineConfiguration, CourseDynamic
 from lms.djangoapps.verify_student.models import VerificationDeadline
 from lms.djangoapps.verify_student.tests.factories import SoftwareSecurePhotoVerificationFactory
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
+from openedx.core.djangoapps.schedules.signals import SCHEDULE_WAFFLE_FLAG
 from openedx.core.djangoapps.self_paced.models import SelfPacedConfiguration
+from openedx.core.djangoapps.site_configuration.tests.factories import SiteFactory
 from openedx.core.djangoapps.user_api.preferences.api import set_user_preference
 from openedx.core.djangoapps.waffle_utils.testutils import override_waffle_flag
 from openedx.features.course_experience import UNIFIED_COURSE_TAB_FLAG
@@ -36,7 +39,6 @@ from xmodule.modulestore.tests.factories import CourseFactory
 
 @attr(shard=1)
 @ddt.ddt
-@waffle.testutils.override_switch('schedules.enable-create-schedule-receiver', True)
 class CourseDateSummaryTest(SharedModuleStoreTestCase):
     """Tests for course date summary blocks."""
 
@@ -44,43 +46,6 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
         super(CourseDateSummaryTest, self).setUp()
         SelfPacedConfiguration.objects.create(enable_course_home_improvements=True)
 
-    def create_course_run(self, days_till_start=1, days_till_end=14, days_till_upgrade_deadline=4,
-                          days_till_verification_deadline=14):
-        """ Create a new course run and course modes.
-
-        All date-related arguments are relative to the current date-time (now) unless otherwise specified.
-
-        Both audit and verified `CourseMode` objects will be created for the course run.
-
-        Arguments:
-            days_till_end (int): Number of days until the course ends.
-            days_till_start (int): Number of days until the course starts.
-            days_till_upgrade_deadline (int): Number of days until the course run's upgrade deadline.
-            days_till_verification_deadline (int): Number of days until the course run's verification deadline. If this
-                value is set to `None` no deadline will be verification deadline will be created.
-        """
-        now = datetime.now(utc)
-        course = CourseFactory.create(start=now + timedelta(days=days_till_start))
-
-        course.end = None
-        if days_till_end is not None:
-            course.end = now + timedelta(days=days_till_end)
-
-        CourseModeFactory(course_id=course.id, mode_slug=CourseMode.AUDIT)
-        CourseModeFactory(
-            course_id=course.id,
-            mode_slug=CourseMode.VERIFIED,
-            expiration_datetime=now + timedelta(days=days_till_upgrade_deadline)
-        )
-
-        if days_till_verification_deadline is not None:
-            VerificationDeadline.objects.create(
-                course_key=course.id,
-                deadline=now + timedelta(days=days_till_verification_deadline)
-            )
-
-        return course
-
     def create_user(self, verification_status=None):
         """ Create a new User instance.
 
@@ -97,7 +62,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
 
     def test_course_info_feature_flag(self):
         SelfPacedConfiguration(enable_course_home_improvements=False).save()
-        course = self.create_course_run()
+        course = create_course_run()
         user = self.create_user()
         CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.VERIFIED)
 
@@ -107,7 +72,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
         self.assertNotIn('date-summary', response.content)
 
     def test_course_info_logged_out(self):
-        course = self.create_course_run()
+        course = create_course_run()
         url = reverse('info', args=(course.id,))
         response = self.client.get(url)
         self.assertEqual(200, response.status_code)
@@ -167,7 +132,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
     )
     @ddt.unpack
     def test_enabled_block_types(self, course_kwargs, user_kwargs, expected_blocks):
-        course = self.create_course_run(**course_kwargs)
+        course = create_course_run(**course_kwargs)
         user = self.create_user(**user_kwargs)
         CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.VERIFIED)
         self.assert_block_types(course, user, expected_blocks)
@@ -183,12 +148,12 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
     )
     @ddt.unpack
     def test_enabled_block_types_without_enrollment(self, course_kwargs, expected_blocks):
-        course = self.create_course_run(**course_kwargs)
+        course = create_course_run(**course_kwargs)
         user = self.create_user()
         self.assert_block_types(course, user, expected_blocks)
 
     def test_enabled_block_types_with_non_upgradeable_course_run(self):
-        course = self.create_course_run(days_till_start=-10, days_till_verification_deadline=None)
+        course = create_course_run(days_till_start=-10, days_till_verification_deadline=None)
         user = self.create_user()
         CourseMode.objects.get(course_id=course.id, mode_slug=CourseMode.VERIFIED).delete()
         CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.AUDIT)
@@ -200,7 +165,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
         and displays the correct time, accounting for daylight savings
         """
         with freeze_time('2015-01-02'):
-            course = self.create_course_run()
+            course = create_course_run()
             user = self.create_user()
             block = TodaysDate(course, user)
             self.assertTrue(block.is_enabled)
@@ -214,7 +179,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
     @override_waffle_flag(UNIFIED_COURSE_TAB_FLAG, active=True)
     def test_todays_date_no_timezone(self, url_name):
         with freeze_time('2015-01-02'):
-            course = self.create_course_run()
+            course = create_course_run()
             user = self.create_user()
             self.client.login(username=user.username, password=TEST_PASSWORD)
 
@@ -239,7 +204,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
     @override_waffle_flag(UNIFIED_COURSE_TAB_FLAG, active=True)
     def test_todays_date_timezone(self, url_name):
         with freeze_time('2015-01-02'):
-            course = self.create_course_run()
+            course = create_course_run()
             user = self.create_user()
             self.client.login(username=user.username, password=TEST_PASSWORD)
             set_user_preference(user, 'time_zone', 'America/Los_Angeles')
@@ -260,7 +225,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
 
     ## Tests Course Start Date
     def test_course_start_date(self):
-        course = self.create_course_run()
+        course = create_course_run()
         user = self.create_user()
         block = CourseStartDate(course, user)
         self.assertEqual(block.date, course.start)
@@ -272,7 +237,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
     @override_waffle_flag(UNIFIED_COURSE_TAB_FLAG, active=True)
     def test_start_date_render(self, url_name):
         with freeze_time('2015-01-02'):
-            course = self.create_course_run()
+            course = create_course_run()
             user = self.create_user()
             self.client.login(username=user.username, password=TEST_PASSWORD)
             url = reverse(url_name, args=(course.id,))
@@ -291,7 +256,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
     @override_waffle_flag(UNIFIED_COURSE_TAB_FLAG, active=True)
     def test_start_date_render_time_zone(self, url_name):
         with freeze_time('2015-01-02'):
-            course = self.create_course_run()
+            course = create_course_run()
             user = self.create_user()
             self.client.login(username=user.username, password=TEST_PASSWORD)
             set_user_preference(user, 'time_zone', 'America/Los_Angeles')
@@ -307,7 +272,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
 
     ## Tests Course End Date Block
     def test_course_end_date_for_certificate_eligible_mode(self):
-        course = self.create_course_run(days_till_start=-1)
+        course = create_course_run(days_till_start=-1)
         user = self.create_user()
         CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.VERIFIED)
         block = CourseEndDate(course, user)
@@ -317,7 +282,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
         )
 
     def test_course_end_date_for_non_certificate_eligible_mode(self):
-        course = self.create_course_run(days_till_start=-1)
+        course = create_course_run(days_till_start=-1)
         user = self.create_user()
         CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.AUDIT)
         block = CourseEndDate(course, user)
@@ -328,7 +293,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
         self.assertEqual(block.title, 'Course End')
 
     def test_course_end_date_after_course(self):
-        course = self.create_course_run(days_till_start=-2, days_till_end=-1)
+        course = create_course_run(days_till_start=-2, days_till_end=-1)
         user = self.create_user()
         CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.VERIFIED)
         block = CourseEndDate(course, user)
@@ -342,7 +307,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
         """Verify the block link redirects to ecommerce checkout if it's enabled."""
         sku = 'TESTSKU'
         configuration = CommerceConfiguration.objects.create(checkout_on_ecommerce_service=True)
-        course = self.create_course_run()
+        course = create_course_run()
         user = self.create_user()
         course_mode = CourseMode.objects.get(course_id=course.id, mode_slug=CourseMode.VERIFIED)
         course_mode.sku = sku
@@ -355,7 +320,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
     ## CertificateAvailableDate
     @waffle.testutils.override_switch('certificates.instructor_paced_only', True)
     def test_no_certificate_available_date(self):
-        course = self.create_course_run(days_till_start=-1)
+        course = create_course_run(days_till_start=-1)
         user = self.create_user()
         CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.AUDIT)
         block = CertificateAvailableDate(course, user)
@@ -365,7 +330,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
     ## CertificateAvailableDate
     @waffle.testutils.override_switch('certificates.instructor_paced_only', True)
     def test_no_certificate_available_date_for_self_paced(self):
-        course = self.create_self_paced_course_run()
+        course = create_self_paced_course_run()
         verified_user = self.create_user()
         CourseEnrollmentFactory(course_id=course.id, user=verified_user, mode=CourseMode.VERIFIED)
         course.certificate_available_date = datetime.now(utc) + timedelta(days=7)
@@ -376,7 +341,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
 
     @waffle.testutils.override_switch('certificates.instructor_paced_only', True)
     def test_certificate_available_date_defined(self):
-        course = self.create_course_run()
+        course = create_course_run()
         audit_user = self.create_user()
         CourseEnrollmentFactory(course_id=course.id, user=audit_user, mode=CourseMode.AUDIT)
         verified_user = self.create_user()
@@ -398,14 +363,14 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
 
     ## VerificationDeadlineDate
     def test_no_verification_deadline(self):
-        course = self.create_course_run(days_till_start=-1, days_till_verification_deadline=None)
+        course = create_course_run(days_till_start=-1, days_till_verification_deadline=None)
         user = self.create_user()
         CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.VERIFIED)
         block = VerificationDeadlineDate(course, user)
         self.assertFalse(block.is_enabled)
 
     def test_no_verified_enrollment(self):
-        course = self.create_course_run(days_till_start=-1)
+        course = create_course_run(days_till_start=-1)
         user = self.create_user()
         CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.AUDIT)
         block = VerificationDeadlineDate(course, user)
@@ -413,7 +378,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
 
     def test_verification_deadline_date_upcoming(self):
         with freeze_time('2015-01-02'):
-            course = self.create_course_run(days_till_start=-1)
+            course = create_course_run(days_till_start=-1)
             user = self.create_user()
             CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.VERIFIED)
 
@@ -430,7 +395,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
 
     def test_verification_deadline_date_retry(self):
         with freeze_time('2015-01-02'):
-            course = self.create_course_run(days_till_start=-1)
+            course = create_course_run(days_till_start=-1)
             user = self.create_user(verification_status='denied')
             CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.VERIFIED)
 
@@ -447,7 +412,7 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
 
     def test_verification_deadline_date_denied(self):
         with freeze_time('2015-01-02'):
-            course = self.create_course_run(days_till_start=-10, days_till_verification_deadline=-1)
+            course = create_course_run(days_till_start=-10, days_till_verification_deadline=-1)
             user = self.create_user(verification_status='denied')
             CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.VERIFIED)
 
@@ -469,47 +434,44 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
     @ddt.unpack
     def test_render_date_string_past(self, delta, expected_date_string):
         with freeze_time('2015-01-02'):
-            course = self.create_course_run(days_till_start=-10, days_till_verification_deadline=delta)
+            course = create_course_run(days_till_start=-10, days_till_verification_deadline=delta)
             user = self.create_user(verification_status='denied')
             CourseEnrollmentFactory(course_id=course.id, user=user, mode=CourseMode.VERIFIED)
 
             block = VerificationDeadlineDate(course, user)
             self.assertEqual(block.relative_datestring, expected_date_string)
 
-    def create_self_paced_course_run(self, **kwargs):
-        defaults = {
-            'days_till_upgrade_deadline': 100,
-        }
-        defaults.update(kwargs)
 
-        course = self.create_course_run(**defaults)
-        course.self_paced = True
-        self.store.update_item(course, None)
-        overview = CourseOverview.get_from_id(course.id)
-        self.assertTrue(overview.self_paced)
+@attr(shard=1)
+class TestScheduleOverrides(SharedModuleStoreTestCase):
 
-        return course
+    def setUp(self):
+        super(TestScheduleOverrides, self).setUp()
 
-    def assert_upgrade_deadline(self, course, expected):
-        """ Asserts the VerifiedUpgradeDeadlineDate block's date matches the expected value. """
-        enrollment = CourseEnrollmentFactory(course_id=course.id, mode=CourseMode.AUDIT)
-        block = VerifiedUpgradeDeadlineDate(course, enrollment.user)
-        self.assertEqual(block.date, expected)
+        patcher = patch('openedx.core.djangoapps.schedules.signals.get_current_site')
+        mock_get_current_site = patcher.start()
+        self.addCleanup(patcher.stop)
 
+        mock_get_current_site.return_value = SiteFactory.create()
+
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, True)
     def test_date_with_self_paced_with_enrollment_before_course_start(self):
         """ Enrolling before a course begins should result in the upgrade deadline being set relative to the
         course start date. """
         global_config = DynamicUpgradeDeadlineConfiguration.objects.create(enabled=True)
-        course = self.create_self_paced_course_run(days_till_start=3)
+        course = create_self_paced_course_run(days_till_start=3)
         overview = CourseOverview.get_from_id(course.id)
         expected = overview.start + timedelta(days=global_config.deadline_days)
-        self.assert_upgrade_deadline(course, expected)
+        enrollment = CourseEnrollmentFactory(course_id=course.id, mode=CourseMode.AUDIT)
+        block = VerifiedUpgradeDeadlineDate(course, enrollment.user)
+        self.assertEqual(block.date, expected)
 
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, True)
     def test_date_with_self_paced_with_enrollment_after_course_start(self):
         """ Enrolling after a course begins should result in the upgrade deadline being set relative to the
         enrollment date. """
         global_config = DynamicUpgradeDeadlineConfiguration.objects.create(enabled=True)
-        course = self.create_self_paced_course_run(days_till_start=-1)
+        course = create_self_paced_course_run(days_till_start=-1)
         enrollment = CourseEnrollmentFactory(course_id=course.id, mode=CourseMode.AUDIT)
         block = VerifiedUpgradeDeadlineDate(course, enrollment.user)
         expected = enrollment.created + timedelta(days=global_config.deadline_days)
@@ -517,28 +479,120 @@ class CourseDateSummaryTest(SharedModuleStoreTestCase):
 
         # Courses should be able to override the deadline
         course_config = CourseDynamicUpgradeDeadlineConfiguration.objects.create(
-            enabled=True, course_id=course.id, opt_out=False, deadline_days=3
+            enabled=True, course_id=course.id, deadline_days=3
         )
         enrollment = CourseEnrollmentFactory(course_id=course.id, mode=CourseMode.AUDIT)
         block = VerifiedUpgradeDeadlineDate(course, enrollment.user)
         expected = enrollment.created + timedelta(days=course_config.deadline_days)
         self.assertEqual(block.date, expected)
 
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, True)
     def test_date_with_self_paced_without_dynamic_upgrade_deadline(self):
         """ Disabling the dynamic upgrade deadline functionality should result in the verified mode's
         expiration date being returned. """
         DynamicUpgradeDeadlineConfiguration.objects.create(enabled=False)
-        course = self.create_self_paced_course_run()
+        course = create_self_paced_course_run()
         expected = CourseMode.objects.get(course_id=course.id, mode_slug=CourseMode.VERIFIED).expiration_datetime
-        self.assert_upgrade_deadline(course, expected)
-
-    def test_date_with_self_paced_with_course_opt_out(self):
-        """ If the course run has opted out of the dynamic deadline, the course mode's deadline should be used. """
-        course = self.create_self_paced_course_run(days_till_start=-1)
-        DynamicUpgradeDeadlineConfiguration.objects.create(enabled=True)
-        CourseDynamicUpgradeDeadlineConfiguration.objects.create(enabled=True, course_id=course.id, opt_out=True)
         enrollment = CourseEnrollmentFactory(course_id=course.id, mode=CourseMode.AUDIT)
+        block = VerifiedUpgradeDeadlineDate(course, enrollment.user)
+        self.assertEqual(block.date, expected)
+
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, True)
+    def test_date_with_self_paced_with_single_course(self):
+        """ If the global switch is off, a single course can still be enabled. """
+        course = create_self_paced_course_run(days_till_start=-1)
+        DynamicUpgradeDeadlineConfiguration.objects.create(enabled=False)
+        course_config = CourseDynamicUpgradeDeadlineConfiguration.objects.create(enabled=True, course_id=course.id)
+        enrollment = CourseEnrollmentFactory(course_id=course.id, mode=CourseMode.AUDIT)
+
+        block = VerifiedUpgradeDeadlineDate(course, enrollment.user)
+        expected = enrollment.created + timedelta(days=course_config.deadline_days)
+        self.assertEqual(block.date, expected)
+
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, True)
+    def test_date_with_existing_schedule(self):
+        """ If a schedule is created while deadlines are disabled, they shouldn't magically appear once the feature is
+        turned on. """
+        course = create_self_paced_course_run(days_till_start=-1)
+        DynamicUpgradeDeadlineConfiguration.objects.create(enabled=False)
+        course_config = CourseDynamicUpgradeDeadlineConfiguration.objects.create(enabled=False, course_id=course.id)
+        enrollment = CourseEnrollmentFactory(course_id=course.id, mode=CourseMode.AUDIT)
+
+        # The enrollment has a schedule, but the upgrade deadline should be None
+        self.assertIsNone(enrollment.schedule.upgrade_deadline)
 
         block = VerifiedUpgradeDeadlineDate(course, enrollment.user)
         expected = CourseMode.objects.get(course_id=course.id, mode_slug=CourseMode.VERIFIED).expiration_datetime
         self.assertEqual(block.date, expected)
+
+        # Now if we turn on the feature for this course, this existing enrollment should be unaffected
+        course_config.enabled = True
+        course_config.save()
+
+        block = VerifiedUpgradeDeadlineDate(course, enrollment.user)
+        self.assertEqual(block.date, expected)
+
+
+def create_course_run(
+    days_till_start=1, days_till_end=14, days_till_upgrade_deadline=4, days_till_verification_deadline=14,
+):
+    """ Create a new course run and course modes.
+
+    All date-related arguments are relative to the current date-time (now) unless otherwise specified.
+
+    Both audit and verified `CourseMode` objects will be created for the course run.
+
+    Arguments:
+        days_till_end (int): Number of days until the course ends.
+        days_till_start (int): Number of days until the course starts.
+        days_till_upgrade_deadline (int): Number of days until the course run's upgrade deadline.
+        days_till_verification_deadline (int): Number of days until the course run's verification deadline. If this
+            value is set to `None` no deadline will be verification deadline will be created.
+    """
+    now = datetime.now(utc)
+    course = CourseFactory.create(start=now + timedelta(days=days_till_start))
+
+    course.end = None
+    if days_till_end is not None:
+        course.end = now + timedelta(days=days_till_end)
+
+    CourseModeFactory(course_id=course.id, mode_slug=CourseMode.AUDIT)
+    CourseModeFactory(
+        course_id=course.id,
+        mode_slug=CourseMode.VERIFIED,
+        expiration_datetime=now + timedelta(days=days_till_upgrade_deadline)
+    )
+
+    if days_till_verification_deadline is not None:
+        VerificationDeadline.objects.create(
+            course_key=course.id,
+            deadline=now + timedelta(days=days_till_verification_deadline)
+        )
+
+    return course
+
+
+def create_self_paced_course_run(days_till_start=1):
+    """ Create a new course run and course modes.
+
+    All date-related arguments are relative to the current date-time (now) unless otherwise specified.
+
+    Both audit and verified `CourseMode` objects will be created for the course run.
+
+    Arguments:
+        days_till_start (int): Number of days until the course starts.
+    """
+    now = datetime.now(utc)
+    course = CourseFactory.create(start=now + timedelta(days=days_till_start), self_paced=True)
+
+    CourseModeFactory(
+        course_id=course.id,
+        mode_slug=CourseMode.AUDIT
+    )
+    CourseModeFactory(
+        course_id=course.id,
+        mode_slug=CourseMode.VERIFIED,
+        expiration_datetime=now + timedelta(days=100)
+    )
+
+    return course
diff --git a/lms/djangoapps/courseware/tests/test_views.py b/lms/djangoapps/courseware/tests/test_views.py
index 054e1608d9..eb08d7934e 100644
--- a/lms/djangoapps/courseware/tests/test_views.py
+++ b/lms/djangoapps/courseware/tests/test_views.py
@@ -1447,12 +1447,12 @@ class ProgressPageTests(ProgressPageBaseTests):
         """Test that query counts remain the same for self-paced and instructor-paced courses."""
         SelfPacedConfiguration(enabled=self_paced_enabled).save()
         self.setup_course(self_paced=self_paced)
-        with self.assertNumQueries(42, table_blacklist=QUERY_COUNT_TABLE_BLACKLIST), check_mongo_calls(2):
+        with self.assertNumQueries(43, table_blacklist=QUERY_COUNT_TABLE_BLACKLIST), check_mongo_calls(2):
             self._get_progress_page()
 
     @ddt.data(
-        (False, 42, 28),
-        (True, 35, 24)
+        (False, 43, 27),
+        (True, 36, 23)
     )
     @ddt.unpack
     def test_progress_queries(self, enable_waffle, initial, subsequent):
diff --git a/lms/envs/aws.py b/lms/envs/aws.py
index 6bbbbf922b..c350e7fe94 100644
--- a/lms/envs/aws.py
+++ b/lms/envs/aws.py
@@ -1028,3 +1028,12 @@ PARENTAL_CONSENT_AGE_LIMIT = ENV_TOKENS.get(
     'PARENTAL_CONSENT_AGE_LIMIT',
     PARENTAL_CONSENT_AGE_LIMIT
 )
+
+############## Settings for ACE ####################################
+ACE_ENABLED_CHANNELS = ENV_TOKENS.get('ACE_ENABLED_CHANNELS', ACE_ENABLED_CHANNELS)
+ACE_ENABLED_POLICIES = ENV_TOKENS.get('ACE_ENABLED_POLICIES', ACE_ENABLED_POLICIES)
+ACE_CHANNEL_SAILTHRU_DEBUG = ENV_TOKENS.get('ACE_CHANNEL_SAILTHRU_DEBUG', ACE_CHANNEL_SAILTHRU_DEBUG)
+ACE_CHANNEL_SAILTHRU_TEMPLATE_NAME = ENV_TOKENS.get('ACE_CHANNEL_SAILTHRU_TEMPLATE_NAME', ACE_CHANNEL_SAILTHRU_TEMPLATE_NAME)
+ACE_CHANNEL_SAILTHRU_API_KEY = AUTH_TOKENS.get('ACE_CHANNEL_SAILTHRU_API_KEY', ACE_CHANNEL_SAILTHRU_API_KEY)
+ACE_CHANNEL_SAILTHRU_API_SECRET = AUTH_TOKENS.get('ACE_CHANNEL_SAILTHRU_API_SECRET', ACE_CHANNEL_SAILTHRU_API_SECRET)
+ACE_ROUTING_KEY = ENV_TOKENS.get('ACE_ROUTING_KEY', ACE_ROUTING_KEY)
diff --git a/lms/envs/common.py b/lms/envs/common.py
index b20d06c969..94c75efc89 100644
--- a/lms/envs/common.py
+++ b/lms/envs/common.py
@@ -2241,7 +2241,7 @@ INSTALLED_APPS = [
     'database_fixups',
 
     'openedx.core.djangoapps.waffle_utils',
-    'openedx.core.djangoapps.schedules',
+    'openedx.core.djangoapps.schedules.apps.SchedulesConfig',
 
     # Features
     'openedx.features.course_bookmarks',
@@ -3288,3 +3288,18 @@ COURSES_API_CACHE_TIMEOUT = 3600  # Value is in seconds
 
 ############## Settings for CourseGraph ############################
 COURSEGRAPH_JOB_QUEUE = LOW_PRIORITY_QUEUE
+
+
+############## Settings for ACE ####################################
+ACE_ENABLED_CHANNELS = [
+    'sailthru_email'
+]
+ACE_ENABLED_POLICIES = [
+    'bulk_email_optout'
+]
+ACE_CHANNEL_SAILTHRU_DEBUG = True
+ACE_CHANNEL_SAILTHRU_TEMPLATE_NAME = 'Automated Communication Engine Email'
+ACE_CHANNEL_SAILTHRU_API_KEY = None
+ACE_CHANNEL_SAILTHRU_API_SECRET = None
+
+ACE_ROUTING_KEY = LOW_PRIORITY_QUEUE
diff --git a/openedx/core/djangoapps/content/course_overviews/tests/__init__.py b/openedx/core/djangoapps/content/course_overviews/tests/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/openedx/core/djangoapps/content/course_overviews/tests/factories.py b/openedx/core/djangoapps/content/course_overviews/tests/factories.py
new file mode 100644
index 0000000000..7d8acc497c
--- /dev/null
+++ b/openedx/core/djangoapps/content/course_overviews/tests/factories.py
@@ -0,0 +1,30 @@
+import json
+
+import factory
+from factory.django import DjangoModelFactory
+
+from ..models import CourseOverview
+from opaque_keys.edx.locator import CourseLocator
+
+
+class CourseOverviewFactory(DjangoModelFactory):
+    class Meta(object):
+        model = CourseOverview
+        django_get_or_create = ('id', )
+
+    version = CourseOverview.VERSION
+    pre_requisite_courses = []
+    start = factory.Faker('past_datetime')
+    org = 'edX'
+
+    @factory.lazy_attribute
+    def _pre_requisite_courses_json(self):
+        return json.dumps(self.pre_requisite_courses)
+
+    @factory.lazy_attribute
+    def _location(self):
+        return self.id.make_usage_key('course', 'course')
+
+    @factory.lazy_attribute
+    def id(self):
+        return CourseLocator(self.org, 'toy', '2012_Fall')
diff --git a/openedx/core/djangoapps/content/course_overviews/tests.py b/openedx/core/djangoapps/content/course_overviews/tests/test_course_overviews.py
similarity index 99%
rename from openedx/core/djangoapps/content/course_overviews/tests.py
rename to openedx/core/djangoapps/content/course_overviews/tests/test_course_overviews.py
index c8e67da636..5558d7e4aa 100644
--- a/openedx/core/djangoapps/content/course_overviews/tests.py
+++ b/openedx/core/djangoapps/content/course_overviews/tests/test_course_overviews.py
@@ -35,7 +35,7 @@ from xmodule.modulestore.django import modulestore
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory, check_mongo_calls, check_mongo_calls_range
 
-from .models import CourseOverview, CourseOverviewImageSet, CourseOverviewImageConfig
+from ..models import CourseOverview, CourseOverviewImageSet, CourseOverviewImageConfig
 
 
 @attr(shard=3)
diff --git a/openedx/core/djangoapps/schedules/admin.py b/openedx/core/djangoapps/schedules/admin.py
index dc4cabb732..34ec24addd 100644
--- a/openedx/core/djangoapps/schedules/admin.py
+++ b/openedx/core/djangoapps/schedules/admin.py
@@ -25,3 +25,9 @@ class ScheduleAdmin(admin.ModelAdmin):
         qs = super(ScheduleAdmin, self).get_queryset(request)
         qs = qs.select_related('enrollment', 'enrollment__user')
         return qs
+
+
+@admin.register(models.ScheduleConfig)
+class ScheduleConfigAdmin(admin.ModelAdmin):
+    search_fields = ('site',)
+    list_display = ('site', 'create_schedules', 'enqueue_recurring_nudge', 'deliver_recurring_nudge')
diff --git a/openedx/core/djangoapps/schedules/apps.py b/openedx/core/djangoapps/schedules/apps.py
index 87f6f1321c..a05be5447f 100644
--- a/openedx/core/djangoapps/schedules/apps.py
+++ b/openedx/core/djangoapps/schedules/apps.py
@@ -8,4 +8,4 @@ class SchedulesConfig(AppConfig):
 
     def ready(self):
         # noinspection PyUnresolvedReferences
-        from . import signals  # pylint: disable=unused-variable
+        from . import signals, tasks  # pylint: disable=unused-variable
diff --git a/openedx/core/djangoapps/schedules/management/__init__.py b/openedx/core/djangoapps/schedules/management/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/openedx/core/djangoapps/schedules/management/commands/__init__.py b/openedx/core/djangoapps/schedules/management/commands/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/openedx/core/djangoapps/schedules/management/commands/send_recurring_nudge.py b/openedx/core/djangoapps/schedules/management/commands/send_recurring_nudge.py
new file mode 100644
index 0000000000..15b8b8b990
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/management/commands/send_recurring_nudge.py
@@ -0,0 +1,80 @@
+from __future__ import print_function
+
+import datetime
+import logging
+
+from django.contrib.sites.models import Site
+from django.core.management.base import BaseCommand
+import pytz
+
+from edx_ace.utils.date import serialize
+from openedx.core.djangoapps.schedules.models import ScheduleConfig
+from openedx.core.djangoapps.schedules.tasks import recurring_nudge_schedule_hour
+from openedx.core.djangoapps.site_configuration.models import SiteConfiguration
+
+from edx_ace.recipient_resolver import RecipientResolver
+
+
+LOG = logging.getLogger(__name__)
+
+
+class ScheduleStartResolver(RecipientResolver):
+    def __init__(self, site, current_date):
+        self.site = site
+        self.current_date = current_date.replace(hour=0, minute=0, second=0)
+
+    def send(self, day, override_recipient_email=None):
+        """
+        Send a message to all users whose schedule started at ``self.current_date`` - ``day``.
+        """
+        if not ScheduleConfig.current(self.site).enqueue_recurring_nudge:
+            return
+
+        try:
+            site_config = SiteConfiguration.objects.get(site_id=self.site.id)
+            org_list = site_config.values.get('course_org_filter', None)
+            exclude_orgs = False
+            if not org_list:
+                not_orgs = set()
+                for other_site_config in SiteConfiguration.objects.all():
+                    not_orgs.update(other_site_config.values.get('course_org_filter', []))
+                org_list = list(not_orgs)
+                exclude_orgs = True
+            elif not isinstance(org_list, list):
+                org_list = [org_list]
+        except SiteConfiguration.DoesNotExist:
+            org_list = None
+            exclude_orgs = False
+
+        target_date = self.current_date - datetime.timedelta(days=day)
+        for hour in range(24):
+            target_hour = target_date + datetime.timedelta(hours=hour)
+            recurring_nudge_schedule_hour.apply_async(
+                (self.site.id, day, serialize(target_hour), org_list, exclude_orgs, override_recipient_email),
+                retry=False,
+            )
+
+
+class Command(BaseCommand):
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--date',
+            default=datetime.datetime.utcnow().date().isoformat(),
+            help='The date to compute weekly messages relative to, in YYYY-MM-DD format',
+        )
+        parser.add_argument(
+            '--override-recipient-email',
+            help='Send all emails to this address instead of the actual recipient'
+        )
+        parser.add_argument('site_domain_name')
+
+    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'])
+        resolver = ScheduleStartResolver(site, current_date)
+        for day in (3, 10):
+            resolver.send(day, options.get('override_recipient_email'))
diff --git a/openedx/core/djangoapps/schedules/management/commands/send_verified_upgrade_deadline_reminder.py b/openedx/core/djangoapps/schedules/management/commands/send_verified_upgrade_deadline_reminder.py
new file mode 100644
index 0000000000..a0d39ab2b2
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/management/commands/send_verified_upgrade_deadline_reminder.py
@@ -0,0 +1,118 @@
+from __future__ import print_function
+
+import datetime
+
+from dateutil.tz import tzutc, gettz
+from django.core.management.base import BaseCommand
+from django.test.utils import CaptureQueriesContext
+from django.db.models import Prefetch
+from django.conf import settings
+from django.core.urlresolvers import reverse
+from django.db import DEFAULT_DB_ALIAS, connections
+from django.utils.http import urlquote
+
+from openedx.core.djangoapps.schedules.models import Schedule
+from openedx.core.djangoapps.user_api.models import UserPreference
+
+from edx_ace.message import MessageType
+from edx_ace.recipient_resolver import RecipientResolver
+from edx_ace import ace
+from edx_ace.recipient import Recipient
+
+
+from course_modes.models import CourseMode, format_course_price
+from lms.djangoapps.experiments.utils import check_and_get_upgrade_link
+
+
+class VerifiedUpgradeDeadlineReminder(MessageType):
+    pass
+
+
+class VerifiedDeadlineResolver(RecipientResolver):
+    def __init__(self, target_deadline):
+        self.target_deadline = target_deadline
+
+    def send(self, msg_type):
+        for (user, language, context) in self.build_email_context():
+            msg = msg_type.personalize(
+                Recipient(
+                    user.username,
+                    user.email,
+                ),
+                language,
+                context
+            )
+            ace.send(msg)
+
+    def build_email_context(self):
+        schedules = Schedule.objects.select_related(
+            'enrollment__user__profile',
+            'enrollment__course',
+        ).prefetch_related(
+            Prefetch(
+                'enrollment__course__modes',
+                queryset=CourseMode.objects.filter(mode_slug=CourseMode.VERIFIED),
+                to_attr='verified_modes'
+            ),
+            Prefetch(
+                'enrollment__user__preferences',
+                queryset=UserPreference.objects.filter(key='time_zone'),
+                to_attr='tzprefs'
+            ),
+        ).filter(
+            upgrade_deadline__year=self.schedule_deadline.year,
+            upgrade_deadline__month=self.schedule_deadline.month,
+            upgrade_deadline__day=self.schedule_deadline.day,
+        )
+
+        if "read_replica" in settings.DATABASES:
+            schedules = schedules.using("read_replica")
+
+        for schedule in schedules:
+            enrollment = schedule.enrollment
+            user = enrollment.user
+
+            user_time_zone = tzutc()
+            for preference in user.tzprefs:
+                user_time_zone = gettz(preference.value)
+
+            course_id_str = str(enrollment.course_id)
+            course = enrollment.course
+
+            course_root = reverse('course_root', kwargs={'course_id': urlquote(course_id_str)})
+
+            def absolute_url(relative_path):
+                return u'{}{}'.format(settings.LMS_ROOT_URL, relative_path)
+
+            template_context = {
+                'user_full_name': user.profile.name,
+                'user_personal_address': user.profile.name if user.profile.name else user.username,
+                'user_username': user.username,
+                'user_time_zone': user_time_zone,
+                'user_schedule_start_time': schedule.start,
+                'user_schedule_verified_upgrade_deadline_time': schedule.upgrade_deadline,
+                'course_id': course_id_str,
+                'course_title': course.display_name,
+                'course_url': absolute_url(course_root),
+                'course_image_url': absolute_url(course.course_image_url),
+                'course_end_time': course.end,
+                'course_verified_upgrade_url': check_and_get_upgrade_link(course, user),
+                'course_verified_upgrade_price': format_course_price(course.verified_modes[0].min_price),
+            }
+
+            yield (user, course.language, template_context)
+
+
+class Command(BaseCommand):
+
+    def add_arguments(self, parser):
+        parser.add_argument('--date', default=datetime.datetime.utcnow().date().isoformat())
+
+    def handle(self, *args, **options):
+        current_date = datetime.date(*[int(x) for x in options['date'].split('-')])
+
+        msg_t = VerifiedUpgradeDeadlineReminder()
+
+        for offset in (2, 9, 16):
+            target_date = current_date + datetime.timedelta(days=offset)
+            VerifiedDeadlineResolver(target_date).send(msg_t)
diff --git a/openedx/core/djangoapps/schedules/management/commands/tests/__init__.py b/openedx/core/djangoapps/schedules/management/commands/tests/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/openedx/core/djangoapps/schedules/management/commands/tests/test_send_recurring_nudge.py b/openedx/core/djangoapps/schedules/management/commands/tests/test_send_recurring_nudge.py
new file mode 100644
index 0000000000..4d381a5f97
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/management/commands/tests/test_send_recurring_nudge.py
@@ -0,0 +1,169 @@
+import datetime
+from mock import patch, Mock
+from unittest import skipUnless
+import pytz
+
+import ddt
+from django.conf import settings
+
+from edx_ace.utils.date import serialize
+
+from opaque_keys.edx.keys import CourseKey
+from openedx.core.djangoapps.schedules import tasks
+from openedx.core.djangoapps.schedules.management.commands import send_recurring_nudge as nudge
+from openedx.core.djangoapps.site_configuration.tests.factories import SiteFactory
+from openedx.core.djangolib.testing.utils import CacheIsolationTestCase, skip_unless_lms
+from openedx.core.djangoapps.schedules.tests.factories import ScheduleFactory, ScheduleConfigFactory
+from openedx.core.djangoapps.site_configuration.tests.factories import SiteConfigurationFactory
+
+
+@ddt.ddt
+@skip_unless_lms
+@skipUnless('openedx.core.djangoapps.schedules.apps.SchedulesConfig' in settings.INSTALLED_APPS, "Can't test schedules if the app isn't installed")
+class TestSendRecurringNudge(CacheIsolationTestCase):
+
+    # pylint: disable=protected-access
+
+    def setUp(self):
+        ScheduleFactory.create(start=datetime.datetime(2017, 8, 1, 15, 44, 30, tzinfo=pytz.UTC))
+        ScheduleFactory.create(start=datetime.datetime(2017, 8, 1, 17, 34, 30, tzinfo=pytz.UTC))
+        ScheduleFactory.create(start=datetime.datetime(2017, 8, 2, 15, 34, 30, tzinfo=pytz.UTC))
+
+        site = SiteFactory.create()
+        self.site_config = SiteConfigurationFactory.create(site=site)
+        ScheduleConfigFactory.create(site=self.site_config.site)
+
+    @patch.object(nudge, 'ScheduleStartResolver')
+    def test_handle(self, mock_resolver):
+        test_time = datetime.datetime(2017, 8, 1, tzinfo=pytz.UTC)
+        nudge.Command().handle(date='2017-08-01', site_domain_name=self.site_config.site.domain)
+        mock_resolver.assert_called_with(self.site_config.site, test_time)
+
+        for day in (3, 10):
+            mock_resolver().send.assert_any_call(day, None)
+
+    @patch.object(tasks, 'ace')
+    @patch.object(nudge, 'recurring_nudge_schedule_hour')
+    def test_resolver_send(self, mock_schedule_hour, mock_ace):
+        current_time = datetime.datetime(2017, 8, 1, tzinfo=pytz.UTC)
+        nudge.ScheduleStartResolver(self.site_config.site, current_time).send(3)
+        test_time = current_time - datetime.timedelta(days=3)
+        self.assertFalse(mock_schedule_hour.called)
+        mock_schedule_hour.apply_async.assert_any_call(
+            (self.site_config.site.id, 3, serialize(test_time), [], True, None),
+            retry=False,
+        )
+        mock_schedule_hour.apply_async.assert_any_call(
+            (self.site_config.site.id, 3, serialize(test_time + datetime.timedelta(hours=23)), [], True, None),
+            retry=False,
+        )
+        self.assertFalse(mock_ace.send.called)
+
+    @ddt.data(1, 10, 100)
+    @patch.object(tasks, 'ace')
+    @patch.object(tasks, '_recurring_nudge_schedule_send')
+    def test_schedule_hour(self, schedule_count, mock_schedule_send, mock_ace):
+        schedules = [
+            ScheduleFactory.create(start=datetime.datetime(2017, 8, 1, 18, 34, 30, tzinfo=pytz.UTC))
+            for _ in range(schedule_count)
+        ]
+
+        test_time_str = serialize(datetime.datetime(2017, 8, 1, 18, tzinfo=pytz.UTC))
+        with self.assertNumQueries(1):
+            tasks.recurring_nudge_schedule_hour(
+                self.site_config.site, 3, test_time_str, [schedules[0].enrollment.course.org],
+            )
+        self.assertEqual(mock_schedule_send.apply_async.call_count, schedule_count)
+        self.assertFalse(mock_ace.send.called)
+
+    @patch.object(tasks, '_recurring_nudge_schedule_send')
+    def test_no_course_overview(self, mock_schedule_send):
+
+        schedule = ScheduleFactory.create(
+            start=datetime.datetime(2017, 8, 1, 20, 34, 30, tzinfo=pytz.UTC),
+        )
+        schedule.enrollment.course_id = CourseKey.from_string('edX/toy/Not_2012_Fall')
+        schedule.enrollment.save()
+
+        test_time_str = serialize(datetime.datetime(2017, 8, 1, 20, tzinfo=pytz.UTC))
+        with self.assertNumQueries(1):
+            tasks.recurring_nudge_schedule_hour(
+                self.site_config.site, 3, test_time_str, [schedule.enrollment.course.org],
+            )
+
+        # There is no database constraint that enforces that enrollment.course_id points
+        # to a valid CourseOverview object. However, in that case, schedules isn't going
+        # to attempt to address it, and will instead simply skip those users.
+        # This happens 'transparently' because django generates an inner-join between
+        # enrollment and course_overview, and thus will skip any rows where course_overview
+        # is null.
+        self.assertEqual(mock_schedule_send.apply_async.call_count, 0)
+
+    @patch.object(tasks, 'ace')
+    def test_delivery_disabled(self, mock_ace):
+        ScheduleConfigFactory.create(site=self.site_config.site, deliver_recurring_nudge=False)
+
+        mock_msg = Mock()
+        tasks._recurring_nudge_schedule_send(self.site_config.site.id, mock_msg)
+        self.assertFalse(mock_ace.send.called)
+
+    @patch.object(tasks, 'ace')
+    @patch.object(nudge, 'recurring_nudge_schedule_hour')
+    def test_enqueue_disabled(self, mock_schedule_hour, mock_ace):
+        ScheduleConfigFactory.create(site=self.site_config.site, enqueue_recurring_nudge=False)
+
+        current_time = datetime.datetime(2017, 8, 1, tzinfo=pytz.UTC)
+        nudge.ScheduleStartResolver(self.site_config.site, current_time).send(3)
+        self.assertFalse(mock_schedule_hour.called)
+        self.assertFalse(mock_schedule_hour.apply_async.called)
+        self.assertFalse(mock_ace.send.called)
+
+    @patch.object(tasks, 'ace')
+    @patch.object(tasks, '_recurring_nudge_schedule_send')
+    @ddt.data(
+        ((['filtered_org'], False, 1)),
+        ((['filtered_org'], True, 2))
+    )
+    @ddt.unpack
+    def test_site_config(self, org_list, exclude_orgs, expected_message_count, mock_schedule_send, mock_ace):
+        filtered_org = 'filtered_org'
+        unfiltered_org = 'unfiltered_org'
+        site1 = SiteFactory.create(domain='foo1.bar', name='foo1.bar')
+        limited_config = SiteConfigurationFactory.create(values={'course_org_filter': [filtered_org]}, site=site1)
+        site2 = SiteFactory.create(domain='foo2.bar', name='foo2.bar')
+        unlimited_config = SiteConfigurationFactory.create(values={'course_org_filter': []}, site=site2)
+
+        for config in (limited_config, unlimited_config):
+            ScheduleConfigFactory.create(site=config.site)
+
+        filtered_sched = ScheduleFactory.create(
+            start=datetime.datetime(2017, 8, 2, 17, 44, 30, tzinfo=pytz.UTC),
+            enrollment__course__org=filtered_org,
+        )
+        unfiltered_scheds = [
+            ScheduleFactory.create(
+                start=datetime.datetime(2017, 8, 2, 17, 44, 30, tzinfo=pytz.UTC),
+                enrollment__course__org=unfiltered_org,
+            )
+            for _ in range(2)
+        ]
+
+        print(filtered_sched.enrollment)
+        print(filtered_sched.enrollment.course)
+        print(filtered_sched.enrollment.course.org)
+        print(unfiltered_scheds[0].enrollment)
+        print(unfiltered_scheds[0].enrollment.course)
+        print(unfiltered_scheds[0].enrollment.course.org)
+        print(unfiltered_scheds[1].enrollment)
+        print(unfiltered_scheds[1].enrollment.course)
+        print(unfiltered_scheds[1].enrollment.course.org)
+
+        test_time_str = serialize(datetime.datetime(2017, 8, 2, 17, tzinfo=pytz.UTC))
+        with self.assertNumQueries(1):
+            tasks.recurring_nudge_schedule_hour(
+                limited_config.site.id, 3, test_time_str, org_list=org_list, exclude_orgs=exclude_orgs,
+            )
+
+        print(mock_schedule_send.mock_calls)
+        self.assertEqual(mock_schedule_send.apply_async.call_count, expected_message_count)
+        self.assertFalse(mock_ace.send.called)
diff --git a/openedx/core/djangoapps/schedules/migrations/0003_scheduleconfig.py b/openedx/core/djangoapps/schedules/migrations/0003_scheduleconfig.py
new file mode 100644
index 0000000000..67e4e4e493
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/migrations/0003_scheduleconfig.py
@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+import django.db.models.deletion
+from django.conf import settings
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('sites', '0001_initial'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('schedules', '0002_auto_20170816_1532'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ScheduleConfig',
+            fields=[
+                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
+                ('change_date', models.DateTimeField(auto_now_add=True, verbose_name='Change date')),
+                ('enabled', models.BooleanField(default=False, verbose_name='Enabled')),
+                ('create_schedules', models.BooleanField(default=False)),
+                ('enqueue_recurring_nudge', models.BooleanField(default=False)),
+                ('deliver_recurring_nudge', models.BooleanField(default=False)),
+                ('changed_by', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, editable=False, to=settings.AUTH_USER_MODEL, null=True, verbose_name='Changed by')),
+                ('site', models.ForeignKey(to='sites.Site')),
+            ],
+            options={
+                'ordering': ('-change_date',),
+                'abstract': False,
+            },
+        ),
+    ]
diff --git a/openedx/core/djangoapps/schedules/models.py b/openedx/core/djangoapps/schedules/models.py
index 5bf0c66a5f..e1391101bf 100644
--- a/openedx/core/djangoapps/schedules/models.py
+++ b/openedx/core/djangoapps/schedules/models.py
@@ -1,6 +1,11 @@
+from collections import namedtuple
+
 from django.db import models
 from django.utils.translation import ugettext_lazy as _
 from django_extensions.db.models import TimeStampedModel
+from django.contrib.sites.models import Site
+
+from config_models.models import ConfigurationModel
 
 
 class Schedule(TimeStampedModel):
@@ -23,3 +28,12 @@ class Schedule(TimeStampedModel):
     class Meta(object):
         verbose_name = _('Schedule')
         verbose_name_plural = _('Schedules')
+
+
+class ScheduleConfig(ConfigurationModel):
+    KEY_FIELDS = ('site',)
+
+    site = models.ForeignKey(Site)
+    create_schedules = models.BooleanField(default=False)
+    enqueue_recurring_nudge = models.BooleanField(default=False)
+    deliver_recurring_nudge = models.BooleanField(default=False)
diff --git a/openedx/core/djangoapps/schedules/signals.py b/openedx/core/djangoapps/schedules/signals.py
index 13b139a5f1..380ff06c2a 100644
--- a/openedx/core/djangoapps/schedules/signals.py
+++ b/openedx/core/djangoapps/schedules/signals.py
@@ -3,68 +3,84 @@ import logging
 
 from django.db.models.signals import post_save
 from django.dispatch import receiver
-from django.utils import timezone
 
 from course_modes.models import CourseMode
 from courseware.models import DynamicUpgradeDeadlineConfiguration, CourseDynamicUpgradeDeadlineConfiguration
-from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
-from openedx.core.djangoapps.waffle_utils import WaffleSwitchNamespace
+from openedx.core.djangoapps.theming.helpers import get_current_site
+from openedx.core.djangoapps.waffle_utils import WaffleFlagNamespace, CourseWaffleFlag
 from student.models import CourseEnrollment
-from .models import Schedule
+from .models import Schedule, ScheduleConfig
 
 log = logging.getLogger(__name__)
 
 
-def _get_upgrade_deadline(enrollment):
-    """ Returns the upgrade deadline for the given enrollment.
-
-    The deadline is determined based on the following data (in priority order):
-        1. Course run-specific deadline configuration (CourseDynamicUpgradeDeadlineConfiguration)
-        2. Global deadline configuration (DynamicUpgradeDeadlineConfiguration)
-        3. Verified course mode expiration
-    """
-    course_key = enrollment.course_id
-    upgrade_deadline = None
-
-    try:
-        verified_mode = CourseMode.verified_mode_for_course(course_key)
-        if verified_mode:
-            upgrade_deadline = verified_mode.expiration_datetime
-    except CourseMode.DoesNotExist:
-        pass
-
-    global_config = DynamicUpgradeDeadlineConfiguration.current()
-    if global_config.enabled:
-        delta = global_config.deadline_days
-
-        # Check if the given course has opted out of the feature
-        course_config = CourseDynamicUpgradeDeadlineConfiguration.current(course_key)
-        if course_config.enabled:
-            if course_config.opt_out:
-                return upgrade_deadline
-
-            delta = course_config.deadline_days
-
-        course_overview = CourseOverview.get_from_id(course_key)
-
-        # This represents the first date at which the learner can access the content. This will be the latter of
-        # either the enrollment date or the course's start date.
-        content_availability_date = max(enrollment.created, course_overview.start)
-        cav_based_deadline = content_availability_date + datetime.timedelta(days=delta)
-
-        # If the deadline from above is None, make sure we have a value for comparison
-        upgrade_deadline = upgrade_deadline or datetime.date.max
-
-        # The content availability-based deadline should never occur after the verified mode's
-        # expiration date, if one is set.
-        upgrade_deadline = min(upgrade_deadline, cav_based_deadline)
-
-    return upgrade_deadline
+SCHEDULE_WAFFLE_FLAG = CourseWaffleFlag(
+    waffle_namespace=WaffleFlagNamespace('schedules'),
+    flag_name='create_schedules_for_course',
+    flag_undefined_default=False
+)
 
 
 @receiver(post_save, sender=CourseEnrollment, dispatch_uid='create_schedule_for_enrollment')
 def create_schedule(sender, **kwargs):
-    if WaffleSwitchNamespace('schedules').is_enabled('enable-create-schedule-receiver') and kwargs['created']:
-        enrollment = kwargs['instance']
-        upgrade_deadline = _get_upgrade_deadline(enrollment)
-        Schedule.objects.create(enrollment=enrollment, start=timezone.now(), upgrade_deadline=upgrade_deadline)
+    if not kwargs['created']:
+        # only create schedules when enrollment records are created
+        return
+
+    current_site = get_current_site()
+    if current_site is None:
+        log.debug('Schedules: No current site')
+        return
+
+    enrollment = kwargs['instance']
+    schedule_config = ScheduleConfig.current(current_site)
+    if (
+        not schedule_config.create_schedules
+        and not SCHEDULE_WAFFLE_FLAG.is_enabled(enrollment.course_id)
+    ):
+        log.debug('Schedules: Creation not enabled for this course or for this site')
+        return
+
+    delta = None
+    if enrollment.course_overview.self_paced:
+        global_config = DynamicUpgradeDeadlineConfiguration.current()
+        if global_config.enabled:
+            # Use the default from this model whether or not the feature is enabled
+            delta = global_config.deadline_days
+
+        # Check if the course has a deadline override
+        course_config = CourseDynamicUpgradeDeadlineConfiguration.current(enrollment.course_id)
+        if course_config.enabled:
+            delta = course_config.deadline_days
+
+    upgrade_deadline = None
+
+    # This represents the first date at which the learner can access the content. This will be the latter of
+    # either the enrollment date or the course's start date.
+    content_availability_date = max(enrollment.created, enrollment.course_overview.start)
+
+    if delta is not None:
+        upgrade_deadline = content_availability_date + datetime.timedelta(days=delta)
+
+        course_upgrade_deadline = None
+        try:
+            verified_mode = CourseMode.verified_mode_for_course(enrollment.course_id)
+        except CourseMode.DoesNotExist:
+            pass
+        else:
+            if verified_mode:
+                course_upgrade_deadline = verified_mode.expiration_datetime
+
+        if course_upgrade_deadline is not None and upgrade_deadline is not None:
+            # The content availability-based deadline should never occur after the verified mode's
+            # expiration date, if one is set.
+            upgrade_deadline = min(upgrade_deadline, course_upgrade_deadline)
+
+    Schedule.objects.create(
+        enrollment=enrollment,
+        start=content_availability_date,
+        upgrade_deadline=upgrade_deadline
+    )
+
+    log.debug('Schedules: created a new schedule starting at %s with an upgrade deadline of %s',
+              content_availability_date, upgrade_deadline)
diff --git a/openedx/core/djangoapps/schedules/tasks.py b/openedx/core/djangoapps/schedules/tasks.py
new file mode 100644
index 0000000000..b8ae3d5b1d
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/tasks.py
@@ -0,0 +1,94 @@
+import datetime
+
+from celery.task import task
+from django.conf import settings
+from django.contrib.sites.models import Site
+from django.core.urlresolvers import reverse
+from django.utils.http import urlquote
+
+from edx_ace import ace
+from edx_ace.message import MessageType, Message
+from edx_ace.recipient import Recipient
+from edx_ace.utils.date import deserialize
+from openedx.core.djangoapps.schedules.models import Schedule, ScheduleConfig
+
+
+ROUTING_KEY = getattr(settings, 'ACE_ROUTING_KEY', None)
+
+
+class RecurringNudge(MessageType):
+    def __init__(self, day, *args, **kwargs):
+        super(RecurringNudge, self).__init__(*args, **kwargs)
+        self.name = "recurringnudge_day{}".format(day)
+
+
+@task(ignore_result=True, routing_key=ROUTING_KEY)
+def recurring_nudge_schedule_hour(
+    site_id, day, target_hour_str, org_list, exclude_orgs=False, override_recipient_email=None,
+):
+    target_hour = deserialize(target_hour_str)
+    msg_type = RecurringNudge(day)
+
+    for (user, language, context) in _recurring_nudge_schedules_for_hour(target_hour, org_list, exclude_orgs):
+        msg = msg_type.personalize(
+            Recipient(
+                user.username,
+                override_recipient_email or user.email,
+            ),
+            language,
+            context,
+        )
+        _recurring_nudge_schedule_send.apply_async((site_id, str(msg)), retry=False)
+
+
+@task(ignore_result=True, routing_key=ROUTING_KEY)
+def _recurring_nudge_schedule_send(site_id, msg_str):
+    site = Site.objects.get(pk=site_id)
+    if not ScheduleConfig.current(site).deliver_recurring_nudge:
+        return
+
+    msg = Message.from_string(msg_str)
+    ace.send(msg)
+
+
+def _recurring_nudge_schedules_for_hour(target_hour, org_list, exclude_orgs=False):
+    schedules = Schedule.objects.select_related(
+        'enrollment__user__profile',
+        'enrollment__course',
+    ).filter(
+        start__gte=target_hour,
+        start__lt=target_hour + datetime.timedelta(minutes=60),
+        enrollment__is_active=True,
+    )
+
+    if org_list is not None:
+        if exclude_orgs:
+            schedules = schedules.exclude(enrollment__course__org__in=org_list)
+        else:
+            schedules = schedules.filter(enrollment__course__org__in=org_list)
+
+    if "read_replica" in settings.DATABASES:
+        schedules = schedules.using("read_replica")
+
+    for schedule in schedules:
+        enrollment = schedule.enrollment
+        user = enrollment.user
+
+        course_id_str = str(enrollment.course_id)
+        course = enrollment.course
+
+        course_root = reverse('course_root', args=[course_id_str])
+
+        def absolute_url(relative_path):
+            return u'{}{}'.format(settings.LMS_ROOT_URL, urlquote(relative_path))
+
+        template_context = {
+            'student_name': user.profile.name,
+            'course_name': course.display_name,
+            'course_url': absolute_url(course_root),
+
+            # This is used by the bulk email optout policy
+            'course_id': course_id_str,
+        }
+
+        yield (user, course.language, template_context)
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/common/base_body.html b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/common/base_body.html
new file mode 100644
index 0000000000..66b55a5b85
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/common/base_body.html
@@ -0,0 +1,179 @@
+{% load i18n %}
+{% get_current_language as LANGUAGE_CODE %}
+<div bgcolor="#f5f5f5" lang="{{ LANGUAGE_CODE|default:"en" }}" style="
+    margin: 0;
+    padding: 0;
+    min-width: 100%;
+">
+    <!-- This is preview text that is visible in the inbox view of many email clients but not visible in the actual
+    email itself. -->
+    <div style="
+        display:none;
+        font-size:1px;
+        line-height:1px;
+        max-height:0px;
+        max-width:0px;
+        opacity:0;
+        overflow:hidden;
+        visibility:hidden;
+    ">
+        {% block preview_text %}{% endblock %}
+    </div>
+
+    <!-- Hack for outlook 2010, which wants to render everything in Times New Roman -->
+    <!--[if mso]>
+    <style type="text/css">
+    body, table, td {font-family: 'Open Sans', 'Helvetica Neue', Helvetica, Arial, sans-serif !important;}
+    </style>
+    <![endif]-->
+
+    <!--[if (gte mso 9)|(IE)]>
+    <table role="presentation" width="600" align="center" cellpadding="0" cellspacing="0" border="0">
+    <tr>
+    <td>
+    <![endif]-->
+
+    <!-- CONTENT -->
+    <table class="content" role="presentation" align="center" cellpadding="0" cellspacing="0" border="0" bgcolor="#f5f5f5" width="100%" style="
+        font-family: 'Open Sans', 'Helvetica Neue', Helvetica, Arial, sans-serif;
+        font-size: 1em;
+        line-height: 1.5;
+        max-width: 600px;
+        padding: 0 20px 0 20px;
+    ">
+        <tr>
+            <!-- HEADER -->
+            <td class="header" style="
+                padding: 20px;
+            ">
+                <table role="presentation" width="100%" align="left" border="0" cellpadding="0" cellspacing="0">
+                    <tr>
+                        <td width="70">
+                            <a href="http://www.edx.org"><img
+                                    src="https://media.sailthru.com/595/1k1/8/o/599f355101b3f.png" width="70"
+                                    height="30" alt="edX Home Page"/></a>
+                        </td>
+                        <td align="right" style="text-align: right;">
+                            <a class="login" href="https://courses.edx.org/dashboard" style="color: #005686;">Sign In</a>
+                        </td>
+                    </tr>
+                </table>
+            </td>
+        </tr>
+
+        <tr>
+            <!-- MAIN -->
+            <td class="main" bgcolor="#ffffff" style="
+                padding: 30px 20px;
+                box-shadow: 0 1px 5px rgba(0,0,0,0.25);
+            ">
+                {% block content %}{% endblock %}
+            </td>
+        </tr>
+
+        <tr>
+            <!-- FOOTER -->
+            <td class="footer" style="padding: 20px;">
+                <table role="presentation" width="100%" align="left" border="0" cellpadding="0" cellspacing="0">
+                    <tr>
+                        <td style="padding-bottom: 20px;">
+                            <!-- LOGO / SOCIAL -->
+                            <table role="presentation" width="100%" align="left" border="0" cellpadding="0" cellspacing="0">
+                                <tr>
+                                    <td width="70">
+                                        <!-- LOGO -->
+                                        <a href="http://www.edx.org"><img
+                                                src="https://media.sailthru.com/595/1k1/8/o/599f355101b3f.png"
+                                                width="70" height="30" alt="edX Home Page"/></a>
+                                    </td>
+                                    <td align="right">
+                                        <!-- SOCIAL -->
+                                        <table role="presentation" border="0" cellpadding="0" cellspacing="0" width="210">
+                                            <tr>
+                                                <td height="32" width="42" align="right">
+                                                    <a href="https://www.linkedin.com/company/edx">
+                                                        <img src="https://media.sailthru.com/595/1k1/8/o/599f354ec70cb.png"
+                                                             width="32" height="32" alt="edX on LinkedIn"/>
+                                                    </a>
+                                                </td>
+                                                <td height="32" width="42" align="right">
+                                                    <a href="https://www.twitter.com/edXOnline/">
+                                                        <img src="https://media.sailthru.com/595/1k1/8/o/599f354d9c26e.png"
+                                                             width="32" height="32" alt="edX on Twitter"/>
+                                                    </a>
+                                                </td>
+                                                <td height="32" width="42" align="right">
+                                                    <a href="http://www.facebook.com/edX">
+                                                        <img src="https://media.sailthru.com/595/1k1/8/o/599f355052c8e.png"
+                                                             width="32" height="32" alt="edX on Facebook"/>
+                                                    </a>
+                                                </td>
+                                                <td height="32" width="42" align="right">
+                                                    <a href="https://plus.google.com/%2BedXOnline">
+                                                        <img src="https://media.sailthru.com/595/1k1/8/o/599f354fc554a.png"
+                                                             width="32" height="32" alt="edX on Google Plus"/>
+                                                    </a>
+                                                </td>
+                                                <td height="32" width="42" align="right">
+                                                    <a href="https://www.reddit.com/r/edX/">
+                                                        <img src="https://media.sailthru.com/595/1k1/8/o/599f354e326b9.png"
+                                                             width="32" height="32" alt="edX on Reddit"/>
+                                                    </a>
+                                                </td>
+                                            </tr>
+                                        </table>
+                                    </td>
+                                </tr>
+                            </table>
+                        </td>
+                    </tr>
+                    <tr>
+                        <!-- ACTIONS / APP BUTTONS -->
+                        <td>
+                            <table role="presentation" width="100%" align="left" border="0" cellpadding="0" cellspacing="0">
+                                <tr>
+                                    <!-- APP BUTTONS -->
+                                    <td class="col" width="148" valign="top" align="right" style="padding-bottom: 20px;">
+                                        <a href="https://itunes.apple.com/us/app/edx/id945480667?mt=8" style="text-decoration: none">
+                                            <img src="https://media.sailthru.com/595/1k1/6/2/5931cfbba391b.png"
+                                                 alt="Download the iOS app on the Apple Store" width="136" height="50"/>
+                                        </a>
+                                        <a href="https://play.google.com/store/apps/details?id=org.edx.mobile" style="text-decoration: none">
+                                            <img src="https://media.sailthru.com/595/1k1/6/2/5931cf879a033.png"
+                                                 alt="Download the Android app on the Google Play Store"
+                                                 width="136" height="50" style="margin: 10px 0 0 5px"/>
+                                        </a>
+                                    </td>
+                                </tr>
+                            </table>
+                        </td>
+                    </tr>
+                    <tr>
+                        <!-- COPYRIGHT -->
+                        <td>
+                            <table role="presentation" border="0" cellpadding="0" cellspacing="0" width="100%">
+                                <tr>
+                                    <td>
+                                        <p><small>Copyright &copy; 2017 edX, All rights
+                                            reserved.</small></p>
+                                        <p>
+                                            Our mailing address is:<br/>
+                                            141 Portland St. 9th Floor, Cambridge, MA 02139
+                                        </p>
+                                    </td>
+                                </tr>
+                            </table>
+                        </td>
+                    </tr>
+                </table>
+            </td>
+        </tr>
+    </table>
+
+    <!--[if (gte mso 9)|(IE)]>
+    </td>
+    </tr>
+    </table>
+    <![endif]-->
+
+</div>
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/common/base_head.html b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/common/base_head.html
new file mode 100644
index 0000000000..f99e4ac307
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/common/base_head.html
@@ -0,0 +1,29 @@
+<meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
+<title>{% block title %}edX Email{% endblock %}</title>
+<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
+
+<style type="text/css">
+    @media only screen and (min-device-width: 601px) {
+      .content {
+        width: 600px !important;
+      }
+    }
+
+    @-ms-viewport{
+      width: device-width;
+    }
+
+    /* Column Drop Layout Pattern CSS */
+    @media only screen and (max-width: 450px) {
+      td[class="col"] {
+        display: block;
+        width: 100%;
+        -moz-box-sizing: border-box;
+        -webkit-box-sizing: border-box;
+        box-sizing: border-box;
+        float: left;
+        text-align: left !important;
+        padding-bottom: 20px;
+      }
+    }
+</style>
\ No newline at end of file
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/body.html b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/body.html
new file mode 100644
index 0000000000..dc402b6222
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/body.html
@@ -0,0 +1,40 @@
+{% extends 'schedules/edx_ace/common/base_body.html' %}
+{% load i18n %}
+
+{% block preview_text %}
+    {% blocktrans %} Learning isn't easy - but it's worth it! Complete some problems and learn something new in {{course_name}}. {% endblocktrans %}
+{% endblock %}
+
+{% block content %}
+<table width="100%" align="left" border="0" cellpadding="0" cellspacing="0" role="presentation">
+    <tr>
+        <td>
+            <h1>{% blocktrans %} Keep up the momentum! {% endblocktrans %}</h1>
+
+            <p>
+                {% blocktrans %} Many edX learners in <strong>{{course_name}}</strong> are
+                completing more problems every week, and participating in the discussion forums. What do you want to do
+                to keep learning?  {% endblocktrans %}
+            </p>
+            <p>
+                <!-- email client support for style sheets is pretty spotty, so we have to inline all of these styles -->
+                <a href="{{ course_url }}" style="
+                    color: #ffffff;
+                    text-decoration: none;
+                    border-radius: 4px;
+                    -webkit-border-radius: 4px;
+                    -moz-border-radius: 4px;
+                    background-color: #005686;
+                    border-top: 10px solid #005686;
+                    border-bottom: 10px solid #005686;
+                    border-right: 16px solid #005686;
+                    border-left: 16px solid #005686;
+                    display: inline-block;
+                ">
+                        <font color="#ffffff"><b>{% blocktrans %} Keep learning {% endblocktrans %}</b></font>
+                </a>
+            </p>
+        </td>
+    </tr>
+</table>
+{% endblock %}
\ No newline at end of file
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/body.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/body.txt
new file mode 100644
index 0000000000..17db613c2a
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/body.txt
@@ -0,0 +1,4 @@
+{% load i18n %}
+{% blocktrans %} Keep up the momentum! Many edX learners in {{course_name}} are completing more problems every week, and participating in the discussion forums. What do you want to do to keep learning?  {% endblocktrans %}
+
+{% blocktrans %} Keep learning {% endblocktrans %} <{{course_url}}>
\ No newline at end of file
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/from_name.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/from_name.txt
new file mode 100644
index 0000000000..dc05f94dca
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/from_name.txt
@@ -0,0 +1 @@
+{{ course_name }}
\ No newline at end of file
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/head.html b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/head.html
new file mode 100644
index 0000000000..588357ec65
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/head.html
@@ -0,0 +1 @@
+{% extends 'schedules/edx_ace/common/base_head.html' %}
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/subject.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/subject.txt
new file mode 100644
index 0000000000..9935fd1307
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day10/email/subject.txt
@@ -0,0 +1,2 @@
+{% load i18n %}
+{% blocktrans %}What do you want to do to keep learning?{% endblocktrans %}
\ No newline at end of file
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/body.html b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/body.html
new file mode 100644
index 0000000000..9e90f29e11
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/body.html
@@ -0,0 +1,37 @@
+{% extends 'schedules/edx_ace/common/base_body.html' %}
+{% load i18n %}
+
+{% block preview_text %}
+    {% blocktrans %} Learning isn't easy - but it's worth it! Learn something new in {{course_name}}. {% endblocktrans %}
+{% endblock %}
+
+{% block content %}
+<table width="100%" align="left" border="0" cellpadding="0" cellspacing="0" role="presentation">
+    <tr>
+        <td>
+            <h1>{% blocktrans %} Remember when you enrolled in <strong>{{course_name}}</strong> on edX.org? {% endblocktrans %}</h1>
+
+            <p>{% blocktrans %} We do! Come see what everyone is learning. {% endblocktrans %}</p>
+
+            <p>
+                <!-- email client support for style sheets is pretty spotty, so we have to inline all of these styles -->
+                <a href="{{ course_url }}" style="
+                    color: #ffffff;
+                    text-decoration: none;
+                    border-radius: 4px;
+                    -webkit-border-radius: 4px;
+                    -moz-border-radius: 4px;
+                    background-color: #005686;
+                    border-top: 10px solid #005686;
+                    border-bottom: 10px solid #005686;
+                    border-right: 16px solid #005686;
+                    border-left: 16px solid #005686;
+                    display: inline-block;
+                ">
+                        <font color="#ffffff"><b>{% blocktrans %} Start learning now {% endblocktrans %}</b></font>
+                </a>
+            </p>
+        </td>
+    </tr>
+</table>
+{% endblock %}
\ No newline at end of file
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/body.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/body.txt
new file mode 100644
index 0000000000..d6c59707de
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/body.txt
@@ -0,0 +1,4 @@
+{% load i18n %}
+{% blocktrans %} Remember when you enrolled in {{course_name}} on edX.org? We do! Come see what everyone is learning.  {% endblocktrans %}
+
+{% blocktrans %} Start learning now {% endblocktrans %} <{{course_url}}>
\ No newline at end of file
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/from_name.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/from_name.txt
new file mode 100644
index 0000000000..dc05f94dca
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/from_name.txt
@@ -0,0 +1 @@
+{{ course_name }}
\ No newline at end of file
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/head.html b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/head.html
new file mode 100644
index 0000000000..588357ec65
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/head.html
@@ -0,0 +1 @@
+{% extends 'schedules/edx_ace/common/base_head.html' %}
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/subject.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/subject.txt
new file mode 100644
index 0000000000..1c625fec86
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/recurringnudge_day3/email/subject.txt
@@ -0,0 +1,3 @@
+{% load i18n %}
+
+{% blocktrans %} {{course_name}} has started on edX {% endblocktrans %}
\ No newline at end of file
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/body.html b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/body.html
new file mode 100644
index 0000000000..8da84e439a
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/body.html
@@ -0,0 +1,7 @@
+Dear {{ user_personal_address }},
+<br/>
+We hope you are enjoying {{ course_title }}.
+Upgrade by {{ user_schedule_verified_upgrade_deadline_time|date:"l, F dS, Y" }}
+to get a shareable certificate!
+<br/>
+<a href="{{course_verified_upgrade_url}}">Upgrade now</a>
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/body.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/body.txt
new file mode 100644
index 0000000000..beb0ae45ba
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/body.txt
@@ -0,0 +1,6 @@
+Dear {{ user_personal_address }},
+
+We hope you are enjoying {{ course_title }}.
+Upgrade by {{ user_schedule_verified_upgrade_deadline_time|date:"l, F dS, Y" }} to get a shareable certificate!
+
+Upgrade now at {{course_verified_upgrade_url}}
diff --git a/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/subject.txt b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/subject.txt
new file mode 100644
index 0000000000..8ca024cf72
--- /dev/null
+++ b/openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/subject.txt
@@ -0,0 +1 @@
+Only two days left to upgrade!
\ No newline at end of file
diff --git a/openedx/core/djangoapps/schedules/tests/factories.py b/openedx/core/djangoapps/schedules/tests/factories.py
index 796a35ce5c..7c286afb7f 100644
--- a/openedx/core/djangoapps/schedules/tests/factories.py
+++ b/openedx/core/djangoapps/schedules/tests/factories.py
@@ -2,6 +2,8 @@ import factory
 import pytz
 
 from openedx.core.djangoapps.schedules import models
+from student.tests.factories import CourseEnrollmentFactory
+from openedx.core.djangoapps.site_configuration.tests.factories import SiteFactory
 
 
 class ScheduleFactory(factory.DjangoModelFactory):
@@ -10,3 +12,14 @@ class ScheduleFactory(factory.DjangoModelFactory):
 
     start = factory.Faker('future_datetime', tzinfo=pytz.UTC)
     upgrade_deadline = factory.Faker('future_datetime', tzinfo=pytz.UTC)
+    enrollment = factory.SubFactory(CourseEnrollmentFactory)
+
+
+class ScheduleConfigFactory(factory.DjangoModelFactory):
+    class Meta(object):
+        model = models.ScheduleConfig
+
+    site = factory.SubFactory(SiteFactory)
+    create_schedules = True
+    enqueue_recurring_nudge = True
+    deliver_recurring_nudge = True
diff --git a/openedx/core/djangoapps/schedules/tests/test_signals.py b/openedx/core/djangoapps/schedules/tests/test_signals.py
index 5661c73385..798f0cad92 100644
--- a/openedx/core/djangoapps/schedules/tests/test_signals.py
+++ b/openedx/core/djangoapps/schedules/tests/test_signals.py
@@ -1,24 +1,110 @@
-from django.test import TestCase
+import datetime
+from mock import patch
+from pytz import utc
 
-from openedx.core.djangoapps.waffle_utils import WaffleSwitchNamespace
+from course_modes.models import CourseMode
+from course_modes.tests.factories import CourseModeFactory
+from courseware.models import DynamicUpgradeDeadlineConfiguration
+from openedx.core.djangoapps.schedules.signals import SCHEDULE_WAFFLE_FLAG
+from openedx.core.djangoapps.site_configuration.tests.factories import SiteFactory
+from openedx.core.djangoapps.waffle_utils.testutils import override_waffle_flag
 from openedx.core.djangolib.testing.utils import skip_unless_lms
 from student.tests.factories import CourseEnrollmentFactory
+from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory
 from ..models import Schedule
+from ..tests.factories import ScheduleConfigFactory
 
 
+@patch('openedx.core.djangoapps.schedules.signals.get_current_site')
 @skip_unless_lms
-class CreateScheduleTests(TestCase):
-    def test_create_schedule(self):
-        """ A schedule should be created for every new enrollment if the switch is active. """
+class CreateScheduleTests(SharedModuleStoreTestCase):
 
-        SWITCH_NAME = 'enable-create-schedule-receiver'
-        switch_namesapce = WaffleSwitchNamespace('schedules')
+    def assert_schedule_created(self):
+        enrollment = CourseEnrollmentFactory()
+        self.assertIsNotNone(enrollment.schedule)
+        self.assertIsNone(enrollment.schedule.upgrade_deadline)
 
-        with switch_namesapce.override(SWITCH_NAME, True):
-            enrollment = CourseEnrollmentFactory()
-            self.assertIsNotNone(enrollment.schedule)
+    def assert_schedule_not_created(self):
+        enrollment = CourseEnrollmentFactory()
+        with self.assertRaises(Schedule.DoesNotExist):
+            enrollment.schedule
 
-        with switch_namesapce.override(SWITCH_NAME, False):
-            enrollment = CourseEnrollmentFactory()
-            with self.assertRaises(Schedule.DoesNotExist):
-                enrollment.schedule
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, True)
+    def test_create_schedule(self, mock_get_current_site):
+        site = SiteFactory.create()
+        mock_get_current_site.return_value = site
+        ScheduleConfigFactory.create(site=site)
+        self.assert_schedule_created()
+
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, True)
+    def test_no_current_site(self, mock_get_current_site):
+        mock_get_current_site.return_value = None
+        self.assert_schedule_not_created()
+
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, True)
+    def test_schedule_config_disabled_waffle_enabled(self, mock_get_current_site):
+        site = SiteFactory.create()
+        mock_get_current_site.return_value = site
+        ScheduleConfigFactory.create(site=site, create_schedules=False)
+        self.assert_schedule_created()
+
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, False)
+    def test_schedule_config_enabled_waffle_disabled(self, mock_get_current_site):
+        site = SiteFactory.create()
+        mock_get_current_site.return_value = site
+        ScheduleConfigFactory.create(site=site, create_schedules=True)
+        self.assert_schedule_created()
+
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, False)
+    def test_schedule_config_disabled_waffle_disabled(self, mock_get_current_site):
+        site = SiteFactory.create()
+        mock_get_current_site.return_value = site
+        ScheduleConfigFactory.create(site=site, create_schedules=False)
+        self.assert_schedule_not_created()
+
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, True)
+    def test_schedule_config_creation_enabled_instructor_paced(self, mock_get_current_site):
+        site = SiteFactory.create()
+        mock_get_current_site.return_value = site
+        ScheduleConfigFactory.create(site=site, enabled=True, create_schedules=True)
+        course = create_self_paced_course_run()
+        DynamicUpgradeDeadlineConfiguration.objects.create(enabled=False)
+        enrollment = CourseEnrollmentFactory(course_id=course.id, mode=CourseMode.AUDIT)
+
+        self.assertEqual(enrollment.schedule.start, enrollment.created)
+        self.assertIsNone(enrollment.schedule.upgrade_deadline)
+
+    @override_waffle_flag(SCHEDULE_WAFFLE_FLAG, True)
+    def test_schedule_config_creation_enabled_instructor_paced_with_deadline(self, mock_get_current_site):
+        site = SiteFactory.create()
+        mock_get_current_site.return_value = site
+        ScheduleConfigFactory.create(site=site, enabled=True, create_schedules=True)
+        course = create_self_paced_course_run()
+        global_config = DynamicUpgradeDeadlineConfiguration.objects.create(enabled=True)
+        enrollment = CourseEnrollmentFactory(course_id=course.id, mode=CourseMode.AUDIT)
+        expected_deadline = enrollment.created + datetime.timedelta(days=global_config.deadline_days)
+
+        self.assertEqual(enrollment.schedule.start, enrollment.created)
+        self.assertEqual(enrollment.schedule.upgrade_deadline, expected_deadline)
+
+
+def create_self_paced_course_run():
+    """ Create a new course run and course modes.
+
+    Both audit and verified `CourseMode` objects will be created for the course run.
+    """
+    now = datetime.datetime.now(utc)
+    course = CourseFactory.create(start=now + datetime.timedelta(days=-1), self_paced=True)
+
+    CourseModeFactory(
+        course_id=course.id,
+        mode_slug=CourseMode.AUDIT
+    )
+    CourseModeFactory(
+        course_id=course.id,
+        mode_slug=CourseMode.VERIFIED,
+        expiration_datetime=now + datetime.timedelta(days=100)
+    )
+
+    return course
diff --git a/openedx/features/course_experience/tests/views/test_course_home.py b/openedx/features/course_experience/tests/views/test_course_home.py
index 1a65f28b55..a1e6876fda 100644
--- a/openedx/features/course_experience/tests/views/test_course_home.py
+++ b/openedx/features/course_experience/tests/views/test_course_home.py
@@ -166,7 +166,7 @@ class TestCourseHomePage(CourseHomePageTestCase):
         course_home_url(self.course)
 
         # Fetch the view and verify the query counts
-        with self.assertNumQueries(42, table_blacklist=QUERY_COUNT_TABLE_BLACKLIST):
+        with self.assertNumQueries(41, table_blacklist=QUERY_COUNT_TABLE_BLACKLIST):
             with check_mongo_calls(4):
                 url = course_home_url(self.course)
                 self.client.get(url)
diff --git a/openedx/features/course_experience/tests/views/test_course_updates.py b/openedx/features/course_experience/tests/views/test_course_updates.py
index cdb1fca549..a346947917 100644
--- a/openedx/features/course_experience/tests/views/test_course_updates.py
+++ b/openedx/features/course_experience/tests/views/test_course_updates.py
@@ -127,7 +127,7 @@ class TestCourseUpdatesPage(SharedModuleStoreTestCase):
         course_updates_url(self.course)
 
         # Fetch the view and verify that the query counts haven't changed
-        with self.assertNumQueries(32, table_blacklist=QUERY_COUNT_TABLE_BLACKLIST):
+        with self.assertNumQueries(33, table_blacklist=QUERY_COUNT_TABLE_BLACKLIST):
             with check_mongo_calls(4):
                 url = course_updates_url(self.course)
                 self.client.get(url)
diff --git a/requirements/edx/github.txt b/requirements/edx/github.txt
index 5fc3b165c0..2be637328c 100644
--- a/requirements/edx/github.txt
+++ b/requirements/edx/github.txt
@@ -97,6 +97,7 @@ git+https://github.com/edx/xblock-utils.git@v1.0.5#egg=xblock-utils==1.0.5
 git+https://github.com/edx/edx-user-state-client.git@1.0.1#egg=edx-user-state-client==1.0.1
 git+https://github.com/edx/xblock-lti-consumer.git@v1.1.5#egg=lti_consumer-xblock==1.1.5
 git+https://github.com/edx/edx-proctoring.git@1.2.0#egg=edx-proctoring==1.2.0
+git+https://github.com/edx/edx-ace.git@v0.1.0#egg=edx-ace
 
 # Third Party XBlocks
 git+https://github.com/open-craft/xblock-poll@7ba819b968fe8faddb78bb22e1fe7637005eb414#egg=xblock-poll==1.2.7
diff --git a/setup.py b/setup.py
index 6d9846c0b0..61bf07eb69 100644
--- a/setup.py
+++ b/setup.py
@@ -58,5 +58,8 @@ setup(
             "milestones = lms.djangoapps.course_api.blocks.transformers.milestones:MilestonesAndSpecialExamsTransformer",
             "grades = lms.djangoapps.grades.transformer:GradesTransformer",
         ],
+        "openedx.ace.policy": [
+            "bulk_email_optout = lms.djangoapps.bulk_email.policies:CourseEmailOptout"
+        ],
     }
 )