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

Merge pull request #5961 from edx/dan-f/cohort-csv-upload

Browse Source 
Upload CSV file for cohorting students
This commit is contained in:
Christina Roberts
2014-12-10 11:04:14 -05:00
parent dfc4da1574 3d91f43030
commit 77f5d90d6a
42 changed files with 2139 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
common/djangoapps/student/management/commands/create_random_users.py
View File

@@ -1,12 +1,27 @@
"""
A script to create some dummy users
"""
import uuid
from django.core.management.base import BaseCommand
from student.models import CourseEnrollment
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import CourseKey
from opaque_keys.edx.locations import SlashSeparatedCourseKey
from student.views import _do_create_account, get_random_post_override
from student.views import _do_create_account
def get_random_post_override():
"""
Generate unique user data for dummy users.
"""
identification = uuid.uuid4().hex[:8]
return {
'username': 'user_{id}'.format(id=identification),
'email': 'email_{id}@example.com'.format(id=identification),
'password': '12345',
'name': 'User {id}'.format(id=identification),
}
def create(num, course_key):

168
common/djangoapps/util/file.py Normal file
View File

@@ -0,0 +1,168 @@
"""
Utility methods related to file handling.
"""
from datetime import datetime
import os
from pytz import UTC
from django.core.exceptions import PermissionDenied
from django.core.files.storage import DefaultStorage, get_valid_filename
from django.utils.translation import ugettext as _
from django.utils.translation import ungettext
class FileValidationException(Exception):
"""
An exception thrown during file validation.
"""
pass
def store_uploaded_file(
request, file_key, allowed_file_types, base_storage_filename, max_file_size, validator=None,
):
"""
Stores an uploaded file to django file storage.
Args:
request (HttpRequest): A request object from which a file will be retrieved.
file_key (str): The key for retrieving the file from `request.FILES`. If no entry exists with this
key, a `ValueError` will be thrown.
allowed_file_types (list): a list of allowable file type extensions. These should start with a period
and be specified in lower-case. For example, ['.txt', '.csv']. If the uploaded file does not end
with one of these extensions, a `PermissionDenied` exception will be thrown. Note that the uploaded file
extension does not need to be lower-case.
base_storage_filename (str): the filename to be used for the stored file, not including the extension.
The same extension as the uploaded file will be appended to this value.
max_file_size (int): the maximum file size in bytes that the uploaded file can be. If the uploaded file
is larger than this size, a `PermissionDenied` exception will be thrown.
validator (function): an optional validation method that, if defined, will be passed the stored file (which
is copied from the uploaded file). This method can do validation on the contents of the file and throw
a `FileValidationException` if the file is not properly formatted. If any exception is thrown, the stored
file will be deleted before the exception is re-raised. Note that the implementor of the validator function
should take care to close the stored file if they open it for reading.
Returns:
Storage: the file storage object where the file can be retrieved from
str: stored_file_name: the name of the stored file (including extension)
"""
if file_key not in request.FILES:
raise ValueError("No file uploaded with key '" + file_key + "'.")
uploaded_file = request.FILES[file_key]
try:
file_extension = os.path.splitext(uploaded_file.name)[1].lower()
if not file_extension in allowed_file_types:
file_types = "', '".join(allowed_file_types)
msg = ungettext(
"The file must end with the extension '{file_types}'.",
"The file must end with one of the following extensions: '{file_types}'.",
len(allowed_file_types)).format(file_types=file_types)
raise PermissionDenied(msg)
if uploaded_file.size > max_file_size:
msg = _("Maximum upload file size is {file_size} bytes.").format(file_size=max_file_size)
raise PermissionDenied(msg)
stored_file_name = base_storage_filename + file_extension
file_storage = DefaultStorage()
file_storage.save(stored_file_name, uploaded_file)
if validator:
try:
validator(file_storage, stored_file_name)
except:
file_storage.delete(stored_file_name)
raise
finally:
uploaded_file.close()
return file_storage, stored_file_name
# pylint: disable=invalid-name
def course_filename_prefix_generator(course_id, separator='_'):
"""
Generates a course-identifying unicode string for use in a file
name.
Args:
course_id (object): A course identification object.
Returns:
str: A unicode string which can safely be inserted into a
filename.
"""
return get_valid_filename(unicode(separator).join([course_id.org, course_id.course, course_id.run]))
# pylint: disable=invalid-name
def course_and_time_based_filename_generator(course_id, base_name):
"""
Generates a filename (without extension) based on the current time and the supplied filename.
Args:
course_id (object): A course identification object (must have org, course, and run).
base_name (str): A name describing what type of file this is. Any characters that are not safe for
filenames will be converted per django.core.files.storage.get_valid_filename (Specifically,
leading and trailing spaces are removed; other spaces are converted to underscores; and anything
that is not a unicode alphanumeric, dash, underscore, or dot, is removed).
Returns:
str: a concatenation of the org, course and run from the input course_id, the input base_name,
and the current time. Note that there will be no extension.
"""
return u"{course_prefix}_{base_name}_{timestamp_str}".format(
course_prefix=course_filename_prefix_generator(course_id),
base_name=get_valid_filename(base_name),
timestamp_str=datetime.now(UTC).strftime("%Y-%m-%d-%H%M%S") # pylint: disable=maybe-no-member
)
class UniversalNewlineIterator(object):
"""
This iterable class can be used as a wrapper around a file-like
object which does not inherently support being read in
universal-newline mode. It returns a line at a time.
"""
def __init__(self, original_file, buffer_size=4096):
self.original_file = original_file
self.buffer_size = buffer_size
def __iter__(self):
return self.generate_lines()
@staticmethod
def sanitize(string):
"""
Replace CR and CRLF with LF within `string`.
"""
return string.replace('\r\n', '\n').replace('\r', '\n')
def generate_lines(self):
"""
Return data from `self.original_file` a line at a time,
replacing CR and CRLF with LF.
"""
buf = self.original_file.read(self.buffer_size)
line = ''
while buf:
for char in buf:
if line.endswith('\r') and char == '\n':
last_line = line
line = ''
yield self.sanitize(last_line)
elif line.endswith('\r') or line.endswith('\n'):
last_line = line
line = char
yield self.sanitize(last_line)
else:
line += char
buf = self.original_file.read(self.buffer_size)
if not buf and line:
yield self.sanitize(line)

