Merge pull request #15141 from edx/adam/clear-tasks-old-queueing-state

write management command to fail old tasks stuck in queueing state

lms/djangoapps/instructor_task/management/commands/fail_old_queueing_tasks.py

+from __future__ import unicode_literals, print_function
+
+from datetime import datetime
+
+from celery.states import FAILURE
+from django.core.management.base import BaseCommand, CommandError
+from pytz import utc
+
+from lms.djangoapps.instructor_task.models import InstructorTask, QUEUING
+
+
+class Command(BaseCommand):
+    """
+    Command to manually fail old "QUEUING" tasks in the instructor task table.
+
+    Example:
+    ./manage.py lms fail_old_queueing_tasks --dry-run --after 2001-01-03 \
+        --before 2001-01-06 --task-type bulk_course_email
+    """
+
+    def add_arguments(self, parser):
+        """
+        Add arguments to the command parser.
+        """
+        parser.add_argument(
+            '--before',
+            type=str,
+            dest='before',
+            help='Manually fail instructor tasks created before or on this date.',
+        )
+
+        parser.add_argument(
+            '--after',
+            type=str,
+            dest='after',
+            help='Manually fail instructor tasks created after or on this date.',
+        )
+
+        parser.add_argument(
+            '--dry-run',
+            action='store_true',
+            dest='dry_run',
+            default=False,
+            help='Return the records this command will update without updating them.',
+        )
+
+        parser.add_argument(
+            '--task-type',
+            dest='task_type',
+            type=str,
+            default=None,
+            help='Specify the type of task that you want to fail.',
+        )
+
+    @staticmethod
+    def parse_date(date_string):
+        """
+        Converts an isoformat string into a python datetime object. Localizes
+        that datetime object to UTC.
+        """
+        return utc.localize(datetime.strptime(date_string, "%Y-%m-%d"))
+
+    def handle(self, *args, **options):
+
+        if options['before'] is None:
+            raise CommandError("Must provide a 'before' date")
+
+        if options['after'] is None:
+            raise CommandError("Must provide an 'after' date")
+
+        before = self.parse_date(options['before'])
+        after = self.parse_date(options['after'])
+        filter_kwargs = {
+            "task_state": QUEUING,
+            "created__lte": before,
+            "created__gte": after,
+        }
+        if options['task_type'] is not None:
+            filter_kwargs.update({"task_type": options['task_type']})
+
+        tasks = InstructorTask.objects.filter(**filter_kwargs)
+
+        for task in tasks:
+            print(
+                "Queueing task '{task_id}', of type '{task_type}', created on '{created}', will be marked as 'FAILURE'".format(
+                    task_id=task.task_id,
+                    task_type=task.task_type,
+                    created=task.created,
+                )
+            )
+
+        if not options['dry_run']:
+            tasks_updated = tasks.update(
+                task_state=FAILURE,
+            )
+            print("{tasks_updated} records updated.".format(
+                tasks_updated=tasks_updated)
+            )
+        else:
+            print("This was a dry run, so no records were updated.")

lms/djangoapps/instructor_task/management/commands/tests/test_fail_old_queueing_tasks.py

