Refactor upgrade_reminder to use async celery task

Finish test_base.py tests

Address some PR comments

Some test_send_recurring_nudge fixes

Fix test_schedule_bin

Fix test_site_config

Fix test_multiple_enrollments

Tests pass now!

Use consistent naming: upgrade_reminder

openedx/core/djangoapps/schedules/management/commands/__init__.py

+import datetime
+import logging
+
+import pytz
+from django.contrib.sites.models import Site
+from django.core.management.base import BaseCommand
+from edx_ace.recipient_resolver import RecipientResolver
+from edx_ace.utils.date import serialize
+
+from openedx.core.djangoapps.schedules.models import ScheduleConfig
+from openedx.core.djangoapps.schedules.tasks import DEFAULT_NUM_BINS
+from openedx.core.djangoapps.site_configuration.models import SiteConfiguration
+
+
+LOG = logging.getLogger(__name__)
+
+
+class PrefixedDebugLoggerMixin(object):
+    def __init__(self, *args, **kwargs):
+        self.log_prefix = self.__class__.__name__
+
+    def log_debug(self, message, *args, **kwargs):
+        LOG.debug(self.log_prefix + ': ' + message, *args, **kwargs)
+
+
+class BinnedSchedulesBaseResolver(RecipientResolver, PrefixedDebugLoggerMixin):
+    """
+    Starts num_bins number of async tasks, each of which sends emails to an equal group of learners.
+    """
+    def __init__(self, site, current_date, *args, **kwargs):
+        super(BinnedSchedulesBaseResolver, self).__init__(*args, **kwargs)
+        self.site = site
+        self.current_date = current_date.replace(hour=0, minute=0, second=0)
+        self.async_send_task = None  # define in subclasses
+        self.num_bins = DEFAULT_NUM_BINS
+        self.enqueue_config_var = None  # define in subclasses
+        self.log_prefix = self.__class__.__name__
+
+    def send(self, day_offset, override_recipient_email=None):
+        if not self.is_enqueue_enabled():
+            self.log_debug('Message queuing disabled for site %s', self.site.domain)
+            return
+
+        exclude_orgs, org_list = self.get_course_org_filter()
+
+        target_date = self.current_date + datetime.timedelta(days=day_offset)
+        self.log_debug('Target date = %s', target_date.isoformat())
+        for bin in range(self.num_bins):
+            task_args = (
+                self.site.id, serialize(target_date), day_offset, bin, org_list, exclude_orgs, override_recipient_email,
+            )
+            self.log_debug('Launching task with args = %r', task_args)
+            self.async_send_task.apply_async(
+                task_args,
+                retry=False,
+            )
+
+    def is_enqueue_enabled(self):
+        if self.enqueue_config_var:
+            return getattr(ScheduleConfig.current(self.site), self.enqueue_config_var)
+        return False
+
+    def get_course_org_filter(self):
+        """
+        Given the configuration of sites, get the list of orgs that should be included or excluded from this send.
+
+        Returns:
+             tuple: Returns a tuple (exclude_orgs, org_list). If exclude_orgs is True, then org_list is a list of the
+                only orgs that should be included in this send. If exclude_orgs is False, then org_list is a list of
+                orgs that should be excluded from this send. All other orgs should be included.
+        """
+        try:
+            site_config = SiteConfiguration.objects.get(site_id=self.site.id)
+            org_list = site_config.get_value('course_org_filter')
+            exclude_orgs = False
+            if not org_list:
+                not_orgs = set()
+                for other_site_config in SiteConfiguration.objects.all():
+                    other = other_site_config.get_value('course_org_filter')
+                    if not isinstance(other, list):
+                        if other is not None:
+                            not_orgs.add(other)
+                    else:
+                        not_orgs.update(other)
+                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
+        finally:
+            return exclude_orgs, org_list
+
+
+class SendEmailBaseCommand(BaseCommand, PrefixedDebugLoggerMixin):
+    def __init__(self, *args, **kwargs):
+        super(SendEmailBaseCommand, self).__init__(*args, **kwargs)
+        self.resolver_class = BinnedSchedulesBaseResolver
+        self.log_prefix = self.__class__.__name__
+
+    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):
+        resolver = self.make_resolver(*args, **options)
+        self.send_emails(resolver, *args, **options)
+
+    def make_resolver(self, *args, **options):
+        current_date = datetime.datetime(
+            *[int(x) for x in options['date'].split('-')],
+            tzinfo=pytz.UTC
+        )
+        self.log_debug('Args = %r', options)
+        self.log_debug('Current date = %s', current_date.isoformat())
+
+        site = Site.objects.get(domain__iexact=options['site_domain_name'])
+        self.log_debug('Running for site %s', site.domain)
+        return self.resolver_class(site, current_date)
+
+    def send_emails(self, resolver, *args, **options):
+        resolver.send(0, options.get('override_recipient_email'))

