Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: handle exam submission and reset (#33323)

Browse Source 
handles exam events from event bus that impact the 'instructor' app. These apis deal with learner completion and managing problem attempt state.
This commit is contained in:
Zachary Hancock
2023-10-06 09:10:13 -04:00
committed by GitHub
parent 46776e92e2
commit 9f16b0f8f6
6 changed files with 520 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
lms/djangoapps/instructor/apps.py
View File

@@ -36,6 +36,7 @@ class InstructorConfig(AppConfig):
}
def ready(self):
from . import handlers # pylint: disable=unused-import,import-outside-toplevel
if settings.FEATURES.get('ENABLE_SPECIAL_EXAMS'):
from .services import InstructorService
set_runtime_service('instructor', InstructorService())

82
lms/djangoapps/instructor/handlers.py Normal file
View File

@@ -0,0 +1,82 @@
"""
Handlers for instructor
"""
import logging
from django.contrib.auth import get_user_model
from django.core.exceptions import ObjectDoesNotExist
from django.dispatch import receiver
from openedx_events.learning.signals import EXAM_ATTEMPT_RESET, EXAM_ATTEMPT_SUBMITTED
from lms.djangoapps.courseware.models import StudentModule
from lms.djangoapps.instructor import enrollment
from lms.djangoapps.instructor.tasks import update_exam_completion_task
User = get_user_model()
log = logging.getLogger(__name__)
@receiver(EXAM_ATTEMPT_SUBMITTED)
def handle_exam_completion(sender, signal, **kwargs):
"""
exam completion event from the event bus
"""
event_data = kwargs.get('exam_attempt')
user_data = event_data.student_user
usage_key = event_data.usage_key
update_exam_completion_task.apply_async((user_data.pii.username, str(usage_key), 1.0))
@receiver(EXAM_ATTEMPT_RESET)
def handle_exam_reset(sender, signal, **kwargs):
"""
exam reset event from the event bus
"""
event_data = kwargs.get('exam_attempt')
user_data = event_data.student_user
requesting_user_data = event_data.requesting_user
usage_key = event_data.usage_key
course_key = event_data.course_key
content_id = str(usage_key)
try:
student = User.objects.get(id=user_data.id)
except ObjectDoesNotExist:
log.error(
'Error occurred while attempting to reset student attempt for user_id '
f'{user_data.id} for content_id {content_id}. '
'User does not exist!'
)
return
try:
requesting_user = User.objects.get(id=requesting_user_data.id)
except ObjectDoesNotExist:
log.error(
'Error occurred while attempting to reset student attempt. Requesting user_id '
f'{requesting_user_data.id} does not exist!'
)
return
# reset problem state
try:
enrollment.reset_student_attempts(
course_key,
student,
usage_key,
requesting_user=requesting_user,
delete_module=True,
)
except (StudentModule.DoesNotExist, enrollment.sub_api.SubmissionError):
log.error(
'Error occurred while attempting to reset module state for user_id '
f'{student.id} for content_id {content_id}.'
)
# In some cases, reset_student_attempts does not clear the entire exam's completion state.
# One example of this is an exam with multiple units (verticals) within it and the learner
# never viewing one of the units. All of the content in that unit will still be marked complete,
# but the reset code is unable to handle clearing the completion in that scenario.
update_exam_completion_task.apply_async((student.username, content_id, 0.0))

4
lms/djangoapps/instructor/tasks.py
View File

@@ -5,14 +5,14 @@ import logging
from celery import shared_task
from celery_utils.logged_task import LoggedTask
from django.core.exceptions import ObjectDoesNotExist
from edx_django_utils.monitoring import set_code_owner_attribute
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import UsageKey
from xblock.completable import XBlockCompletionMode
from edx_django_utils.monitoring import set_code_owner_attribute
from common.djangoapps.student.models import get_user_by_username_or_email
from lms.djangoapps.courseware.model_data import FieldDataCache
from lms.djangoapps.courseware.block_render import get_block_for_descriptor
from lms.djangoapps.courseware.model_data import FieldDataCache
from openedx.core.lib.request_utils import get_request_or_stub
from xmodule.modulestore.django import modulestore # lint-amnesty, pylint: disable=wrong-import-order
from xmodule.modulestore.exceptions import ItemNotFoundError # lint-amnesty, pylint: disable=wrong-import-order