+from datetime import datetime
+
+import ddt
+from celery.states import FAILURE
+from django.core.management import call_command
+from django.core.management.base import CommandError
+
+from lms.djangoapps.instructor_task.models import InstructorTask, QUEUING
+from lms.djangoapps.instructor_task.tests.factories import InstructorTaskFactory
+from lms.djangoapps.instructor_task.tests.test_base import InstructorTaskTestCase
+
+
+@ddt.ddt
+class TestFailOldQueueingTasksCommand(InstructorTaskTestCase):
+    """
+    Tests for the `fail_old_queueing_tasks` management command
+    """
+
+    def setUp(self):
+        super(TestFailOldQueueingTasksCommand, self).setUp()
+
+        type_1_queueing = InstructorTaskFactory.create(
+            task_state=QUEUING,
+            task_type="type_1",
+            task_key='',
+            task_id=1,
+        )
+        type_1_non_queueing = InstructorTaskFactory.create(
+            task_state='NOT QUEUEING',
+            task_type="type_1",
+            task_key='',
+            task_id=2,
+        )
+
+        type_2_queueing = InstructorTaskFactory.create(
+            task_state=QUEUING,
+            task_type="type_2",
+            task_key='',
+            task_id=3,
+        )
+        self.tasks = [type_1_queueing, type_1_non_queueing, type_2_queueing]
+
+    def update_task_created(self, created_date):
+        """
+        Override each task's "created" date
+        """
+        for task in self.tasks:
+            task.created = datetime.strptime(created_date, "%Y-%m-%d")
+            task.save()
+
+    def get_tasks(self):
+        """
+        After the command is run, this queries again for the tasks we created
+        in `setUp`.
+        """
+        type_1_queueing = InstructorTask.objects.get(task_id=1)
+        type_1_non_queueing = InstructorTask.objects.get(task_id=2)
+        type_2_queueing = InstructorTask.objects.get(task_id=3)
+        return type_1_queueing, type_1_non_queueing, type_2_queueing
+
+    @ddt.data(
+        ('2015-05-05', '2015-05-07', '2015-05-06'),
+        ('2015-05-05', '2015-05-07', '2015-05-08'),
+        ('2015-05-05', '2015-05-07', '2015-05-04'),
+    )
+    @ddt.unpack
+    def test_dry_run(self, after, before, created):
+        """
+        Tests that nothing is updated when run with the `dry_run` option
+        """
+        self.update_task_created(created)
+        call_command(
+            'fail_old_queueing_tasks',
+            dry_run=True,
+            before=before,
+            after=after,
+        )
+
+        type_1_queueing, type_1_non_queueing, type_2_queueing = self.get_tasks()
+        self.assertEqual(type_1_queueing.task_state, QUEUING)
+        self.assertEqual(type_2_queueing.task_state, QUEUING)
+        self.assertEqual(type_1_non_queueing.task_state, 'NOT QUEUEING')
+
+    @ddt.data(
+        ('2015-05-05', '2015-05-07', '2015-05-06', FAILURE),
+        ('2015-05-05', '2015-05-07', '2015-05-08', QUEUING),
+        ('2015-05-05', '2015-05-07', '2015-05-04', QUEUING),
+    )
+    @ddt.unpack
+    def test_tasks_updated(self, after, before, created, expected_state):
+        """
+        Test that tasks created outside the window of dates don't get changed,
+        while tasks created in the window do get changed.
+        Verifies that non-queueing tasks never get changed.
+        """
+        self.update_task_created(created)
+
+        call_command('fail_old_queueing_tasks', before=before, after=after)
+
+        type_1_queueing, type_1_non_queueing, type_2_queueing = self.get_tasks()
+        self.assertEqual(type_1_queueing.task_state, expected_state)
+        self.assertEqual(type_2_queueing.task_state, expected_state)
+        self.assertEqual(type_1_non_queueing.task_state, 'NOT QUEUEING')
+
+    def test_filter_by_task_type(self):
+        """
+        Test that if we specify which task types to update, only tasks with
+        those types are updated
+        """
+        self.update_task_created('2015-05-06')
+        call_command(
+            'fail_old_queueing_tasks',
+            before='2015-05-07',
+            after='2015-05-05',
+            task_type="type_1",
+        )
+        type_1_queueing, type_1_non_queueing, type_2_queueing = self.get_tasks()
+        self.assertEqual(type_1_queueing.task_state, FAILURE)
+        # the other type of task shouldn't be updated
+        self.assertEqual(type_2_queueing.task_state, QUEUING)
+        self.assertEqual(type_1_non_queueing.task_state, 'NOT QUEUEING')
+
+    @ddt.data(
+        ('2015-05-05', None),
+        (None, '2015-05-05'),
+    )
+    @ddt.unpack
+    def test_date_errors(self, after, before):
+        """
+        Test that we get a CommandError when we don't supply before and after
+        dates.
+        """
+        with self.assertRaises(CommandError):
+            call_command('fail_old_queueing_tasks', before=before, after=after)