openedx/core/djangoapps/schedules/management/commands/send_recurring_nudge.py

 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
+from openedx.core.djangoapps.schedules.management.commands import SendEmailBaseCommand, BinnedSchedulesBaseResolver
+from openedx.core.djangoapps.schedules.tasks import RECURRING_NUDGE_NUM_BINS, recurring_nudge_schedule_bin
 
 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:
-            LOG.debug('Recurring Nudge: Message queuing disabled for site %s', self.site.domain)
-            return
-
-        exclude_orgs, org_list = self.get_org_filter()
-
-        target_date = self.current_date - datetime.timedelta(days=day)
-        LOG.debug('Scheduled Nudge: Target date = %s', target_date.isoformat())
-        for hour in range(24):
-            target_hour = target_date + datetime.timedelta(hours=hour)
-            task_args = (self.site.id, day, serialize(target_hour), org_list, exclude_orgs, override_recipient_email)
-            LOG.debug('Scheduled Nudge: Launching task with args = %r', task_args)
-            recurring_nudge_schedule_hour.apply_async(task_args, retry=False)
-
-    def get_org_filter(self):
-        """
-        Given the configuration of sites, get the list of orgs that should be included or excluded from this send.
-
-        Returns:
-             tuple: Returns a tuple (exclude_orgs, org_list). If exclude_orgs is True, then org_list is a list of the
-                only orgs that should be included in this send. If exclude_orgs is False, then org_list is a list of
-                orgs that should be excluded from this send. All other orgs should be included.
-        """
-        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
-        return exclude_orgs, org_list
-
-
-class Command(BaseCommand):
+class ScheduleStartResolver(BinnedSchedulesBaseResolver):
+    """
+    Send a message to all users whose schedule started at ``self.current_date`` + ``day_offset``.
+    """
+    def __init__(self, *args, **kwargs):
+        super(ScheduleStartResolver, self).__init__(*args, **kwargs)
+        self.async_send_task = recurring_nudge_schedule_bin
+        self.num_bins = RECURRING_NUDGE_NUM_BINS
+        self.log_prefix = 'Scheduled Nudge'
+        self.enqueue_config_var = 'enqueue_recurring_nudge'
 
-    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
-        )
-        LOG.debug('Scheduled Nudge: Args = %r', options)
-        LOG.debug('Scheduled Nudge: Current date = %s', current_date.isoformat())
+class Command(SendEmailBaseCommand):
+    def __init__(self, *args, **kwargs):
+        super(Command, self).__init__(*args, **kwargs)
+        self.resolver_class = ScheduleStartResolver
+        self.log_prefix = 'Scheduled Nudge'
 
-        site = Site.objects.get(domain__iexact=options['site_domain_name'])
-        LOG.debug('Scheduled Nudge: Running for site %s', site.domain)
-        resolver = ScheduleStartResolver(site, current_date)
-        for day in (3, 10):
-            resolver.send(day, options.get('override_recipient_email'))
+    def send_emails(self, resolver, *args, **options):
+        for day_offset in (-3, -10):
+            resolver.send(day_offset, options.get('override_recipient_email'))

