Bin by course with failing tests

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

-from collections import namedtuple, defaultdict
-from copy import deepcopy
 import datetime
-import ddt
 import logging
+from collections import defaultdict, namedtuple
+from copy import deepcopy
+from zlib import crc32
+
+import ddt
 
 import attr
 from django.conf import settings
@@ -69,6 +71,7 @@ class ScheduleSendEmailTestBase(FilteredQueryCountMixin, CacheIsolationTestCase)
 
     queries_deadline_for_each_course = False
     consolidates_emails_for_learner = False
+    bin_by = 'user'
 
     def setUp(self):
         super(ScheduleSendEmailTestBase, self).setUp()
@@ -82,8 +85,11 @@ class ScheduleSendEmailTestBase(FilteredQueryCountMixin, CacheIsolationTestCase)
 
         self._courses_with_verified_modes = set()
 
-    def _calculate_bin_for_user(self, user):
-        return user.id % self.task.num_bins
+    def _calculate_bin(self, user, course_key_str):
+        if self.bin_by == 'user':
+            return user.id % self.task.num_bins
+        elif self.bin_by == 'course':
+            return crc32(course_key_str) % self.task.num_bins
 
     def _get_dates(self, offset=None):
         current_day = _get_datetime_beginning_of_day(datetime.datetime.now(pytz.UTC))
@@ -159,7 +165,8 @@ class ScheduleSendEmailTestBase(FilteredQueryCountMixin, CacheIsolationTestCase)
                 self._schedule_factory() for _ in range(schedule_count)
             ]
 
-            bins_in_use = frozenset((self._calculate_bin_for_user(s.enrollment.user)) for s in schedules)
+            bins_in_use = frozenset((self._calculate_bin(s.enrollment.user, unicode(s.enrollment.course.id)))
+                                    for s in schedules)
             is_first_match = True
             target_day_str = serialize(target_day)
 
@@ -331,10 +338,10 @@ class ScheduleSendEmailTestBase(FilteredQueryCountMixin, CacheIsolationTestCase)
             with patch.object(self.task, 'async_send_task') as mock_schedule_send:
                 self.task.apply(kwargs=dict(
                     site_id=self.site_config.site.id, target_day_str=serialize(target_day), day_offset=offset,
-                    bin_num=self._calculate_bin_for_user(user),
+                    bin_num=self._calculate_bin(user, 'edX/toy/course0'),
                 ))
 
-        expected_call_count = 1 if self.consolidates_emails_for_learner else num_courses
+        expected_call_count = 1 if self.consolidates_emails_for_learner or self.bin_by == 'course' else num_courses
         self.assertEqual(mock_schedule_send.apply_async.call_count, expected_call_count)
         self.assertFalse(mock_ace.send.called)
 
@@ -379,9 +386,10 @@ class ScheduleSendEmailTestBase(FilteredQueryCountMixin, CacheIsolationTestCase)
                 with self.assertNumQueries(num_expected_queries, table_blacklist=WAFFLE_TABLES):
                     self.task.apply(kwargs=dict(
                         site_id=self.site_config.site.id, target_day_str=serialize(target_day), day_offset=offset,
-                        bin_num=self._calculate_bin_for_user(user),
+                        bin_num=self._calculate_bin(user, 'edX/toy/course0'),
                     ))
-            num_expected_messages = 1 if self.consolidates_emails_for_learner else message_count
+            num_expected_messages = (1 if self.consolidates_emails_for_learner or self.bin_by == 'course' else
+                                     message_count)
             self.assertEqual(len(sent_messages), num_expected_messages)
 
             with self.assertNumQueries(2):
@@ -409,7 +417,7 @@ class ScheduleSendEmailTestBase(FilteredQueryCountMixin, CacheIsolationTestCase)
         with patch.object(tasks, 'ace') as mock_ace:
             self.task.apply(kwargs=dict(
                 site_id=self.site_config.site.id, target_day_str=serialize(target_day), day_offset=offset,
-                bin_num=self._calculate_bin_for_user(schedule.enrollment.user),
+                bin_num=self._calculate_bin(schedule.enrollment.user, unicode(schedule.enrollment.course.id)),
             ))
 
             self.assertEqual(mock_ace.send.called, test_config.email_sent)

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

@@ -35,6 +35,7 @@ class TestSendCourseUpdate(ScheduleUpsellTestMixin, ScheduleSendEmailTestBase):
     experience_type = ScheduleExperience.EXPERIENCES.course_updates
 
     queries_deadline_for_each_course = True
+    bin_by = 'course'
 
     def setUp(self):
         super(TestSendCourseUpdate, self).setUp()

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

@@ -49,7 +49,7 @@ class TestUpgradeReminder(ScheduleSendEmailTestBase):
 
         self.task.apply(kwargs=dict(
             site_id=self.site_config.site.id, target_day_str=serialize(target_day), day_offset=offset,
-            bin_num=self._calculate_bin_for_user(schedule.enrollment.user),
+            bin_num=self._calculate_bin(schedule.enrollment.user, unicode(schedule.enrollment.course.id)),
         ))
 
         self.assertEqual(mock_ace.send.called, not is_verified)
