Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ disabilites. (LMS-1303)

Common: Add skip links for accessibility to CMS and LMS. (LMS-1311)

Studio: Change course overview page, checklists, assets, and course staff
Studio: Change course overview page, checklists, assets, import, export, and course staff
management page URLs to a RESTful interface. Also removed "\listing", which
duplicated "\index".

Expand Down
5 changes: 1 addition & 4 deletions cms/djangoapps/contentstore/tests/test_contentstore.py
Original file line number Diff line number Diff line change
Expand Up @@ -1594,10 +1594,7 @@ def test_cms_imported_course_walkthrough(self):
self.assertEqual(resp.status_code, 200)

# export page
resp = self.client.get(reverse('export_course',
kwargs={'org': loc.org,
'course': loc.course,
'name': loc.name}))
resp = self.client.get_html(new_location.url_reverse('export/', ''))
self.assertEqual(resp.status_code, 200)

# course team
Expand Down
99 changes: 82 additions & 17 deletions cms/djangoapps/contentstore/tests/test_import_export.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from xmodule.modulestore.django import loc_mapper

from xmodule.contentstore.django import _CONTENTSTORE
from xmodule.modulestore.tests.factories import ItemFactory

TEST_DATA_CONTENTSTORE = copy.deepcopy(settings.CONTENTSTORE)
TEST_DATA_CONTENTSTORE['DOC_STORE_CONFIG']['db'] = 'test_xcontent_%s' % uuid4().hex
Expand All @@ -29,7 +30,6 @@ class ImportTestCase(CourseTestCase):
"""
Unit tests for importing a course
"""