openedx/core/djangoapps/schedules/management/commands/send_upgrade_reminder.py

+from __future__ import print_function
+
+import logging
+
+from openedx.core.djangoapps.schedules.management.commands import SendEmailBaseCommand, BinnedSchedulesBaseResolver
+from openedx.core.djangoapps.schedules.tasks import (
+    UPGRADE_REMINDER_NUM_BINS,
+    upgrade_reminder_schedule_bin
+)
+
+
+LOG = logging.getLogger(__name__)
+
+
+class UpgradeReminderResolver(BinnedSchedulesBaseResolver):
+    """
+    Send a message to all users whose verified upgrade deadline is at ``self.current_date`` + ``day_offset``.
+    """
+    def __init__(self, *args, **kwargs):
+        super(UpgradeReminderResolver, self).__init__(*args, **kwargs)
+        self.async_send_task = upgrade_reminder_schedule_bin
+        self.num_bins = UPGRADE_REMINDER_NUM_BINS
+        self.log_prefix = 'Upgrade Reminder'
+        self.enqueue_config_var = 'enqueue_upgrade_reminder'
+
+
+class Command(SendEmailBaseCommand):
+    def __init__(self, *args, **kwargs):
+        super(Command, self).__init__(*args, **kwargs)
+        self.resolver_class = UpgradeReminderResolver
+        self.log_prefix = 'Upgrade Reminder'
+
+    def send_emails(self, resolver, *args, **options):
+        logging.basicConfig(level=logging.DEBUG)
+        resolver.send(2, options.get('override_recipient_email'))

openedx/core/djangoapps/schedules/management/commands/send_verified_upgrade_deadline_reminder.py

-from __future__ import print_function
-
-import datetime
-
-from dateutil.tz import tzutc, gettz
-from django.core.management.base import BaseCommand
-from django.db.models import Prefetch
-from django.conf import settings
-from django.core.urlresolvers import reverse
-from django.utils.http import urlquote
-
-from openedx.core.djangoapps.schedules.message_type import ScheduleMessageType
-from openedx.core.djangoapps.schedules.models import Schedule
-from openedx.core.djangoapps.user_api.models import UserPreference
-
-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(ScheduleMessageType):
-    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)

openedx/core/djangoapps/schedules/management/commands/tests/test_base.py