@@ -73,7 +73,7 @@ class TestUpgradeReminder(ScheduleSendEmailTestBase):
 
             self.task.apply(kwargs=dict(
                 site_id=self.site_config.site.id, target_day_str=serialize(target_day), day_offset=offset,
-                bin_num=self._calculate_bin_for_user(user),
+                bin_num=self._calculate_bin(user, 'edX/toy/Course0'),
             ))
 
             messages = [Message.from_string(m) for m in sent_messages]
@@ -92,7 +92,7 @@ class TestUpgradeReminder(ScheduleSendEmailTestBase):
 
         self.task.apply(kwargs=dict(
             site_id=self.site_config.site.id, target_day_str=serialize(target_day), day_offset=offset,
-            bin_num=self._calculate_bin_for_user(schedule.enrollment.user),
+            bin_num=self._calculate_bin(schedule.enrollment.user, unicode(schedule.enrollment.course.id)),
         ))
         self.assertEqual(mock_ace.send.called, False)
 

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

@@ -42,7 +42,7 @@ class ScheduleUpsellTestMixin(object):
             mock_schedule_send.apply_async = lambda args, *_a, **_kw: sent_messages.append(args[1])
             self.task.apply(kwargs=dict(
                 site_id=self.site_config.site.id, target_day_str=serialize(target_day), day_offset=offset,
-                bin_num=self._calculate_bin_for_user(schedule.enrollment.user),
+                bin_num=self._calculate_bin(schedule.enrollment.user, unicode(schedule.enrollment.course.id)),
             ))
         self.assertEqual(len(sent_messages), 1)
 

openedx/core/djangoapps/schedules/resolvers.py

@@ -7,7 +7,7 @@ from django.conf import settings
 from django.contrib.auth.models import User
 from django.contrib.staticfiles.templatetags.staticfiles import static
 from django.core.urlresolvers import reverse
-from django.db.models import F, Q
+from django.db.models import ExpressionWrapper, F, Func, IntegerField, Q
 from django.utils.formats import dateformat, get_format
 
 
@@ -98,7 +98,7 @@ class BinnedSchedulesBaseResolver(PrefixedDebugLoggerMixin, RecipientResolver):
                 self.async_send_task.apply_async((self.site.id, str(msg)), retry=False)
 
     def get_schedules_with_target_date_by_bin_and_orgs(
-        self, order_by='enrollment__user__id'
+        self, order_by='enrollment__user__id', bin_by='user'
     ):
         """
         Returns Schedules with the target_date, related to Users whose id matches the bin_num, and filtered by org_list.
@@ -111,15 +111,19 @@ class BinnedSchedulesBaseResolver(PrefixedDebugLoggerMixin, RecipientResolver):
             'courseenrollment__schedule__{}__gte'.format(self.schedule_date_field): target_day,
             'courseenrollment__schedule__{}__lt'.format(self.schedule_date_field): target_day + datetime.timedelta(days=1),
         }
+
         users = User.objects.filter(
             courseenrollment__is_active=True,
             **schedule_day_equals_target_day_filter
-        ).annotate(
-            id_mod=F('id') % self.num_bins
-        ).filter(
-            id_mod=self.bin_num
         )
 
+        if bin_by == 'user':
+            users = users.annotate(
+                id_mod=F('id') % self.num_bins
+            ).filter(
+                id_mod=self.bin_num
+            )
+
         schedule_day_equals_target_day_filter = {
             '{}__gte'.format(self.schedule_date_field): target_day,
             '{}__lt'.format(self.schedule_date_field): target_day + datetime.timedelta(days=1),
@@ -134,7 +138,18 @@ class BinnedSchedulesBaseResolver(PrefixedDebugLoggerMixin, RecipientResolver):
             enrollment__user__in=users,
             enrollment__is_active=True,
             **schedule_day_equals_target_day_filter
-        ).order_by(order_by)
+        )
+
+        if bin_by == 'course':
+            schedules = schedules.annotate(
+                hashed_course_id=ExpressionWrapper(
+                    Func(F('enrollment__course_id'), function='CRC32') % self.num_bins,
+                    output_field=IntegerField())
+            ).filter(
+                hashed_course_id=self.bin_num
+            )
+
+        schedules = schedules.order_by(order_by)
 
         schedules = self.filter_by_org(schedules)
 
@@ -353,6 +368,7 @@ class CourseUpdateResolver(BinnedSchedulesBaseResolver):
         week_num = abs(self.day_offset) / 7
         schedules = self.get_schedules_with_target_date_by_bin_and_orgs(
             order_by='enrollment__course',
+            bin_by='course',
         )
 
         template_context = get_base_template_context(self.site)