151bd13666
* Generate common/djangoapps import shims for LMS * Generate common/djangoapps import shims for Studio * Stop appending project root to sys.path * Stop appending common/djangoapps to sys.path * Import from common.djangoapps.course_action_state instead of course_action_state * Import from common.djangoapps.course_modes instead of course_modes * Import from common.djangoapps.database_fixups instead of database_fixups * Import from common.djangoapps.edxmako instead of edxmako * Import from common.djangoapps.entitlements instead of entitlements * Import from common.djangoapps.pipline_mako instead of pipeline_mako * Import from common.djangoapps.static_replace instead of static_replace * Import from common.djangoapps.student instead of student * Import from common.djangoapps.terrain instead of terrain * Import from common.djangoapps.third_party_auth instead of third_party_auth * Import from common.djangoapps.track instead of track * Import from common.djangoapps.util instead of util * Import from common.djangoapps.xblock_django instead of xblock_django * Add empty common/djangoapps/__init__.py to fix pytest collection * Fix pylint formatting violations * Exclude import_shims/ directory tree from linting
85 lines
3.5 KiB
Python
85 lines
3.5 KiB
Python
"""
|
|
Unit tests for instructor_task subtasks.
|
|
"""
|
|
|
|
|
|
from uuid import uuid4
|
|
|
|
from mock import Mock, patch
|
|
from six.moves import range
|
|
|
|
from lms.djangoapps.instructor_task.subtasks import queue_subtasks_for_query
|
|
from lms.djangoapps.instructor_task.tests.factories import InstructorTaskFactory
|
|
from lms.djangoapps.instructor_task.tests.test_base import InstructorTaskCourseTestCase
|
|
from common.djangoapps.student.models import CourseEnrollment
|
|
|
|
|
|
class TestSubtasks(InstructorTaskCourseTestCase):
|
|
"""Tests for subtasks."""
|
|
|
|
def setUp(self):
|
|
super(TestSubtasks, self).setUp()
|
|
self.initialize_course()
|
|
|
|
def _enroll_students_in_course(self, course_id, num_students):
|
|
"""Create and enroll some students in the course."""
|
|
|
|
for _ in range(num_students):
|
|
random_id = uuid4().hex[:8]
|
|
self.create_student(username='student{0}'.format(random_id))
|
|
|
|
def _queue_subtasks(self, create_subtask_fcn, items_per_task, initial_count, extra_count):
|
|
"""Queue subtasks while enrolling more students into course in the middle of the process."""
|
|
|
|
task_id = str(uuid4())
|
|
instructor_task = InstructorTaskFactory.create(
|
|
course_id=self.course.id,
|
|
task_id=task_id,
|
|
task_key='dummy_task_key',
|
|
task_type='bulk_course_email',
|
|
)
|
|
|
|
self._enroll_students_in_course(self.course.id, initial_count)
|
|
task_querysets = [CourseEnrollment.objects.filter(course_id=self.course.id)]
|
|
|
|
def initialize_subtask_info(*args): # pylint: disable=unused-argument
|
|
"""Instead of initializing subtask info enroll some more students into course."""
|
|
self._enroll_students_in_course(self.course.id, extra_count)
|
|
return {}
|
|
|
|
with patch('lms.djangoapps.instructor_task.subtasks.initialize_subtask_info') as mock_initialize_subtask_info:
|
|
mock_initialize_subtask_info.side_effect = initialize_subtask_info
|
|
queue_subtasks_for_query(
|
|
entry=instructor_task,
|
|
action_name='action_name',
|
|
create_subtask_fcn=create_subtask_fcn,
|
|
item_querysets=task_querysets,
|
|
item_fields=[],
|
|
items_per_task=items_per_task,
|
|
total_num_items=initial_count,
|
|
)
|
|
|
|
def test_queue_subtasks_for_query1(self):
|
|
"""Test queue_subtasks_for_query() if the last subtask only needs to accommodate < items_per_tasks items."""
|
|
|
|
mock_create_subtask_fcn = Mock()
|
|
self._queue_subtasks(mock_create_subtask_fcn, 3, 7, 1)
|
|
|
|
# Check number of items for each subtask
|
|
mock_create_subtask_fcn_args = mock_create_subtask_fcn.call_args_list
|
|
self.assertEqual(len(mock_create_subtask_fcn_args[0][0][0]), 3)
|
|
self.assertEqual(len(mock_create_subtask_fcn_args[1][0][0]), 3)
|
|
self.assertEqual(len(mock_create_subtask_fcn_args[2][0][0]), 2)
|
|
|
|
def test_queue_subtasks_for_query2(self):
|
|
"""Test queue_subtasks_for_query() if the last subtask needs to accommodate > items_per_task items."""
|
|
|
|
mock_create_subtask_fcn = Mock()
|
|
self._queue_subtasks(mock_create_subtask_fcn, 3, 8, 3)
|
|
|
|
# Check number of items for each subtask
|
|
mock_create_subtask_fcn_args = mock_create_subtask_fcn.call_args_list
|
|
self.assertEqual(len(mock_create_subtask_fcn_args[0][0][0]), 3)
|
|
self.assertEqual(len(mock_create_subtask_fcn_args[1][0][0]), 3)
|
|
self.assertEqual(len(mock_create_subtask_fcn_args[2][0][0]), 5)
|