109 lines
3.9 KiB
Python
109 lines
3.9 KiB
Python
"""
|
|
Unit tests for course import and export Celery tasks
|
|
"""
|
|
from __future__ import absolute_import, division, print_function
|
|
|
|
import copy
|
|
import json
|
|
import mock
|
|
from uuid import uuid4
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import User
|
|
from django.test.utils import override_settings
|
|
|
|
from user_tasks.models import UserTaskArtifact, UserTaskStatus
|
|
|
|
from contentstore.tasks import export_olx
|
|
from contentstore.tests.test_libraries import LibraryTestCase
|
|
from contentstore.tests.utils import CourseTestCase
|
|
|
|
TEST_DATA_CONTENTSTORE = copy.deepcopy(settings.CONTENTSTORE)
|
|
TEST_DATA_CONTENTSTORE['DOC_STORE_CONFIG']['db'] = 'test_xcontent_%s' % uuid4().hex
|
|
|
|
|
|
def side_effect_exception(*args, **kwargs): # pylint: disable=unused-argument
|
|
"""
|
|
Side effect for mocking which raises an exception
|
|
"""
|
|
raise Exception('Boom!')
|
|
|
|
|
|
@override_settings(CONTENTSTORE=TEST_DATA_CONTENTSTORE)
|
|
class ExportCourseTestCase(CourseTestCase):
|
|
"""
|
|
Tests of the export_olx task applied to courses
|
|
"""
|
|
|
|
def test_success(self):
|
|
"""
|
|
Verify that a routine course export task succeeds
|
|
"""
|
|
key = str(self.course.location.course_key)
|
|
result = export_olx.delay(self.user.id, key, u'en')
|
|
status = UserTaskStatus.objects.get(task_id=result.id)
|
|
self.assertEqual(status.state, UserTaskStatus.SUCCEEDED)
|
|
artifacts = UserTaskArtifact.objects.filter(status=status)
|
|
self.assertEqual(len(artifacts), 1)
|
|
output = artifacts[0]
|
|
self.assertEqual(output.name, 'Output')
|
|
|
|
@mock.patch('contentstore.tasks.export_course_to_xml', side_effect=side_effect_exception)
|
|
def test_exception(self, mock_export): # pylint: disable=unused-argument
|
|
"""
|
|
The export task should fail gracefully if an exception is thrown
|
|
"""
|
|
key = str(self.course.location.course_key)
|
|
result = export_olx.delay(self.user.id, key, u'en')
|
|
self._assert_failed(result, json.dumps({u'raw_error_msg': u'Boom!'}))
|
|
|
|
def test_invalid_user_id(self):
|
|
"""
|
|
Verify that attempts to export a course as an invalid user fail
|
|
"""
|
|
user_id = User.objects.order_by(u'-id').first().pk + 100
|
|
key = str(self.course.location.course_key)
|
|
result = export_olx.delay(user_id, key, u'en')
|
|
self._assert_failed(result, u'Unknown User ID: {}'.format(user_id))
|
|
|
|
def test_non_course_author(self):
|
|
"""
|
|
Verify that users who aren't authors of the course are unable to export it
|
|
"""
|
|
_, nonstaff_user = self.create_non_staff_authed_user_client()
|
|
key = str(self.course.location.course_key)
|
|
result = export_olx.delay(nonstaff_user.id, key, u'en')
|
|
self._assert_failed(result, u'Permission denied')
|
|
|
|
def _assert_failed(self, task_result, error_message):
|
|
"""
|
|
Verify that a task failed with the specified error message
|
|
"""
|
|
status = UserTaskStatus.objects.get(task_id=task_result.id)
|
|
self.assertEqual(status.state, UserTaskStatus.FAILED)
|
|
artifacts = UserTaskArtifact.objects.filter(status=status)
|
|
self.assertEqual(len(artifacts), 1)
|
|
error = artifacts[0]
|
|
self.assertEqual(error.name, u'Error')
|
|
self.assertEqual(error.text, error_message)
|
|
|
|
|
|
@override_settings(CONTENTSTORE=TEST_DATA_CONTENTSTORE)
|
|
class ExportLibraryTestCase(LibraryTestCase):
|
|
"""
|
|
Tests of the export_olx task applied to libraries
|
|
"""
|
|
|
|
def test_success(self):
|
|
"""
|
|
Verify that a routine library export task succeeds
|
|
"""
|
|
key = str(self.lib_key)
|
|
result = export_olx.delay(self.user.id, key, u'en') # pylint: disable=no-member
|
|
status = UserTaskStatus.objects.get(task_id=result.id)
|
|
self.assertEqual(status.state, UserTaskStatus.SUCCEEDED)
|
|
artifacts = UserTaskArtifact.objects.filter(status=status)
|
|
self.assertEqual(len(artifacts), 1)
|
|
output = artifacts[0]
|
|
self.assertEqual(output.name, 'Output')
|