229
lms/djangoapps/instructor/tests/test_handlers.py Normal file
View File

@@ -0,0 +1,229 @@
"""
Unit tests for instructor signals
"""
from datetime import datetime, timezone
from unittest import mock
from uuid import uuid4
from django.test import TestCase
from opaque_keys.edx.keys import CourseKey, UsageKey
from openedx_events.data import EventsMetadata
from openedx_events.learning.data import ExamAttemptData, UserData, UserPersonalData
from openedx_events.learning.signals import EXAM_ATTEMPT_RESET, EXAM_ATTEMPT_SUBMITTED
from common.djangoapps.student.tests.factories import UserFactory
from lms.djangoapps.instructor import enrollment
from lms.djangoapps.instructor.handlers import handle_exam_completion, handle_exam_reset
class ExamCompletionEventBusTests(TestCase):
"""
Tests completion events from the event bus.
"""
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.course_key = CourseKey.from_string('course-v1:edX+TestX+Test_Course')
cls.subsection_id = 'block-v1:edX+TestX+Test_Course+type@sequential+block@subsection'
cls.subsection_key = UsageKey.from_string(cls.subsection_id)
cls.student_user = UserFactory(
username='student_user',
)
@staticmethod
def _get_exam_event_data(student_user, course_key, usage_key, requesting_user=None):
""" create ExamAttemptData object for exam based event """
if requesting_user:
requesting_user_data = UserData(
id=requesting_user.id,
is_active=True,
pii=None
)
else:
requesting_user_data = None
return ExamAttemptData(
student_user=UserData(
id=student_user.id,
is_active=True,
pii=UserPersonalData(
username=student_user.username,
email=student_user.email,
),
),
course_key=course_key,
usage_key=usage_key,
exam_type='timed',
requesting_user=requesting_user_data,
)
@staticmethod
def _get_exam_event_metadata(event_signal):
""" create metadata object for event """
return EventsMetadata(
event_type=event_signal.event_type,
id=uuid4(),
minorversion=0,
source='openedx/lms/web',
sourcehost='lms.test',
time=datetime.now(timezone.utc)
)
@mock.patch('lms.djangoapps.instructor.tasks.update_exam_completion_task.apply_async', autospec=True)
def test_submit_exam_completion_event(self, mock_task_apply):
"""
Assert update completion task is scheduled
"""
exam_event_data = self._get_exam_event_data(self.student_user, self.course_key, self.subsection_key)
event_metadata = self._get_exam_event_metadata(EXAM_ATTEMPT_SUBMITTED)
event_kwargs = {
'exam_attempt': exam_event_data,
'metadata': event_metadata
}
handle_exam_completion(None, EXAM_ATTEMPT_SUBMITTED, **event_kwargs)
mock_task_apply.assert_called_once_with(('student_user', self.subsection_id, 1.0))
@mock.patch('lms.djangoapps.instructor.tasks.update_exam_completion_task.apply_async', autospec=True)
@mock.patch('lms.djangoapps.instructor.enrollment.reset_student_attempts', autospec=True)
def test_exam_reset_event(self, mock_reset, mock_task_apply):
"""
Assert problem state and completion are reset
"""
staff_user = UserFactory(
username='staff_user',
is_staff=True,
)
exam_event_data = self._get_exam_event_data(
self.student_user,
self.course_key,
self.subsection_key,
requesting_user=staff_user
)
event_metadata = self._get_exam_event_metadata(EXAM_ATTEMPT_RESET)
event_kwargs = {
'exam_attempt': exam_event_data,
'metadata': event_metadata
}
# reset signal
handle_exam_reset(None, EXAM_ATTEMPT_RESET, **event_kwargs)
# make sure problem attempts have been deleted
mock_reset.assert_called_once_with(
self.course_key,
self.student_user,
self.subsection_key,
requesting_user=staff_user,
delete_module=True,
)
# Assert we update completion with 0.0
mock_task_apply.assert_called_once_with(('student_user', self.subsection_id, 0.0))
def test_exam_reset_bad_user(self):
staff_user = UserFactory(
username='staff_user',
is_staff=True,
)
# not a user
bad_user_data = ExamAttemptData(
student_user=UserData(
id=999,
is_active=True,
pii=UserPersonalData(
username='user_dne',
email='user_dne@example.com',
),
),
course_key=self.course_key,
usage_key=self.subsection_key,
exam_type='timed',
requesting_user=UserData(
id=staff_user.id,
is_active=True,
pii=None
),
)
event_metadata = self._get_exam_event_metadata(EXAM_ATTEMPT_RESET)
event_kwargs = {
'exam_attempt': bad_user_data,
'metadata': event_metadata
}
# reset signal
with mock.patch('lms.djangoapps.instructor.handlers.log.error') as mock_log:
handle_exam_reset(None, EXAM_ATTEMPT_RESET, **event_kwargs)
mock_log.assert_called_once_with(
'Error occurred while attempting to reset student attempt for user_id '
f'999 for content_id {bad_user_data.usage_key}. '
'User does not exist!'
)
def test_exam_reset_bad_requesting_user(self):
# requesting user is not a user
bad_user_data = ExamAttemptData(
student_user=UserData(
id=self.student_user.id,
is_active=True,
pii=UserPersonalData(
username='user_dne',
email='user_dne@example.com',
),
),
course_key=self.course_key,
usage_key=self.subsection_key,
exam_type='timed',
requesting_user=UserData(
id=999,
is_active=True,
pii=None
),
)
event_metadata = self._get_exam_event_metadata(EXAM_ATTEMPT_RESET)
event_kwargs = {
'exam_attempt': bad_user_data,
'metadata': event_metadata
}
# reset signal
with mock.patch('lms.djangoapps.instructor.handlers.log.error') as mock_log:
handle_exam_reset(None, EXAM_ATTEMPT_RESET, **event_kwargs)
mock_log.assert_called_once_with(
'Error occurred while attempting to reset student attempt. Requesting user_id '
'999 does not exist!'
)
@mock.patch(
'lms.djangoapps.instructor.enrollment.reset_student_attempts',
side_effect=enrollment.sub_api.SubmissionError
)
def test_module_reset_failure(self, mock_reset):
staff_user = UserFactory(
username='staff_user',
is_staff=True,
)
exam_event_data = self._get_exam_event_data(
self.student_user,
self.course_key,
self.subsection_key,
requesting_user=staff_user
)
event_metadata = self._get_exam_event_metadata(EXAM_ATTEMPT_RESET)
event_kwargs = {
'exam_attempt': exam_event_data,
'metadata': event_metadata
}
with mock.patch('lms.djangoapps.instructor.handlers.log.error') as mock_log:
# reset signal
handle_exam_reset(None, EXAM_ATTEMPT_RESET, **event_kwargs)
mock_log.assert_called_once_with(
'Error occurred while attempting to reset module state for user_id '
f'{self.student_user.id} for content_id {self.subsection_id}.'
)

