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

Make export URL restful.

Browse Source 
STUD-846
This commit is contained in:
cahrens
2013-10-29 16:43:13 -04:00
parent e071ebb981
commit 763ff9c818
8 changed files with 188 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
cms/djangoapps/contentstore/tests/test_contentstore.py
View File

@@ -1594,10 +1594,7 @@ class ContentStoreTest(ModuleStoreTestCase):
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

99
cms/djangoapps/contentstore/tests/test_import_export.py
View File

@@ -18,6 +18,7 @@ from django.conf import settings
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
@@ -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(
@@ -66,13 +66,11 @@ class ImportTestCase(CourseTestCase):
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
@@ -97,30 +95,25 @@ class ImportTestCase(CourseTestCase):
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:
@@ -136,7 +129,7 @@ class ImportTestCase(CourseTestCase):
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
@@ -185,10 +178,8 @@ class ImportTestCase(CourseTestCase):
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)
@@ -207,3 +198,77 @@ class ImportTestCase(CourseTestCase):
)
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, "Download Files")
def test_export_json_unsupported(self):
"""
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)

166
cms/djangoapps/contentstore/views/import_export.py
View File

@@ -29,17 +29,17 @@ from xmodule.modulestore.xml_importer import import_from_xml
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__)
@@ -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'))
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())
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_to_xml(modulestore('direct'), contentstore(), old_location, root_dir, name, modulestore())
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
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)