+import datetime
+from unittest import skipUnless
+
+import ddt
+import pytz
+from django.conf import settings
+from mock import patch, Mock
+
+from openedx.core.djangoapps.schedules.management.commands import (
+    DEFAULT_NUM_BINS,
+    SendEmailBaseCommand,
+    BinnedSchedulesBaseResolver
+)
+from openedx.core.djangoapps.schedules.tests.factories import ScheduleConfigFactory, ScheduleFactory
+from openedx.core.djangoapps.site_configuration.tests.factories import SiteConfigurationFactory, SiteFactory
+from openedx.core.djangolib.testing.utils import CacheIsolationTestCase, skip_unless_lms
+
+
+@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 TestBinnedSchedulesBaseResolver(CacheIsolationTestCase):
+    def setUp(self):
+        super(TestBinnedSchedulesBaseResolver, self).setUp()
+
+        self.site = SiteFactory.create()
+        self.site_config = SiteConfigurationFactory.create(site=self.site)
+        self.schedule_config = ScheduleConfigFactory.create(site=self.site)
+
+    def setup_resolver(self, site=None, current_date=None):
+        if site is None:
+            site = self.site
+        if current_date is None:
+            current_date = datetime.datetime.now()
+        resolver = BinnedSchedulesBaseResolver(self.site, current_date)
+        return resolver
+
+    def test_init_site(self):
+        resolver = self.setup_resolver()
+        assert resolver.site == self.site
+
+    def test_init_current_date(self):
+        current_time = datetime.datetime.now()
+        resolver = self.setup_resolver(current_date=current_time)
+        current_date = current_time.replace(hour=0, minute=0, second=0)
+        assert resolver.current_date == current_date
+
+    def test_init_async_send_task(self):
+        resolver = self.setup_resolver()
+        assert resolver.async_send_task is None
+
+    def test_init_num_bins(self):
+        resolver = self.setup_resolver()
+        assert resolver.num_bins == DEFAULT_NUM_BINS
+
+    def test_send_enqueue_disabled(self):
+        resolver = self.setup_resolver()
+        resolver.is_enqueue_enabled = lambda: False
+        with patch.object(resolver, 'async_send_task') as send:
+            with patch.object(resolver, 'log_debug') as log_debug:
+                resolver.send(day_offset=2)
+                log_debug.assert_called_once_with('Message queuing disabled for site %s', self.site.domain)
+                send.apply_async.assert_not_called()
+
+    @ddt.data(0, 2, -3)
+    def test_send_enqueue_enabled(self, day_offset):
+        resolver = self.setup_resolver()
+        resolver.is_enqueue_enabled = lambda: True
+        resolver.get_course_org_filter = lambda: (False, None)
+        with patch.object(resolver, 'async_send_task') as send:
+            with patch.object(resolver, 'log_debug') as log_debug:
+                resolver.send(day_offset=day_offset)
+                target_date = resolver.current_date + datetime.timedelta(day_offset)
+                log_debug.assert_any_call('Target date = %s', target_date.isoformat())
+                assert send.apply_async.call_count == DEFAULT_NUM_BINS
+
+    @ddt.data(True, False)
+    def test_is_enqueue_enabled(self, enabled):
+        resolver = self.setup_resolver()
+        resolver.enqueue_config_var = 'enqueue_recurring_nudge'
+        self.schedule_config.enqueue_recurring_nudge = enabled
+        self.schedule_config.save()
+        assert resolver.is_enqueue_enabled() == enabled
+
+    @ddt.unpack
+    @ddt.data(
+        ('course1', ['course1']),
+        (['course1', 'course2'], ['course1', 'course2'])
+    )
+    def test_get_course_org_filter_include(self, course_org_filter, expected_org_list):
+        resolver = self.setup_resolver()
+        self.site_config.values['course_org_filter'] = course_org_filter
+        self.site_config.save()
+        exclude_orgs, org_list = resolver.get_course_org_filter()
+        assert not exclude_orgs
+        assert org_list == expected_org_list
+
+    # factory_boy doesn't make sense at all
+    @ddt.unpack
+    @ddt.data(
+        (None, []),
+        ('course1', [u'course1']),
+        (['course1', 'course2'], [u'course1', u'course2'])
+    )
+    def test_get_course_org_filter_exclude(self, course_org_filter, expected_org_list):
+        resolver = self.setup_resolver()
+        self.other_site = SiteFactory.create()
+        self.other_site_config = SiteConfigurationFactory.create(
+            site=self.other_site,
+            values={'course_org_filter': course_org_filter},
+        )
+        exclude_orgs, org_list = resolver.get_course_org_filter()
+        assert exclude_orgs
+        self.assertItemsEqual(org_list, expected_org_list)
+
+
+@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 TestSendEmailBaseCommand(CacheIsolationTestCase):
+    def setUp(self):
+        self.command = SendEmailBaseCommand()
+
+    def test_init_resolver_class(self):
+        assert self.command.resolver_class == BinnedSchedulesBaseResolver
+
+    def test_make_resolver(self):
+        with patch.object(self.command, 'resolver_class') as resolver_class:
+            example_site = SiteFactory(domain='example.com')
+            self.command.make_resolver(site_domain_name='example.com', date='2017-09-29')
+            resolver_class.assert_called_once_with(
+                example_site,
+                datetime.datetime(2017, 9, 29, tzinfo=pytz.UTC)
+            )
+
+    def test_send_emails(self):
+        resolver = Mock()
+        self.command.send_emails(resolver, override_recipient_email='foo@example.com')
+        resolver.send.assert_called_once_with(0, 'foo@example.com')
+
+    def test_handle(self):
+        with patch.object(self.command, 'make_resolver') as make_resolver:
+            make_resolver.return_value = 'resolver'
+            with patch.object(self.command, 'send_emails') as send_emails:
+                self.command.handle(date='2017-09-29')
+                make_resolver.assert_called_once_with(date='2017-09-29')
+                send_emails.assert_called_once_with('resolver', date='2017-09-29')

