Use separate retry count for calculating retry delay.

lms/djangoapps/bulk_email/tests/test_err_handling.py

@@ -4,7 +4,6 @@ Unit tests for handling email sending errors
 from itertools import cycle
 from mock import patch, Mock
 from smtplib import SMTPDataError, SMTPServerDisconnected, SMTPConnectError
-from unittest import skip
 
 from django.test.utils import override_settings
 from django.conf import settings
@@ -93,9 +92,9 @@ class TestEmailErrors(ModuleStoreTestCase):
         # Test that after the rejected email, the rest still successfully send
         ((_initial_results), kwargs) = result.call_args
         self.assertEquals(kwargs['skipped'], 0)
-        expectedNumFails = int((settings.EMAILS_PER_TASK + 3) / 4.0)
-        self.assertEquals(kwargs['failed'], expectedNumFails)
-        self.assertEquals(kwargs['succeeded'], settings.EMAILS_PER_TASK - expectedNumFails)
+        expected_fails = int((settings.EMAILS_PER_TASK + 3) / 4.0)
+        self.assertEquals(kwargs['failed'], expected_fails)
+        self.assertEquals(kwargs['succeeded'], settings.EMAILS_PER_TASK - expected_fails)
 
     @patch('bulk_email.tasks.get_connection', autospec=True)
     @patch('bulk_email.tasks.send_course_email.retry')
