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

Throw error if messaging queue is not available

Browse Source
This commit is contained in:
noraiz-anwar
2017-07-12 20:30:13 +05:00
parent 3e7243ea67
commit d1da044520
7 changed files with 302 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
lms/djangoapps/instructor_task/tests/test_api.py
View File

@@ -32,7 +32,7 @@ from lms.djangoapps.instructor_task.api import (
submit_reset_problem_attempts_for_all_students,
submit_reset_problem_attempts_in_entrance_exam
)
from lms.djangoapps.instructor_task.api_helper import AlreadyRunningError
from lms.djangoapps.instructor_task.api_helper import AlreadyRunningError, QueueConnectionError
from lms.djangoapps.instructor_task.models import PROGRESS, InstructorTask
from lms.djangoapps.instructor_task.tasks import export_ora2_data
from lms.djangoapps.instructor_task.tests.test_base import (
@@ -43,6 +43,7 @@ from lms.djangoapps.instructor_task.tests.test_base import (
TestReportMixin
)
from xmodule.modulestore.exceptions import ItemNotFoundError
from celery.states import FAILURE
class InstructorTaskReportTest(InstructorTaskTestCase):
@@ -164,15 +165,31 @@ class InstructorTaskModuleSubmitTest(InstructorTaskModuleTestCase):
)
@ddt.unpack
def test_submit_task(self, task_function, expected_task_type, params=None):
"""
Tests submission of instructor task.
"""
if params is None:
params = {}
if params.get('student'):
params['student'] = self.student
# tests submit, and then tests a second identical submission.
problem_url_name = 'H1P1'
self.define_option_problem(problem_url_name)
location = InstructorTaskModuleTestCase.problem_location(problem_url_name)
# unsuccessful submission, exception raised while submitting.
with patch('lms.djangoapps.instructor_task.tasks_base.BaseInstructorTask.apply_async') as apply_async:
error = Exception()
apply_async.side_effect = error
with self.assertRaises(QueueConnectionError):
instructor_task = task_function(self.create_task_request(self.instructor), location, **params)
most_recent_task = InstructorTask.objects.latest('id')
self.assertEquals(most_recent_task.task_state, FAILURE)
# successful submission
instructor_task = task_function(self.create_task_request(self.instructor), location, **params)
self.assertEquals(instructor_task.task_type, expected_task_type)