openedx/core/djangoapps/schedules/migrations/0004_auto_20170922_1428.py

+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('schedules', '0003_scheduleconfig'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='scheduleconfig',
+            name='deliver_upgrade_reminder',
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AddField(
+            model_name='scheduleconfig',
+            name='enqueue_upgrade_reminder',
+            field=models.BooleanField(default=False),
+        ),
+    ]

openedx/core/djangoapps/schedules/models.py

@@ -35,3 +35,5 @@ class ScheduleConfig(ConfigurationModel):
     create_schedules = models.BooleanField(default=False)
     enqueue_recurring_nudge = models.BooleanField(default=False)
     deliver_recurring_nudge = models.BooleanField(default=False)
+    enqueue_upgrade_reminder = models.BooleanField(default=False)
+    deliver_upgrade_reminder = models.BooleanField(default=False)

openedx/core/djangoapps/schedules/tasks.py

 import datetime
 from itertools import groupby
 import logging
-from urlparse import urlparse
 
 from celery.task import task
 from django.conf import settings
@@ -9,9 +8,8 @@ from django.contrib.auth.models import User
 from django.contrib.sites.models import Site
 from django.core.exceptions import ValidationError
 from django.core.urlresolvers import reverse
-from django.db.models import Min
+from django.db.models import F, Min, Prefetch
 from django.db.utils import DatabaseError
-from django.utils.http import urlquote
 
 from edx_ace import ace
 from edx_ace.message import Message
@@ -19,9 +17,17 @@ from edx_ace.recipient import Recipient
 from edx_ace.utils.date import deserialize
 from opaque_keys.edx.keys import CourseKey
 
+from course_modes.models import CourseMode
+from edxmako.shortcuts import marketing_link
 from openedx.core.djangoapps.schedules.message_type import ScheduleMessageType
 from openedx.core.djangoapps.schedules.models import Schedule, ScheduleConfig
-from openedx.core.djangoapps.schedules.template_context import absolute_url, get_base_template_context
+from openedx.core.djangoapps.schedules.template_context import (
+    absolute_url,
+    encode_url,
+    encode_urls_in_dict,
+    get_base_template_context
+)
+from openedx.core.djangoapps.user_api.models import UserPreference
 
 
 LOG = logging.getLogger(__name__)
@@ -32,6 +38,9 @@ KNOWN_RETRY_ERRORS = (  # Errors we expect occasionally that could resolve on re
     DatabaseError,
     ValidationError,
 )
+DEFAULT_NUM_BINS = 24
+RECURRING_NUDGE_NUM_BINS = DEFAULT_NUM_BINS
+UPGRADE_REMINDER_NUM_BINS = DEFAULT_NUM_BINS
 
 
 @task(bind=True, default_retry_delay=30, routing_key=ROUTING_KEY)
@@ -57,6 +66,7 @@ class RecurringNudge(ScheduleMessageType):
         self.name = "recurringnudge_day{}".format(day)
 
 
+# TODO: delete once recurring_nudge_schedule_bin is fully rolled out
 @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,
@@ -88,6 +98,7 @@ def _recurring_nudge_schedule_send(site_id, msg_str):
     ace.send(msg)
 
 