@@ -144,7 +143,7 @@ class TestEmailErrors(ModuleStoreTestCase):
     @patch('bulk_email.tasks.get_connection', Mock(return_value=EmailTestException))
     def test_general_exception(self, mock_log, retry, result):
         """
-        Tests the if the error is not SMTP-related, we log and reraise
+        Tests the if the error is unexpected, we log and retry
         """
         test_email = {
             'action': 'Send email',
@@ -156,11 +155,10 @@ class TestEmailErrors(ModuleStoreTestCase):
         # so we assert on the arguments of log.exception
         self.client.post(self.url, test_email)
         self.assertTrue(mock_log.exception.called)
-        ((log_str, _task_id, email_id, to_list), _) = mock_log.exception.call_args
-        self.assertIn('caused send_course_email task to fail with uncaught exception.', log_str)
+        ((log_str, _task_id, email_id), _) = mock_log.exception.call_args
+        self.assertIn('caused send_course_email task to fail with unexpected exception.', log_str)
         self.assertEqual(email_id, 1)
-        self.assertEqual(to_list, [self.instructor.email])
-        self.assertFalse(retry.called)
+        self.assertTrue(retry.called)
         # check the results being returned
         self.assertTrue(result.called)
         ((initial_results, ), kwargs) = result.call_args
@@ -180,7 +178,7 @@ class TestEmailErrors(ModuleStoreTestCase):
         entry = InstructorTask.create(course_id, "task_type", "task_key", "task_input", self.instructor)
         task_input = {"email_id": -1}
         with self.assertRaises(CourseEmail.DoesNotExist):
-            perform_delegate_email_batches(entry.id, course_id, task_input, "action_name")
+            perform_delegate_email_batches(entry.id, course_id, task_input, "action_name")  # pylint: disable=E1101
         ((log_str, _, email_id), _) = mock_log.warning.call_args
         self.assertTrue(mock_log.warning.called)
         self.assertIn('Failed to get CourseEmail with id', log_str)
@@ -196,9 +194,9 @@ class TestEmailErrors(ModuleStoreTestCase):
         email = CourseEmail(course_id=course_id)
         email.save()
         entry = InstructorTask.create(course_id, "task_type", "task_key", "task_input", self.instructor)
-        task_input = {"email_id": email.id}
+        task_input = {"email_id": email.id}  # pylint: disable=E1101
         with self.assertRaises(Exception):
-            perform_delegate_email_batches(entry.id, course_id, task_input, "action_name")
+            perform_delegate_email_batches(entry.id, course_id, task_input, "action_name")  # pylint: disable=E1101
         ((log_str, _, _), _) = mock_log.exception.call_args
         self.assertTrue(mock_log.exception.called)
         self.assertIn('get_course_by_id failed:', log_str)
@@ -211,9 +209,9 @@ class TestEmailErrors(ModuleStoreTestCase):
         email = CourseEmail(course_id=self.course.id, to_option="IDONTEXIST")
         email.save()
         entry = InstructorTask.create(self.course.id, "task_type", "task_key", "task_input", self.instructor)
-        task_input = {"email_id": email.id}
+        task_input = {"email_id": email.id}  # pylint: disable=E1101
         with self.assertRaises(Exception):
-            perform_delegate_email_batches(entry.id, self.course.id, task_input, "action_name")
+            perform_delegate_email_batches(entry.id, self.course.id, task_input, "action_name")  # pylint: disable=E1101
         ((log_str, opt_str), _) = mock_log.error.call_args
         self.assertTrue(mock_log.error.called)
         self.assertIn('Unexpected bulk email TO_OPTION found', log_str)

lms/djangoapps/bulk_email/tests/test_tasks.py

+"""
+Unit tests for LMS instructor-initiated background tasks.
+
+Runs tasks on answers to course problems to validate that code
+paths actually work.
+
+"""
+import json
+from uuid import uuid4
+from itertools import cycle
+from mock import patch, Mock
+from smtplib import SMTPDataError, SMTPServerDisconnected
+
+from celery.states import SUCCESS
+
+# from django.test.utils import override_settings
+from django.conf import settings
+from django.core.management import call_command
+
+from bulk_email.models import CourseEmail, SEND_TO_ALL
+
+# from instructor_task.tests.test_tasks import TestInstructorTasks
+from instructor_task.tasks import send_bulk_course_email
+from instructor_task.models import InstructorTask
+from instructor_task.tests.test_base import InstructorTaskCourseTestCase
+from instructor_task.tests.factories import InstructorTaskFactory
+
+
+class TestTaskFailure(Exception):
+    """Dummy exception used for unit tests."""
+    pass
+
+
+class TestBulkEmailInstructorTask(InstructorTaskCourseTestCase):
+    """Tests instructor task that send bulk email."""
+
+    def setUp(self):
+        super(TestBulkEmailInstructorTask, self).setUp()
+        self.initialize_course()
+        self.instructor = self.create_instructor('instructor')
+
+        # load initial content (since we don't run migrations as part of tests):
+        call_command("loaddata", "course_email_template.json")
+
+    def _create_input_entry(self, course_id=None):
+        """
+        Creates a InstructorTask entry for testing.
+
+        Overrides the base class version in that this creates CourseEmail.
+        """
+        to_option = SEND_TO_ALL
+        course_id = course_id or self.course.id
+        course_email = CourseEmail.create(course_id, self.instructor, to_option, "Test Subject", "<p>This is a test message</p>")
+        task_input = {'email_id': course_email.id}
+        task_id = str(uuid4())
+        instructor_task = InstructorTaskFactory.create(
+            course_id=course_id,
+            requester=self.instructor,
+            task_input=json.dumps(task_input),
+            task_key='dummy value',
+            task_id=task_id,
+        )
+        return instructor_task
+
+    def _run_task_with_mock_celery(self, task_class, entry_id, task_id, expected_failure_message=None):
+        """Submit a task and mock how celery provides a current_task."""
+        self.current_task = Mock()
+        self.current_task.max_retries = settings.BULK_EMAIL_MAX_RETRIES
+        self.current_task.default_retry_delay = settings.BULK_EMAIL_DEFAULT_RETRY_DELAY
+        task_args = [entry_id, {}]
+
+        with patch('bulk_email.tasks._get_current_task') as mock_get_task:
+            mock_get_task.return_value = self.current_task
+            return task_class.apply(task_args, task_id=task_id).get()
+
+    def test_email_missing_current_task(self):
+        task_entry = self._create_input_entry()
+        with self.assertRaises(ValueError):
+            send_bulk_course_email(task_entry.id, {})
+
+    def test_email_undefined_course(self):
+        # Check that we fail when passing in a course that doesn't exist.
+        task_entry = self._create_input_entry(course_id="bogus/course/id")
+        with self.assertRaises(ValueError):
+            self._run_task_with_mock_celery(send_bulk_course_email, task_entry.id, task_entry.task_id)
+
+    def _create_students(self, num_students):
+        """Create students, a problem, and StudentModule objects for testing"""
+        students = [
+            self.create_student('robot%d' % i) for i in xrange(num_students)
+        ]
+        return students
+
+    def _test_run_with_task(self, task_class, action_name, total, succeeded, failed=0, skipped=0):
+        """Run a task and check the number of emails processed."""
+        task_entry = self._create_input_entry()
+        parent_status = self._run_task_with_mock_celery(task_class, task_entry.id, task_entry.task_id)
+        # check return value
+        self.assertEquals(parent_status.get('total'), total)
+        self.assertEquals(parent_status.get('action_name'), action_name)
+        # compare with entry in table:
+        entry = InstructorTask.objects.get(id=task_entry.id)
+        status = json.loads(entry.task_output)
+        self.assertEquals(status.get('attempted'), succeeded + failed)
+        self.assertEquals(status.get('succeeded'), succeeded)
+        self.assertEquals(status['skipped'], skipped)
+        self.assertEquals(status['failed'], failed)
+        self.assertEquals(status.get('total'), total)
+        self.assertEquals(status.get('action_name'), action_name)
+        self.assertGreater(status.get('duration_ms'), 0)
+        self.assertEquals(entry.task_state, SUCCESS)
+
+    def test_successful(self):
+        num_students = settings.EMAILS_PER_TASK
+        self._create_students(num_students)
+        # we also send email to the instructor:
+        num_emails = num_students + 1
+        with patch('bulk_email.tasks.get_connection', autospec=True) as get_conn:
+            get_conn.return_value.send_messages.side_effect = cycle([None])
+            self._test_run_with_task(send_bulk_course_email, 'emailed', num_emails, num_emails)
+
+    def test_data_err_fail(self):
+        # Test that celery handles permanent SMTPDataErrors by failing and not retrying.
+        num_students = settings.EMAILS_PER_TASK
+        self._create_students(num_students)
+        # we also send email to the instructor:
+        num_emails = num_students + 1
+        expected_fails = int((num_emails + 3) / 4.0)
+        expected_succeeds = num_emails - expected_fails
+        with patch('bulk_email.tasks.get_connection', autospec=True) as get_conn:
+            # have every fourth email fail due to blacklisting:
+            get_conn.return_value.send_messages.side_effect = cycle([SMTPDataError(554, "Email address is blacklisted"),
+                                                                     None, None, None])
+            self._test_run_with_task(send_bulk_course_email, 'emailed', num_emails, expected_succeeds, failed=expected_fails)
+
+    def test_retry_after_limited_retry_error(self):
+        # Test that celery handles connection failures by retrying.
+        num_students = 1
+        self._create_students(num_students)
+        # we also send email to the instructor:
+        num_emails = num_students + 1
+        expected_fails = 0
+        expected_succeeds = num_emails
+        with patch('bulk_email.tasks.get_connection', autospec=True) as get_conn:
+            # have every other mail attempt fail due to disconnection:
+            get_conn.return_value.send_messages.side_effect = cycle([SMTPServerDisconnected(425, "Disconnecting"), None])
+            self._test_run_with_task(send_bulk_course_email, 'emailed', num_emails, expected_succeeds, failed=expected_fails)
+
+    def test_max_retry(self):
+        # Test that celery can hit a maximum number of retries.
+        num_students = 1
+        self._create_students(num_students)
+        # we also send email to the instructor:
+        num_emails = num_students + 1
+        # This is an ugly hack:  the failures that are reported by the EAGER version of retry
+        # are multiplied by the attempted number of retries (equals max plus one).
+        expected_fails = num_emails * (settings.BULK_EMAIL_MAX_RETRIES + 1)
+        expected_succeeds = 0
+        with patch('bulk_email.tasks.get_connection', autospec=True) as get_conn:
+            # always fail to connect, triggering repeated retries until limit is hit:
+            get_conn.return_value.send_messages.side_effect = cycle([SMTPServerDisconnected(425, "Disconnecting")])
+            self._test_run_with_task(send_bulk_course_email, 'emailed', num_emails, expected_succeeds, failed=expected_fails)

lms/djangoapps/instructor/views/legacy.py

@@ -720,18 +720,13 @@ def instructor_dashboard(request, course_id):
         email_subject = request.POST.get("subject")
         html_message = request.POST.get("message")
 
-        # TODO: make sure this is committed before submitting it to the task.
-        # However, it should probably be enough to do the submit below, which
-        # will commit the transaction for the InstructorTask object.  Both should
-        # therefore be committed.  (Still, it might be clearer to do so here as well.)
-        # Actually, this should probably be moved out, so that all the validation logic
-        # we might want to add to it can be added.  There might also be something
-        # that would permit validation of the email beforehand.
+        # Create the CourseEmail object.  This is saved immediately, so that
+        # any transaction that has been pending up to this point will also be
+        # committed.
         email = CourseEmail.create(course_id, request.user, email_to_option, email_subject, html_message)
 
-        # TODO: make this into a task submission, so that the correct
-        # InstructorTask object gets created (for monitoring purposes)
-        submit_bulk_course_email(request, course_id, email.id)
+        # Submit the task, so that the correct InstructorTask object gets created (for monitoring purposes)
+        submit_bulk_course_email(request, course_id, email.id)  # pylint: disable=E1101
 
         if email_to_option == "all":
             email_msg = '<div class="msg msg-confirm"><p class="copy">Your email was successfully queued for sending. Please note that for large public classes (~10k), it may take 1-2 hours to send all emails.</p></div>'
@@ -1535,7 +1530,6 @@ def get_background_task_table(course_id, problem_url=None, student=None, task_ty
     # (note that we don't have to check that the arguments are valid; it
     # just won't find any entries.)
     if (history_entries.count()) == 0:
-        # TODO: figure out how to deal with task_type better here...
         if problem_url is None:
             msg += '<font color="red">Failed to find any background tasks for course "{course}".</font>'.format(course=course_id)
         elif student is not None:

lms/djangoapps/instructor_task/api.py

@@ -178,8 +178,8 @@ def submit_bulk_course_email(request, course_id, email_id):
     The specified CourseEmail object will be sent be updated for all students who have enrolled
     in a course.  Parameters are the `course_id` and the `email_id`, the id of the CourseEmail object.
 
-    AlreadyRunningError is raised if the course's students are already being emailed.
-    TODO: is this the right behavior?  Or should multiple emails be allowed in the pipeline at the same time?
+    AlreadyRunningError is raised if the same recipients are already being emailed with the same
+    CourseEmail object.
 
     This method makes sure the InstructorTask entry is committed.
     When called from any view that is wrapped by TransactionMiddleware,
@@ -188,11 +188,9 @@ def submit_bulk_course_email(request, course_id, email_id):
     save here.  Any future database operations will take place in a
     separate transaction.
     """
-    # check arguments:  make sure that the course is defined?
-    # TODO: what is the right test here?
-
-    # This should also make sure that the email exists.
-    # We can also pull out the To argument here, so that is displayed in
+    # Assume that the course is defined, and that the user has already been verified to have
+    # appropriate access to the course. But make sure that the email exists.
+    # We also pull out the To argument here, so that is displayed in
     # the InstructorTask status.
     email_obj = CourseEmail.objects.get(id=email_id)
     to_option = email_obj.to_option

lms/djangoapps/instructor_task/api_helper.py

@@ -268,7 +268,7 @@ def submit_task(request, task_type, task_class, course_id, task_input, task_key)
 
     # submit task:
     task_id = instructor_task.task_id
-    task_args = [instructor_task.id, _get_xmodule_instance_args(request, task_id)]
+    task_args = [instructor_task.id, _get_xmodule_instance_args(request, task_id)]  # pylint: disable=E1101
     task_class.apply_async(task_args, task_id=task_id)
 
     return instructor_task

lms/djangoapps/instructor_task/tasks.py

@@ -32,7 +32,7 @@ from instructor_task.tasks_helper import (
 from bulk_email.tasks import perform_delegate_email_batches
 
 
-@task(base=BaseInstructorTask)
+@task(base=BaseInstructorTask)  # pylint: disable=E1102
 def rescore_problem(entry_id, xmodule_instance_args):
     """Rescores a problem in a course, for all students or one specific student.
 
@@ -55,13 +55,14 @@ def rescore_problem(entry_id, xmodule_instance_args):
     update_fcn = partial(rescore_problem_module_state, xmodule_instance_args)
 
     def filter_fcn(modules_to_update):
+        """Filter that matches problems which are marked as being done"""
         return modules_to_update.filter(state__contains='"done": true')
 
     visit_fcn = partial(perform_module_state_update, update_fcn, filter_fcn)
     return run_main_task(entry_id, visit_fcn, action_name)
 
 
-@task(base=BaseInstructorTask)
+@task(base=BaseInstructorTask)  # pylint: disable=E1102
 def reset_problem_attempts(entry_id, xmodule_instance_args):
     """Resets problem attempts to zero for a particular problem for all students in a course.
 
@@ -82,7 +83,7 @@ def reset_problem_attempts(entry_id, xmodule_instance_args):
     return run_main_task(entry_id, visit_fcn, action_name)
 
 
-@task(base=BaseInstructorTask)
+@task(base=BaseInstructorTask)  # pylint: disable=E1102
 def delete_problem_state(entry_id, xmodule_instance_args):
     """Deletes problem state entirely for all students on a particular problem in a course.
 
@@ -103,18 +104,20 @@ def delete_problem_state(entry_id, xmodule_instance_args):
     return run_main_task(entry_id, visit_fcn, action_name)
 
 
-@task(base=BaseInstructorTask)
-def send_bulk_course_email(entry_id, xmodule_instance_args):
-    """Sends emails to in a course.
+@task(base=BaseInstructorTask)  # pylint: disable=E1102
+def send_bulk_course_email(entry_id, _xmodule_instance_args):
+    """Sends emails to recipients enrolled in a course.
 
     `entry_id` is the id value of the InstructorTask entry that corresponds to this task.
     The entry contains the `course_id` that identifies the course, as well as the
     `task_input`, which contains task-specific input.
 
-    The task_input should be a dict with no entries.
+    The task_input should be a dict with the following entries:
 
-    `xmodule_instance_args` provides information needed by _get_module_instance_for_task()
-    to instantiate an xmodule instance.
+      'email_id': the full URL to the problem to be rescored.  (required)
+
+    `_xmodule_instance_args` provides information needed by _get_module_instance_for_task()
+    to instantiate an xmodule instance.  This is unused here.
     """
     action_name = 'emailed'
     visit_fcn = perform_delegate_email_batches

lms/djangoapps/instructor_task/tests/test_api.py

@@ -152,15 +152,16 @@ class InstructorTaskCourseSubmitTest(InstructorTaskCourseTestCase):
         self.instructor = UserFactory.create(username="instructor", email="instructor@edx.org")
 
     def _define_course_email(self):
+        """Create CourseEmail object for testing."""
         course_email = CourseEmail.create(self.course.id, self.instructor, SEND_TO_ALL, "Test Subject", "<p>This is a test message</p>")
-        return course_email.id
+        return course_email.id  # pylint: disable=E1101
 
     def test_submit_bulk_email_all(self):
         email_id = self._define_course_email()
         instructor_task = submit_bulk_course_email(self.create_task_request(self.instructor), self.course.id, email_id)
 
         # test resubmitting, by updating the existing record:
-        instructor_task = InstructorTask.objects.get(id=instructor_task.id)
+        instructor_task = InstructorTask.objects.get(id=instructor_task.id)  # pylint: disable=E1101
         instructor_task.task_state = PROGRESS
         instructor_task.save()
 

lms/djangoapps/instructor_task/tests/test_tasks.py

@@ -85,11 +85,11 @@ class TestInstructorTasks(InstructorTaskModuleTestCase):
     def _test_missing_current_task(self, task_class):
         """Check that a task_class fails when celery doesn't provide a current_task."""
         task_entry = self._create_input_entry()
-        with self.assertRaises(UpdateProblemModuleStateError):
+        with self.assertRaises(ValueError):
             task_class(task_entry.id, self._get_xmodule_instance_args())
 
     def _test_undefined_course(self, task_class):
-        # run with celery, but no course defined
+        """Run with celery, but with no course defined."""
         task_entry = self._create_input_entry(course_id="bogus/course/id")
         with self.assertRaises(ItemNotFoundError):
             self._run_task_with_mock_celery(task_class, task_entry.id, task_entry.task_id)

lms/envs/aws.py

@@ -93,6 +93,10 @@ CELERY_QUEUES = {
     DEFAULT_PRIORITY_QUEUE: {}
 }
 
+# We want Bulk Email running on the high-priority queue, so we define the
+# routing key that points to it.  At the moment, the name is the same.
+BULK_EMAIL_ROUTING_KEY = HIGH_PRIORITY_QUEUE
+
 ########################## NON-SECURE ENV CONFIG ##############################
 # Things like server locations, ports, etc.
 
@@ -128,7 +132,7 @@ LOG_DIR = ENV_TOKENS['LOG_DIR']
 
 CACHES = ENV_TOKENS['CACHES']
 
-#Email overrides
+# Email overrides
 DEFAULT_FROM_EMAIL = ENV_TOKENS.get('DEFAULT_FROM_EMAIL', DEFAULT_FROM_EMAIL)
 DEFAULT_FEEDBACK_EMAIL = ENV_TOKENS.get('DEFAULT_FEEDBACK_EMAIL', DEFAULT_FEEDBACK_EMAIL)
 DEFAULT_BULK_FROM_EMAIL = ENV_TOKENS.get('DEFAULT_BULK_FROM_EMAIL', DEFAULT_BULK_FROM_EMAIL)
@@ -140,8 +144,10 @@ BUGS_EMAIL = ENV_TOKENS.get('BUGS_EMAIL', BUGS_EMAIL)
 PAYMENT_SUPPORT_EMAIL = ENV_TOKENS.get('PAYMENT_SUPPORT_EMAIL', PAYMENT_SUPPORT_EMAIL)
 PAID_COURSE_REGISTRATION_CURRENCY = ENV_TOKENS.get('PAID_COURSE_REGISTRATION_CURRENCY',
                                                    PAID_COURSE_REGISTRATION_CURRENCY)
+BULK_EMAIL_DEFAULT_RETRY_DELAY = ENV_TOKENS.get('BULK_EMAIL_DEFAULT_RETRY_DELAY', BULK_EMAIL_DEFAULT_RETRY_DELAY)
+BULK_EMAIL_MAX_RETRIES = ENV_TOKENS.get('BULK_EMAIL_MAX_RETRIES', BULK_EMAIL_MAX_RETRIES)
 
-#Theme overrides
+# Theme overrides
 THEME_NAME = ENV_TOKENS.get('THEME_NAME', None)
 if not THEME_NAME is None:
     enable_theme(THEME_NAME)
@@ -150,10 +156,10 @@ if not THEME_NAME is None:
 # Marketing link overrides
 MKTG_URL_LINK_MAP.update(ENV_TOKENS.get('MKTG_URL_LINK_MAP', {}))
 
-#Timezone overrides
+# Timezone overrides
 TIME_ZONE = ENV_TOKENS.get('TIME_ZONE', TIME_ZONE)
 
-#Additional installed apps
+# Additional installed apps
 for app in ENV_TOKENS.get('ADDL_INSTALLED_APPS', []):
     INSTALLED_APPS += (app,)
 

lms/envs/common.py

@@ -795,6 +795,17 @@ CELERY_QUEUES = {
     DEFAULT_PRIORITY_QUEUE: {}
 }
 
+# let logging work as configured:
+CELERYD_HIJACK_ROOT_LOGGER = False
+
+################################ Bulk Email ###################################
+
+# We want Bulk Email running on the high-priority queue, so we define the
+# routing key that points to it.  At the moment, the name is the same.
+BULK_EMAIL_ROUTING_KEY = HIGH_PRIORITY_QUEUE
+BULK_EMAIL_DEFAULT_RETRY_DELAY = 15
+BULK_EMAIL_MAX_RETRIES = 5
+
 ################################### APPS ######################################
 INSTALLED_APPS = (
     # Standard ones that are always installed...