155
lms/djangoapps/instructor/tests/test_services.py
View File

@@ -5,9 +5,7 @@ import json
from unittest import mock
import pytest
from completion.waffle import ENABLE_COMPLETION_TRACKING_SWITCH
from django.core.exceptions import ObjectDoesNotExist
from edx_toggles.toggles.testutils import override_waffle_switch
from opaque_keys import InvalidKeyError
from common.djangoapps.student.models import CourseEnrollment
@@ -15,9 +13,8 @@ from common.djangoapps.student.tests.factories import UserFactory
from lms.djangoapps.courseware.models import StudentModule
from lms.djangoapps.instructor.access import allow_access
from lms.djangoapps.instructor.services import InstructorService
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase # lint-amnesty, pylint: disable=wrong-import-order
from xmodule.modulestore.tests.factories import CourseFactory, BlockFactory # lint-amnesty, pylint: disable=wrong-import-order
from xmodule.partitions.partitions import Group, UserPartition # lint-amnesty, pylint: disable=wrong-import-order
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory
class InstructorServiceTests(SharedModuleStoreTestCase):
@@ -54,8 +51,8 @@ class InstructorServiceTests(SharedModuleStoreTestCase):
)
@mock.patch('lms.djangoapps.grades.signals.handlers.PROBLEM_WEIGHTED_SCORE_CHANGED.send')
@mock.patch('completion.handlers.BlockCompletion.objects.submit_completion')
def test_reset_student_attempts_delete(self, mock_submit, _mock_signal):
@mock.patch('lms.djangoapps.instructor.tasks.update_exam_completion_task.apply_async', autospec=True)
def test_reset_student_attempts_delete(self, mock_completion_task, _mock_signal):
"""
Test delete student state.
"""
@@ -64,22 +61,19 @@ class InstructorServiceTests(SharedModuleStoreTestCase):
assert StudentModule.objects.filter(student=self.module_to_reset.student, course_id=self.course.id,
module_state_key=self.module_to_reset.module_state_key).count() == 1
with override_waffle_switch(ENABLE_COMPLETION_TRACKING_SWITCH, True):
self.service.delete_student_attempt(
self.student.username,
str(self.course.id),
str(self.subsection.location),
requesting_user=self.student,
)
self.service.delete_student_attempt(
self.student.username,
str(self.course.id),
str(self.subsection.location),
requesting_user=self.student,
)
# make sure the module has been deleted
assert StudentModule.objects.filter(student=self.module_to_reset.student, course_id=self.course.id,
module_state_key=self.module_to_reset.module_state_key).count() == 0
# Assert we send completion == 0.0 for both problems even though the second problem was never viewed
assert mock_submit.call_count == 2
mock_submit.assert_any_call(user=self.student, block_key=self.problem.location, completion=0.0)
mock_submit.assert_any_call(user=self.student, block_key=self.problem_2.location, completion=0.0)
# Assert we send update completion with 0.0
mock_completion_task.assert_called_once_with((self.student.username, str(self.subsection.location), 0.0))
def test_reset_bad_content_id(self):
"""
@@ -120,128 +114,13 @@ class InstructorServiceTests(SharedModuleStoreTestCase):
)
assert result is None
@mock.patch('completion.handlers.BlockCompletion.objects.submit_completion')
def test_complete_student_attempt_success(self, mock_submit):
@mock.patch('lms.djangoapps.instructor.tasks.update_exam_completion_task.apply_async', autospec=True)
def test_complete_student_attempt_success(self, mock_completion_task):
"""
Assert complete_student_attempt correctly publishes completion for all
completable children of the given content_id
Assert update_exam_completion task is triggered
"""
# Section, subsection, and unit are all aggregators and not completable so should
# not be submitted.
section = BlockFactory.create(parent=self.course, category='chapter')
subsection = BlockFactory.create(parent=section, category='sequential')
unit = BlockFactory.create(parent=subsection, category='vertical')
# should both be submitted
video = BlockFactory.create(parent=unit, category='video')
problem = BlockFactory.create(parent=unit, category='problem')
# Not a completable block
BlockFactory.create(parent=unit, category='discussion')
with override_waffle_switch(ENABLE_COMPLETION_TRACKING_SWITCH, True):
self.service.complete_student_attempt(self.student.username, str(subsection.location))
# Only Completable leaf blocks should have completion published
assert mock_submit.call_count == 2
mock_submit.assert_any_call(user=self.student, block_key=video.location, completion=1.0)
mock_submit.assert_any_call(user=self.student, block_key=problem.location, completion=1.0)
@mock.patch('completion.handlers.BlockCompletion.objects.submit_completion')
def test_complete_student_attempt_split_test(self, mock_submit):
"""
Asserts complete_student_attempt correctly publishes completion when a split test is involved
This test case exists because we ran into a bug about the user_service not existing
when a split_test existed inside of a subsection. Associated with this change was adding
in the user state into the module before attempting completion and this ensures that is
working properly.
"""
partition = UserPartition(
0,
'first_partition',
'First Partition',
[
Group(0, 'alpha'),
Group(1, 'beta')
]
)
course = CourseFactory.create(user_partitions=[partition])
section = BlockFactory.create(parent=course, category='chapter')
subsection = BlockFactory.create(parent=section, category='sequential')
c0_url = course.id.make_usage_key('vertical', 'split_test_cond0')
c1_url = course.id.make_usage_key('vertical', 'split_test_cond1')
split_test = BlockFactory.create(
parent=subsection,
category='split_test',
user_partition_id=0,
group_id_to_child={'0': c0_url, '1': c1_url},
)
cond0vert = BlockFactory.create(parent=split_test, category='vertical', location=c0_url)
BlockFactory.create(parent=cond0vert, category='video')
BlockFactory.create(parent=cond0vert, category='problem')
cond1vert = BlockFactory.create(parent=split_test, category='vertical', location=c1_url)
BlockFactory.create(parent=cond1vert, category='video')
BlockFactory.create(parent=cond1vert, category='html')
with override_waffle_switch(ENABLE_COMPLETION_TRACKING_SWITCH, True):
self.service.complete_student_attempt(self.student.username, str(subsection.location))
# Only the group the user was assigned to should have completion published.
# Either cond0vert's children or cond1vert's children
assert mock_submit.call_count == 2
@mock.patch('lms.djangoapps.instructor.tasks.log.error')
def test_complete_student_attempt_bad_user(self, mock_logger):
"""
Assert complete_student_attempt with a bad user raises error and returns None
"""
username = 'bad_user'
block_id = str(self.problem.location)
self.service.complete_student_attempt(username, block_id)
mock_logger.assert_called_once_with(
self.complete_error_prefix.format(user=username, content_id=block_id) + 'User does not exist!'
)
@mock.patch('lms.djangoapps.instructor.tasks.log.error')
def test_complete_student_attempt_bad_content_id(self, mock_logger):
"""
Assert complete_student_attempt with a bad content_id raises error and returns None
"""
username = self.student.username
self.service.complete_student_attempt(username, 'foo/bar/baz')
mock_logger.assert_called_once_with(
self.complete_error_prefix.format(user=username, content_id='foo/bar/baz') + 'Invalid content_id!'
)
@mock.patch('lms.djangoapps.instructor.tasks.log.error')
def test_complete_student_attempt_nonexisting_item(self, mock_logger):
"""
Assert complete_student_attempt with nonexisting item in the modulestore
raises error and returns None
"""
username = self.student.username
block = 'i4x://org.0/course_0/problem/fake_problem'
self.service.complete_student_attempt(username, block)
mock_logger.assert_called_once_with(
self.complete_error_prefix.format(user=username, content_id=block) + 'Block not found in the modulestore!'
)
@mock.patch('lms.djangoapps.instructor.tasks.log.error')
def test_complete_student_attempt_failed_module(self, mock_logger):
"""
Assert complete_student_attempt with failed get_block raises error and returns None
"""
username = self.student.username
with mock.patch('lms.djangoapps.instructor.tasks.get_block_for_descriptor', return_value=None):
self.service.complete_student_attempt(username, str(self.course.location))
mock_logger.assert_called_once_with(
self.complete_error_prefix.format(user=username, content_id=self.course.location) +
'Block unable to be created from descriptor!'
)
self.service.complete_student_attempt(self.student.username, str(self.subsection.location))
mock_completion_task.assert_called_once_with((self.student.username, str(self.subsection.location), 1.0))
def test_is_user_staff(self):
"""