+# TODO: delete once _recurring_nudge_schedules_for_bin is fully rolled out
 def _recurring_nudge_schedules_for_hour(target_hour, org_list, exclude_orgs=False):
     beginning_of_day = target_hour.replace(hour=0, minute=0, second=0)
     users = User.objects.filter(
@@ -129,6 +140,84 @@ def _recurring_nudge_schedules_for_hour(target_hour, org_list, exclude_orgs=Fals
         course_id_strs = [str(schedule.enrollment.course_id) for schedule in user_schedules]
 
         first_schedule = user_schedules[0]
+        template_context = {
+            'student_name': user.profile.name,
+
+            'course_name': first_schedule.enrollment.course.display_name,
+            'course_url': absolute_url(reverse('course_root', args=[str(first_schedule.enrollment.course_id)])),
+
+            # This is used by the bulk email optout policy
+            'course_ids': course_id_strs,
+
+            # Platform information
+            'homepage_url': encode_url(marketing_link('ROOT')),
+            'dashboard_url': absolute_url(dashboard_relative_url),
+            'template_revision': settings.EDX_PLATFORM_REVISION,
+            'platform_name': settings.PLATFORM_NAME,
+            'contact_mailing_address': settings.CONTACT_MAILING_ADDRESS,
+            'social_media_urls': encode_urls_in_dict(getattr(settings, 'SOCIAL_MEDIA_FOOTER_URLS', {})),
+            'mobile_store_urls': encode_urls_in_dict(getattr(settings, 'MOBILE_STORE_URLS', {})),
+        }
+        yield (user, first_schedule.enrollment.course.language, template_context)
+
+
+@task(ignore_result=True, routing_key=ROUTING_KEY)
+def recurring_nudge_schedule_bin(
+    site_id, target_day_str, day_offset, bin_num, org_list, exclude_orgs=False, override_recipient_email=None,
+):
+    target_day = deserialize(target_day_str)
+    msg_type = RecurringNudge(abs(day_offset))
+
+    for (user, language, context) in _recurring_nudge_schedules_for_bin(target_day, bin_num, 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)
+
+
+def _recurring_nudge_schedules_for_bin(target_day, bin_num, org_list, exclude_orgs=False):
+    beginning_of_day = target_day.replace(hour=0, minute=0, second=0)
+    users = User.objects.filter(
+        courseenrollment__schedule__start__gte=beginning_of_day,
+        courseenrollment__schedule__start__lt=beginning_of_day + datetime.timedelta(days=1),
+        courseenrollment__is_active=True,
+    ).annotate(
+        first_schedule=Min('courseenrollment__schedule__start')
+    ).annotate(
+        id_mod=F('id') % RECURRING_NUDGE_NUM_BINS
+    ).filter(
+        id_mod=bin_num
+    )
+
+    schedules = Schedule.objects.select_related(
+        'enrollment__user__profile',
+        'enrollment__course',
+    ).filter(
+        enrollment__user__in=users,
+        start__gte=beginning_of_day,
+        start__lt=beginning_of_day + datetime.timedelta(days=1),
+        enrollment__is_active=True,
+    ).order_by('enrollment__user__id')
+
+    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 (user, user_schedules) in groupby(schedules, lambda s: s.enrollment.user):
+        user_schedules = list(user_schedules)
+        course_id_strs = [str(schedule.enrollment.course_id) for schedule in user_schedules]
+
+        first_schedule = user_schedules[0]
         template_context = get_base_template_context()
         template_context.update({
             'student_name': user.profile.name,
@@ -140,3 +229,93 @@ def _recurring_nudge_schedules_for_hour(target_hour, org_list, exclude_orgs=Fals
             'course_ids': course_id_strs,
         })
         yield (user, first_schedule.enrollment.course.language, template_context)
+
+
+class UpgradeReminder(ScheduleMessageType):
+    def __init__(self, day, *args, **kwargs):
+        super(UpgradeReminder, self).__init__(*args, **kwargs)
+        self.name = "upgradereminder".format(day)
+
+
+@task(ignore_result=True, routing_key=ROUTING_KEY)
+def upgrade_reminder_schedule_bin(
+    site_id, target_day_str, day_offset, bin_num, org_list, exclude_orgs=False, override_recipient_email=None,
+):
+    target_day = deserialize(target_day_str)
+    msg_type = UpgradeReminder(abs(day_offset))
+
+    for (user, language, context) in _upgrade_reminder_schedules_for_bin(target_day, bin_num, org_list, exclude_orgs):
+        msg = msg_type.personalize(
+            Recipient(
+                user.username,
+                override_recipient_email or user.email,
+            ),
+            language,
+            context,
+        )
+        _upgrade_reminder_schedule_send.apply_async((site_id, str(msg)), retry=False)
+
+
+@task(ignore_result=True, routing_key=ROUTING_KEY)
+def _upgrade_reminder_schedule_send(site_id, msg_str):
+    site = Site.objects.get(pk=site_id)
+    if not ScheduleConfig.current(site).deliver_upgrade_reminder:
+        return
+
+    msg = Message.from_string(msg_str)
+    ace.send(msg)
+
+
+def _upgrade_reminder_schedules_for_bin(target_day, bin_num, org_list, exclude_orgs=False):
+    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'
+        ),
+    ).annotate(
+        id_mod=F('enrollment__user__id') % UPGRADE_REMINDER_NUM_BINS
+    ).filter(
+        id_mod=bin_num
+    ).filter(
+        upgrade_deadline__year=target_day.year,
+        upgrade_deadline__month=target_day.month,
+        upgrade_deadline__day=target_day.day,
+    )
+
+    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
+
+        # TODO: group by schedule and user like recurring nudge
+        course_id_strs = [course_id_str]
+        first_schedule = schedule
+
+        template_context = get_base_template_context()
+        template_context.update({
+            'student_name': user.profile.name,
+            'user_personal_address': user.profile.name if user.profile.name else user.username,
+            'user_schedule_upgrade_deadline_time': schedule.upgrade_deadline,
+
+            'course_name': first_schedule.enrollment.course.display_name,
+            'course_url': absolute_url(reverse('course_root', args=[str(first_schedule.enrollment.course_id)])),
+
+            # This is used by the bulk email optout policy
+            'course_ids': course_id_strs,
+        })
+
+        yield (user, course.language, template_context)

openedx/core/djangoapps/schedules/templates/schedules/edx_ace/upgradereminder/email/body.html

+{% extends 'schedules/edx_ace/common/base_body.html' %}
+{% load i18n %}
+
+{% block preview_text %}
+    {% if courses|length > 1 %}
+        {% blocktrans trimmed %}
+            We hope you are enjoying {{ course_name }}, and other courses on edX.org.
+            Upgrade by {{ user_schedule_upgrade_deadline_time|date:"l, F dS, Y" }} to get a shareable certificate!
+        {% endblocktrans %}
+    {% else %}
+        {% blocktrans trimmed %}
+            We hope you are enjoying {{ course_name }}.
+            Upgrade by {{ user_schedule_upgrade_deadline_time|date:"l, F dS, Y" }} to get a shareable certificate!
+        {% endblocktrans %}
+    {% endif %}
+{% endblock %}
+
+{% block content %}
+<table width="100%" align="left" border="0" cellpadding="0" cellspacing="0" role="presentation">
+    <tr>
+        <td>
+            <h1>{% trans "Upgrade now" %}</h1>
+
+            <p>
+                {% if courses|length > 1 %}
+                    {% blocktrans trimmed %}
+                        We hope you are enjoying <strong>{{ course_name }}</strong>, and other courses on edX.org.
+                        {{ user_schedule_upgrade_deadline_time|date }}
+                        Upgrade by {{ user_schedule_upgrade_deadline_time|date:"l, F dS, Y" }} to get a shareable certificate!
+                    {% endblocktrans %}
+                {% else %}
+                    {% blocktrans trimmed %}
+                        We hope you are enjoying <strong>{{ course_name }}</strong>.
+                        Upgrade by {{ user_schedule_upgrade_deadline_time|date:"l, F dS, Y" }} to get a shareable certificate!
+                    {% endblocktrans %}
+                {% endif %}
+            </p>
+
+            <p>
+                <!-- email client support for style sheets is pretty spotty, so we have to inline all of these styles -->
+                <a
+                {% if courses|length > 1 %}
+                    href="{{ dashboard_url }}"
+                {% else %}
+                    href="{{ course_url }}"
+                {% endif %}
+                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;
+                ">
+                    <!-- old email clients require the use of the font tag :( -->
+                    <font color="#ffffff"><b>{% trans "Upgrade now" %}</b></font>
+                </a>
+            </p>
+        </td>
+    </tr>
+</table>
+{% endblock %}

openedx/core/djangoapps/schedules/templates/schedules/edx_ace/upgradereminder/email/body.txt

+{% load i18n %}
+
+{% blocktrans trimmed %}
+Dear {{ user_personal_address }},
+{% endblocktrans %}
+
+{% if courses|length > 1 %}
+{% blocktrans trimmed %}
+    We hope you are enjoying {{ course_name }}, and other courses on edX.org.
+    Upgrade by {{ user_schedule_upgrade_deadline_time|date:"l, F dS, Y" }} to get a shareable certificate!
+{% endblocktrans %}
+
+{% trans "Upgrade now at" %} <{{ dashboard_url }}>
+{% else %}
+{% blocktrans trimmed %}
+    We hope you are enjoying {{ course_name }}.
+    Upgrade by {{ user_schedule_upgrade_deadline_time|date:"l, F dS, Y" }} to get a shareable certificate!
+{% endblocktrans %}
+
+{% trans "Upgrade now at" %} <{{ course_url }}>
+{% endif %}

openedx/core/djangoapps/schedules/templates/schedules/edx_ace/upgradereminder/email/from_name.txt

+{% if courses|length > 1 %}
+{{ platform_name }}
+{% else %}
+{{ course_name }}
+{% endif %}

openedx/core/djangoapps/schedules/templates/schedules/edx_ace/upgradereminder/email/head.html

+{% extends 'schedules/edx_ace/common/base_head.html' %}

openedx/core/djangoapps/schedules/templates/schedules/edx_ace/upgradereminder/email/subject.txt

+{% load i18n %}
+
+{% if courses|length > 1 %}
+{% blocktrans %}Only two days left to upgrade on {{ platform_name }}!{% endblocktrans %}
+{% else %}
+{% blocktrans %}Only two days left to upgrade in {{course_name}} !{% endblocktrans %}
+{% endif %}

openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/body.html

-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>

openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/body.txt

-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}}

openedx/core/djangoapps/schedules/templates/schedules/edx_ace/verifiedupgradedeadlinereminder/email/subject.txt

-Only two days left to upgrade!
\ No newline at end of file

openedx/core/djangoapps/site_configuration/tests/factories.py

@@ -3,7 +3,7 @@ Model factories for unit testing views or models.
 """
 from django.contrib.sites.models import Site
 from factory.django import DjangoModelFactory
-from factory import SubFactory, Sequence, SelfAttribute
+from factory import SubFactory, Sequence, SelfAttribute, lazy_attribute
 
 from openedx.core.djangoapps.site_configuration.models import SiteConfiguration
 
@@ -27,6 +27,9 @@ class SiteConfigurationFactory(DjangoModelFactory):
     class Meta(object):
         model = SiteConfiguration
 
-    values = {}
     enabled = True
     site = SubFactory(SiteFactory)
+
+    @lazy_attribute
+    def values(self):
+        return {}