def setUp(self):
super(ImportTestCase, self).setUp()
self.new_location = loc_mapper().translate_location(
Expand Down Expand Up @@ -66,13 +66,11 @@ def touch(name):

self.unsafe_common_dir = path(tempfile.mkdtemp(dir=self.content_dir))


def tearDown(self):
shutil.rmtree(self.content_dir)
MongoClient().drop_database(TEST_DATA_CONTENTSTORE['DOC_STORE_CONFIG']['db'])
_CONTENTSTORE.clear()


def test_no_coursexml(self):
"""
Check that the response for a tar.gz import without a course.xml is
Expand All @@ -97,30 +95,25 @@ def test_no_coursexml(self):

self.assertEquals(json.loads(resp_status.content)["ImportStatus"], 2)


def test_with_coursexml(self):
"""
Check that the response for a tar.gz import with a course.xml is
correct.
"""
with open(self.good_tar) as gtar:
resp = self.client.post(
self.url,
{
"name": self.good_tar,
"course-data": [gtar]
})
args = {"name": self.good_tar, "course-data": [gtar]}
resp = self.client.post(self.url, args)

self.assertEquals(resp.status_code, 200)

## Unsafe tar methods #####################################################
# Each of these methods creates a tarfile with a single type of unsafe
# content.

def _fifo_tar(self):
"""
Tar file with FIFO
"""
fifop = self.unsafe_common_dir / "fifo.file"
fifop = self.unsafe_common_dir / "fifo.file"
fifo_tar = self.unsafe_common_dir / "fifo.tar.gz"
os.mkfifo(fifop)
with tarfile.open(fifo_tar, "w:gz") as tar:
Expand All @@ -136,7 +129,7 @@ def _symlink_tar(self):
symlinkp = self.unsafe_common_dir / "symlink.txt"
symlink_tar = self.unsafe_common_dir / "symlink.tar.gz"
outsidep.symlink(symlinkp)
with tarfile.open(symlink_tar, "w:gz" ) as tar:
with tarfile.open(symlink_tar, "w:gz") as tar:
tar.add(symlinkp)

return symlink_tar
Expand Down Expand Up @@ -185,10 +178,8 @@ def test_unsafe_tar(self):

def try_tar(tarpath):
with open(tarpath) as tar:
resp = self.client.post(
self.url,
{ "name": tarpath, "course-data": [tar] }
)
args = { "name": tarpath, "course-data": [tar] }
resp = self.client.post(self.url, args)
self.assertEquals(resp.status_code, 400)
self.assertTrue("SuspiciousFileOperation" in resp.content)

Expand All @@ -207,3 +198,77 @@ def try_tar(tarpath):
)
import_status = json.loads(resp_status.content)["ImportStatus"]
self.assertIn(import_status, (0, 3))


@override_settings(CONTENTSTORE=TEST_DATA_CONTENTSTORE)
class ExportTestCase(CourseTestCase):
"""
Tests for export_handler.
"""
def setUp(self):
"""
Sets up the test course.
"""
super(ExportTestCase, self).setUp()
location = loc_mapper().translate_location(self.course.location.course_id, self.course.location, False, True)
self.url = location.url_reverse('export/', '')

def test_export_html(self):
"""
Get the HTML for the page.
"""
resp = self.client.get_html(self.url)
self.assertEquals(resp.status_code, 200)
self.assertContains(resp, "Export a Copy of My Course Data")

def test_export_json_unsupported(self):

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it unsupported by design or by work priority? I believe tests should only test intentional design and act as a quasi-spec.

"""
JSON is unsupported.
"""
resp = self.client.get(self.url, HTTP_ACCEPT='application/json')
self.assertEquals(resp.status_code, 406)

def test_export_targz(self):
"""
Get tar.gz file, using HTTP_ACCEPT.
"""
resp = self.client.get(self.url, HTTP_ACCEPT='application/x-tgz')
self._verify_export_succeeded(resp)

def test_export_targz_urlparam(self):
"""
Get tar.gz file, using URL parameter.
"""
resp = self.client.get(self.url + '?_accept=application/x-tgz')
self._verify_export_succeeded(resp)

def _verify_export_succeeded(self, resp):
""" Export success helper method. """
self.assertEquals(resp.status_code, 200)
self.assertTrue(resp.get('Content-Disposition').startswith('attachment'))

def test_export_failure_top_level(self):
"""
Export failure.
"""
ItemFactory.create(parent_location=self.course.location, category='aawefawef')
self._verify_export_failure('/course/MITx.999.Robot_Super_Course/branch/draft/block/Robot_Super_Course')

def test_export_failure_subsection_level(self):
"""
Slightly different export failure.
"""
vertical = ItemFactory.create(parent_location=self.course.location, category='vertical', display_name='foo')
ItemFactory.create(
parent_location=vertical.location,
category='aawefawef'
)
self._verify_export_failure('/edit/i4x://MITx/999/vertical/foo')

def _verify_export_failure(self, expectedText):
""" Export failure helper method. """
resp = self.client.get(self.url, HTTP_ACCEPT='application/x-tgz')
self.assertEquals(resp.status_code, 200)
self.assertIsNone(resp.get('Content-Disposition'))
self.assertContains(resp, 'Unable to create xml for module')
self.assertContains(resp, expectedText)
168 changes: 91 additions & 77 deletions cms/djangoapps/contentstore/views/import_export.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,17 +29,17 @@
from xmodule.contentstore.django import contentstore
from xmodule.modulestore.xml_exporter import export_to_xml
from xmodule.modulestore.django import modulestore, loc_mapper
from xmodule.modulestore import Location
from xmodule.exceptions import SerializationError

from xmodule.modulestore.locator import BlockUsageLocator
from .access import has_access

from .access import get_location_and_verify_access
from util.json_request import JsonResponse
from extract_tar import safetar_extractall


__all__ = ['import_handler', 'import_status_handler', 'generate_export_course', 'export_course']
__all__ = ['import_handler', 'import_status_handler', 'export_handler']


log = logging.getLogger(__name__)

Expand Down Expand Up @@ -287,88 +287,102 @@ def import_status_handler(request, tag=None, course_id=None, branch=None, versio

@ensure_csrf_cookie
@login_required
def generate_export_course(request, org, course, name):
@require_http_methods(("GET",))
def export_handler(request, tag=None, course_id=None, branch=None, version_guid=None, block=None):
"""
This method will serialize out a course to a .tar.gz file which contains a
XML-based representation of the course
The restful handler for exporting a course.

GET
html: return html page for import page
application/x-tgz: return tar.gz file containing exported course
json: not supported

Note that there are 2 ways to request the tar.gz file. The request header can specify
application/x-tgz via HTTP_ACCEPT, or a query parameter can be used (?_accept=application/x-tgz).

If the tar.gz file has been requested but the export operation fails, an HTML page will be returned
which describes the error.
"""
location = get_location_and_verify_access(request, org, course, name)
course_module = modulestore().get_instance(location.course_id, location)
loc = Location(location)
export_file = NamedTemporaryFile(prefix=name + '.', suffix=".tar.gz")
location = BlockUsageLocator(course_id=course_id, branch=branch, version_guid=version_guid, usage_id=block)
if not has_access(request.user, location):
raise PermissionDenied()

new_location = loc_mapper().translate_location(course_module.location.course_id, course_module.location, False, True)
old_location = loc_mapper().translate_locator_to_location(location)
course_module = modulestore().get_item(old_location)

root_dir = path(mkdtemp())
# an _accept URL parameter will be preferred over HTTP_ACCEPT in the header.
requested_format = request.REQUEST.get('_accept', request.META.get('HTTP_ACCEPT', 'text/html'))

try:
export_to_xml(modulestore('direct'), contentstore(), loc, root_dir, name, modulestore())
except SerializationError, e:
logging.exception('There was an error exporting course {0}. {1}'.format(course_module.location, unicode(e)))
unit = None
failed_item = None
parent = None
try:
failed_item = modulestore().get_instance(course_module.location.course_id, e.location)
parent_locs = modulestore().get_parent_locations(failed_item.location, course_module.location.course_id)
export_url = location.url_reverse('export/', '') + '?_accept=application/x-tgz'
if 'application/x-tgz' in requested_format:
name = old_location.name
export_file = NamedTemporaryFile(prefix=name + '.', suffix=".tar.gz")
root_dir = path(mkdtemp())

if len(parent_locs) > 0:
parent = modulestore().get_item(parent_locs[0])
if parent.location.category == 'vertical':
unit = parent
except:
# if we have a nested exception, then we'll show the more generic error message
pass
try:
export_to_xml(modulestore('direct'), contentstore(), old_location, root_dir, name, modulestore())

except SerializationError, e:
logging.exception('There was an error exporting course {0}. {1}'.format(course_module.location, unicode(e)))
unit = None
failed_item = None
parent = None
try:
failed_item = modulestore().get_instance(course_module.location.course_id, e.location)
parent_locs = modulestore().get_parent_locations(failed_item.location, course_module.location.course_id)

if len(parent_locs) > 0:
parent = modulestore().get_item(parent_locs[0])
if parent.location.category == 'vertical':
unit = parent
except:
# if we have a nested exception, then we'll show the more generic error message
pass

return render_to_response('export.html', {
'context_course': course_module,
'in_err': True,
'raw_err_msg': str(e),
'failed_module': failed_item,
'unit': unit,
'edit_unit_url': reverse('edit_unit', kwargs={
'location': parent.location
}) if parent else '',
'course_home_url': location.url_reverse("course/", ""),
'export_url': export_url

})
except Exception, e:
logging.exception('There was an error exporting course {0}. {1}'.format(course_module.location, unicode(e)))
return render_to_response('export.html', {
'context_course': course_module,
'in_err': True,
'unit': None,
'raw_err_msg': str(e),
'course_home_url': location.url_reverse("course/", ""),
'export_url': export_url
})

logging.debug('tar file being generated at {0}'.format(export_file.name))
tar_file = tarfile.open(name=export_file.name, mode='w:gz')
tar_file.add(root_dir / name, arcname=name)
tar_file.close()

# remove temp dir
shutil.rmtree(root_dir / name)

wrapper = FileWrapper(export_file)
response = HttpResponse(wrapper, content_type='application/x-tgz')
response['Content-Disposition'] = 'attachment; filename=%s' % os.path.basename(export_file.name)
response['Content-Length'] = os.path.getsize(export_file.name)
return response

elif 'text/html' in requested_format:
return render_to_response('export.html', {
'context_course': course_module,
'successful_import_redirect_url': '',
'in_err': True,
'raw_err_msg': str(e),
'failed_module': failed_item,
'unit': unit,
'edit_unit_url': reverse('edit_unit', kwargs={
'location': parent.location
}) if parent else '',
'course_home_url': new_location.url_reverse("course/", "")
})
except Exception, e:
logging.exception('There was an error exporting course {0}. {1}'.format(course_module.location, unicode(e)))
return render_to_response('export.html', {
'context_course': course_module,
'successful_import_redirect_url': '',
'in_err': True,
'unit': None,
'raw_err_msg': str(e),
'course_home_url': new_location.url_reverse("course/", "")
'export_url': export_url
})

logging.debug('tar file being generated at {0}'.format(export_file.name))
tar_file = tarfile.open(name=export_file.name, mode='w:gz')
tar_file.add(root_dir / name, arcname=name)
tar_file.close()

# remove temp dir
shutil.rmtree(root_dir / name)

wrapper = FileWrapper(export_file)
response = HttpResponse(wrapper, content_type='application/x-tgz')
response['Content-Disposition'] = 'attachment; filename=%s' % os.path.basename(export_file.name)
response['Content-Length'] = os.path.getsize(export_file.name)
return response


@ensure_csrf_cookie
@login_required
def export_course(request, org, course, name):
"""
This method serves up the 'Export Course' page
"""
location = get_location_and_verify_access(request, org, course, name)

course_module = modulestore().get_item(location)

return render_to_response('export.html', {
'context_course': course_module,
'successful_import_redirect_url': ''
})
else:
# Only HTML or x-tgz request formats are supported (no JSON).
return HttpResponse(status=406)
Loading