189
lms/djangoapps/instructor/tests/test_tasks.py Normal file
View File

@@ -0,0 +1,189 @@
"""
Tests for tasks.py
"""
import json
from unittest import mock
from completion.waffle import ENABLE_COMPLETION_TRACKING_SWITCH
from edx_toggles.toggles.testutils import override_waffle_switch
from common.djangoapps.student.models import CourseEnrollment
from common.djangoapps.student.tests.factories import UserFactory
from lms.djangoapps.courseware.models import StudentModule
from lms.djangoapps.instructor.tasks import update_exam_completion_task
from xmodule.modulestore.tests.django_utils import \
SharedModuleStoreTestCase # lint-amnesty, pylint: disable=wrong-import-order
from xmodule.modulestore.tests.factories import ( # lint-amnesty, pylint: disable=wrong-import-order
BlockFactory,
CourseFactory
)
from xmodule.partitions.partitions import Group, UserPartition # lint-amnesty, pylint: disable=wrong-import-order
class UpdateCompletionTests(SharedModuleStoreTestCase):
"""
Test the update_exam_completion_task
"""
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.email = 'escalation@test.com'
cls.course = CourseFactory.create(proctoring_escalation_email=cls.email)
cls.section = BlockFactory.create(parent=cls.course, category='chapter')
cls.subsection = BlockFactory.create(parent=cls.section, category='sequential')
cls.unit = BlockFactory.create(parent=cls.subsection, category='vertical')
cls.problem = BlockFactory.create(parent=cls.unit, category='problem')
cls.unit_2 = BlockFactory.create(parent=cls.subsection, category='vertical')
cls.problem_2 = BlockFactory.create(parent=cls.unit_2, category='problem')
cls.complete_error_prefix = ('Error occurred while attempting to complete student attempt for '
'user {user} for content_id {content_id}. ')
def setUp(self):
super().setUp()
self.student = UserFactory()
CourseEnrollment.enroll(self.student, self.course.id)
self.module_to_reset = StudentModule.objects.create(
student=self.student,
course_id=self.course.id,
module_state_key=self.problem.location,
state=json.dumps({'attempts': 2}),
)
@mock.patch('completion.handlers.BlockCompletion.objects.submit_completion')
def test_update_completion_success(self, mock_submit):
"""
Assert correctly publishes completion for all
completable children of the given content_id
"""
# Section, subsection, and unit are all aggregators and not completable so should
# not be submitted.
section = BlockFactory.create(parent=self.course, category='chapter')
subsection = BlockFactory.create(parent=section, category='sequential')
unit = BlockFactory.create(parent=subsection, category='vertical')
# should both be submitted
video = BlockFactory.create(parent=unit, category='video')
problem = BlockFactory.create(parent=unit, category='problem')
# Not a completable block
BlockFactory.create(parent=unit, category='discussion')
with override_waffle_switch(ENABLE_COMPLETION_TRACKING_SWITCH, True):
update_exam_completion_task(self.student.username, str(subsection.location), 1.0)
# Only Completable leaf blocks should have completion published
assert mock_submit.call_count == 2
mock_submit.assert_any_call(user=self.student, block_key=video.location, completion=1.0)
mock_submit.assert_any_call(user=self.student, block_key=problem.location, completion=1.0)
@mock.patch('completion.handlers.BlockCompletion.objects.submit_completion')
def test_update_completion_delete(self, mock_submit):
"""
Test update completion with a value of 0.0
"""
with override_waffle_switch(ENABLE_COMPLETION_TRACKING_SWITCH, True):
update_exam_completion_task(self.student.username, str(self.subsection.location), 0.0)
# Assert we send completion == 0.0 for both problems
assert mock_submit.call_count == 2
mock_submit.assert_any_call(user=self.student, block_key=self.problem.location, completion=0.0)
mock_submit.assert_any_call(user=self.student, block_key=self.problem_2.location, completion=0.0)
@mock.patch('completion.handlers.BlockCompletion.objects.submit_completion')
def test_update_completion_split_test(self, mock_submit):
"""
Asserts correctly publishes completion when a split test is involved
This test case exists because we ran into a bug about the user_service not existing
when a split_test existed inside of a subsection. Associated with this change was adding
in the user state into the module before attempting completion and this ensures that is
working properly.
"""
partition = UserPartition(
0,
'first_partition',
'First Partition',
[
Group(0, 'alpha'),
Group(1, 'beta')
]
)
course = CourseFactory.create(user_partitions=[partition])
section = BlockFactory.create(parent=course, category='chapter')
subsection = BlockFactory.create(parent=section, category='sequential')
c0_url = course.id.make_usage_key('vertical', 'split_test_cond0')
c1_url = course.id.make_usage_key('vertical', 'split_test_cond1')
split_test = BlockFactory.create(
parent=subsection,
category='split_test',
user_partition_id=0,
group_id_to_child={'0': c0_url, '1': c1_url},
)
cond0vert = BlockFactory.create(parent=split_test, category='vertical', location=c0_url)
BlockFactory.create(parent=cond0vert, category='video')
BlockFactory.create(parent=cond0vert, category='problem')
cond1vert = BlockFactory.create(parent=split_test, category='vertical', location=c1_url)
BlockFactory.create(parent=cond1vert, category='video')
BlockFactory.create(parent=cond1vert, category='html')
with override_waffle_switch(ENABLE_COMPLETION_TRACKING_SWITCH, True):
update_exam_completion_task(self.student.username, str(subsection.location), 1.0)
# Only the group the user was assigned to should have completion published.
# Either cond0vert's children or cond1vert's children
assert mock_submit.call_count == 2
@mock.patch('lms.djangoapps.instructor.tasks.log.error')
def test_update_completion_bad_user(self, mock_logger):
"""
Assert a bad user raises error and returns None
"""
username = 'bad_user'
block_id = str(self.problem.location)
update_exam_completion_task(username, block_id, 1.0)
mock_logger.assert_called_once_with(
self.complete_error_prefix.format(user=username, content_id=block_id) + 'User does not exist!'
)
@mock.patch('lms.djangoapps.instructor.tasks.log.error')
def test_update_completion_bad_content_id(self, mock_logger):
"""
Assert a bad content_id raises error and returns None
"""
username = self.student.username
update_exam_completion_task(username, 'foo/bar/baz', 1.0)
mock_logger.assert_called_once_with(
self.complete_error_prefix.format(user=username, content_id='foo/bar/baz') + 'Invalid content_id!'
)
@mock.patch('lms.djangoapps.instructor.tasks.log.error')
def test_update_completion_nonexisting_item(self, mock_logger):
"""
Assert nonexisting item in the modulestore
raises error and returns None
"""
username = self.student.username
block = 'i4x://org.0/course_0/problem/fake_problem'
update_exam_completion_task(username, block, 1.0)
mock_logger.assert_called_once_with(
self.complete_error_prefix.format(user=username, content_id=block) + 'Block not found in the modulestore!'
)
@mock.patch('lms.djangoapps.instructor.tasks.log.error')
def test_update_completion_failed_module(self, mock_logger):
"""
Assert failed get_block raises error and returns None
"""
username = self.student.username
with mock.patch('lms.djangoapps.instructor.tasks.get_block_for_descriptor', return_value=None):
update_exam_completion_task(username, str(self.course.location), 1.0)
mock_logger.assert_called_once_with(
self.complete_error_prefix.format(user=username, content_id=self.course.location) +
'Block unable to be created from descriptor!'
)