257
common/djangoapps/util/tests/test_file.py Normal file
View File

@@ -0,0 +1,257 @@
# -*- coding: utf-8 -*-
"""
Tests for file.py
"""
import ddt
from io import StringIO
from django.test import TestCase
from datetime import datetime
from django.utils.timezone import UTC
from mock import patch, Mock
from django.http import HttpRequest
from django.core.files.uploadedfile import SimpleUploadedFile
import util.file
from util.file import (
course_and_time_based_filename_generator,
course_filename_prefix_generator,
store_uploaded_file,
FileValidationException,
UniversalNewlineIterator
)
from opaque_keys.edx.locations import CourseLocator, SlashSeparatedCourseKey
from django.core import exceptions
import os
@ddt.ddt
class FilenamePrefixGeneratorTestCase(TestCase):
"""
Tests for course_filename_prefix_generator
"""
@ddt.data(CourseLocator, SlashSeparatedCourseKey)
def test_locators(self, course_key_class):
self.assertEqual(
course_filename_prefix_generator(course_key_class(org='foo', course='bar', run='baz')),
u'foo_bar_baz'
)
@ddt.data(CourseLocator, SlashSeparatedCourseKey)
def test_custom_separator(self, course_key_class):
self.assertEqual(
course_filename_prefix_generator(course_key_class(org='foo', course='bar', run='baz'), separator='-'),
u'foo-bar-baz'
)
@ddt.ddt
class FilenameGeneratorTestCase(TestCase):
"""
Tests for course_and_time_based_filename_generator
"""
NOW = datetime.strptime('1974-06-22T01:02:03', '%Y-%m-%dT%H:%M:%S').replace(tzinfo=UTC())
def setUp(self):
datetime_patcher = patch.object(
util.file, 'datetime',
Mock(wraps=datetime)
)
mocked_datetime = datetime_patcher.start()
mocked_datetime.now.return_value = self.NOW
self.addCleanup(datetime_patcher.stop)
@ddt.data(CourseLocator, SlashSeparatedCourseKey)
def test_filename_generator(self, course_key_class):
"""
Tests that the generator creates names based on course_id, base name, and date.
"""
self.assertEqual(
u'foo_bar_baz_file_1974-06-22-010203',
course_and_time_based_filename_generator(course_key_class(org='foo', course='bar', run='baz'), 'file')
)
self.assertEqual(
u'foo_bar_baz_base_name_ø_1974-06-22-010203',
course_and_time_based_filename_generator(
course_key_class(org='foo', course='bar', run='baz'), ' base` name ø '
)
)
class StoreUploadedFileTestCase(TestCase):
"""
Tests for store_uploaded_file.
"""
def setUp(self):
self.request = Mock(spec=HttpRequest)
self.file_content = "test file content"
self.request.FILES = {"uploaded_file": SimpleUploadedFile("tempfile.csv", self.file_content)}
self.stored_file_name = None
self.file_storage = None
self.default_max_size = 2000000
def tearDown(self):
if self.file_storage and self.stored_file_name:
self.file_storage.delete(self.stored_file_name)
def verify_exception(self, expected_message, error):
"""
Helper method to verify exception text.
"""
self.assertEqual(expected_message, error.exception.message)
def test_error_conditions(self):
"""
Verifies that exceptions are thrown in the expected cases.
"""
with self.assertRaises(ValueError) as error:
store_uploaded_file(self.request, "wrong_key", [".txt", ".csv"], "stored_file", self.default_max_size)
self.verify_exception("No file uploaded with key 'wrong_key'.", error)
with self.assertRaises(exceptions.PermissionDenied) as error:
store_uploaded_file(self.request, "uploaded_file", [], "stored_file", self.default_max_size)
self.verify_exception("The file must end with one of the following extensions: ''.", error)
with self.assertRaises(exceptions.PermissionDenied) as error:
store_uploaded_file(self.request, "uploaded_file", [".bar"], "stored_file", self.default_max_size)
self.verify_exception("The file must end with the extension '.bar'.", error)
with self.assertRaises(exceptions.PermissionDenied) as error:
store_uploaded_file(self.request, "uploaded_file", [".xxx", ".bar"], "stored_file", self.default_max_size)
self.verify_exception("The file must end with one of the following extensions: '.xxx', '.bar'.", error)
with self.assertRaises(exceptions.PermissionDenied) as error:
store_uploaded_file(self.request, "uploaded_file", [".csv"], "stored_file", 2)
self.verify_exception("Maximum upload file size is 2 bytes.", error)
def test_validator(self):
"""
Verify that a validator function can throw an exception.
"""
validator_data = {}
def verify_file_presence(should_exist):
""" Verify whether or not the stored file, passed to the validator, exists. """
self.assertEqual(should_exist, validator_data["storage"].exists(validator_data["filename"]))
def store_file_data(storage, filename):
""" Stores file validator data for testing after validation is complete. """
validator_data["storage"] = storage
validator_data["filename"] = filename
verify_file_presence(True)
def exception_validator(storage, filename):
""" Validation test function that throws an exception """
self.assertEqual("error_file.csv", os.path.basename(filename))
with storage.open(filename, 'rU') as f:
self.assertEqual(self.file_content, f.read())
store_file_data(storage, filename)
raise FileValidationException("validation failed")
def success_validator(storage, filename):
""" Validation test function that is a no-op """
self.assertEqual("success_file.csv", os.path.basename(filename))
store_file_data(storage, filename)
with self.assertRaises(FileValidationException) as error:
store_uploaded_file(
self.request, "uploaded_file", [".csv"], "error_file",
self.default_max_size, validator=exception_validator
)
self.verify_exception("validation failed", error)
# Verify the file was deleted.
verify_file_presence(False)
store_uploaded_file(
self.request, "uploaded_file", [".csv"], "success_file", self.default_max_size, validator=success_validator
)
# Verify the file still exists
verify_file_presence(True)
def test_file_upload_lower_case_extension(self):
"""
Tests uploading a file with lower case extension. Verifies that the stored file contents are correct.
"""
self.file_storage, self.stored_file_name = store_uploaded_file(
self.request, "uploaded_file", [".csv"], "stored_file", self.default_max_size
)
self._verify_successful_upload()
def test_file_upload_upper_case_extension(self):
"""
Tests uploading a file with upper case extension. Verifies that the stored file contents are correct.
"""
self.request.FILES = {"uploaded_file": SimpleUploadedFile("tempfile.CSV", self.file_content)}
self.file_storage, self.stored_file_name = store_uploaded_file(
self.request, "uploaded_file", [".gif", ".csv"], "second_stored_file", self.default_max_size
)
self._verify_successful_upload()
def _verify_successful_upload(self):
""" Helper method that checks that the stored version of the uploaded file has the correct content """
self.assertTrue(self.file_storage.exists(self.stored_file_name))
with self.file_storage.open(self.stored_file_name, 'r') as f:
self.assertEqual(self.file_content, f.read())
@ddt.ddt
class TestUniversalNewlineIterator(TestCase):
"""
Tests for the UniversalNewlineIterator class.
"""
@ddt.data(1, 2, 999)
def test_line_feeds(self, buffer_size):
self.assertEqual(
[thing for thing in UniversalNewlineIterator(StringIO(u'foo\nbar\n'), buffer_size=buffer_size)],
['foo\n', 'bar\n']
)
@ddt.data(1, 2, 999)
def test_carriage_returns(self, buffer_size):
self.assertEqual(
[thing for thing in UniversalNewlineIterator(StringIO(u'foo\rbar\r'), buffer_size=buffer_size)],
['foo\n', 'bar\n']
)
@ddt.data(1, 2, 999)
def test_carriage_returns_and_line_feeds(self, buffer_size):
self.assertEqual(
[thing for thing in UniversalNewlineIterator(StringIO(u'foo\r\nbar\r\n'), buffer_size=buffer_size)],
['foo\n', 'bar\n']
)
@ddt.data(1, 2, 999)
def test_no_trailing_newline(self, buffer_size):
self.assertEqual(
[thing for thing in UniversalNewlineIterator(StringIO(u'foo\nbar'), buffer_size=buffer_size)],
['foo\n', 'bar']
)
@ddt.data(1, 2, 999)
def test_only_one_line(self, buffer_size):
self.assertEqual(
[thing for thing in UniversalNewlineIterator(StringIO(u'foo\n'), buffer_size=buffer_size)],
['foo\n']
)
@ddt.data(1, 2, 999)
def test_only_one_line_no_trailing_newline(self, buffer_size):
self.assertEqual(
[thing for thing in UniversalNewlineIterator(StringIO(u'foo'), buffer_size=buffer_size)],
['foo']
)
@ddt.data(1, 2, 999)
def test_empty_file(self, buffer_size):
self.assertEqual(
[thing for thing in UniversalNewlineIterator(StringIO(u''), buffer_size=buffer_size)],
[]
)
@ddt.data(1, 2, 999)
def test_unicode_data(self, buffer_size):
self.assertEqual(
[thing for thing in UniversalNewlineIterator(StringIO(u'héllø wo®ld'), buffer_size=buffer_size)],
[u'héllø wo®ld']
)