def get_student_list_csv(experiment, file_title):
    def row_generator():
        yield ["Student Name", "Student ID", "Experiment", "Assigned Track",
                   "Timestamp Last Updated"]
        for s in ExperimentStudent.objects.filter(experiment=experiment):
            yield [s.student_name, s.student_id, s.experiment.name,
                   s.track.name, s.updated_on]
    return streamed_csv_response(row_generator(), file_title) def get_intervention_point_interactions_csv(experiment, file_title):
    def row_generator():
        yield ["Student Name", "Student ID", "Experiment", "Assigned Track",
                   "Intervention Point", "Intervention Point URL", "Timestamp Encountered"]
        for i in InterventionPointInteraction.objects.filter(experiment=experiment):
            yield [i.student.student_name, i.student.student_id,
                   i.experiment.name, i.track.name,
                   i.intervention_point.name, i.url, i.created_on]
    return streamed_csv_response(row_generator(), file_title) def get_track_selection_xlsx(request, experiment, file_title="test.xlsx"):
    output = StringIO.StringIO()
    workbook = xlsxwriter.Workbook(output, {"in_memory": True})
    worksheet = workbook.add_worksheet()
    worksheet.set_column(0, 3, 20)
    headers = ["Student Name", "Student ID", "Experiment", "Assigned Track"]
    worksheet.write_row(0, 0, headers)
    students = get_unassigned_students(request, experiment)
    # Row offsets of +1 below are to account for header
    for i, student_id in enumerate(students):
        row = [students[student_id], student_id, experiment.name, ""]
        worksheet.write_row(i + 1, 0, row)
    # Add drop-down with track names as the only valid options for the missing column
    track_names = [t.name for t in experiment.tracks.all()]
    worksheet.data_validation(1, 3, len(students) + 1, 3,
                              {'validate': 'list', 'source': track_names,})
    workbook.close()
    output.seek(0)
    content_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
    response = HttpResponse(output.read(), content_type=content_type)
    response['Content-Disposition'] = ("attachment; filename=%s" % file_title)
    return response def get_track_selection_csv(request, experiment, file_title="test.xlsx"):
    def row_generator():
        yield ["Student Name", "Student ID", "Experiment", "Assigned Track"]
        # Write data to CSV file
        for student_id, student_name in get_unassigned_students(request, experiment).items():
            yield [student_name, student_id, experiment.name, ""]
    return streamed_csv_response(row_generator(), file_title) def parse_uploaded_file(experiment, unassigned_students, input_spreadsheet, filename):
    """ This function parses an uploaded spreadsheet of student track assignments.
        It returns a dictionary `students`, mapping sis_ids to track
        names as well as a list of errors (in string form) encountered while
        parsing the spreadsheet.  If errors is empty, no errors were encountered. """
    students = {}
    errors = []
    track_names = {track.name: track for track in experiment.tracks.all()}
    if filename.endswith('.csv'):
        reader = csv.reader(input_spreadsheet.splitlines())
        reader.next()  # Skip the header
        for row_number, row in enumerate(reader):
            # Row number is off by 2 because we trimmed the headers and
            # CSV files are 1-indexed while enumerate is 0-indexed
            parse_row(row, row_number + 2, experiment, track_names,
                      unassigned_students, students, errors)
    elif filename.endswith('.xlsx') or filename.endswith('.xls'):
        book = xlrd.open_workbook(file_contents=input_spreadsheet)
        sheet = book.sheet_by_index(0)
        # Start at 1 to skip headers
        for row_number in range(1, sheet.nrows):
            row = sheet.row_values(row_number)
            parse_row(row, row_number, experiment, track_names,
                      unassigned_students, students, errors)
    else:
        raise INVALID_FILE_TYPE
    return students, errors def parse_row(row, row_number, experiment, tracks, unassigned_students, students, errors):
    """ The arguments `students` and `errors` are each data structures
        from `parse_uploaded_file` indented to be modified by this function.
        If an error is encountered in parsing the row, a string representing the
        error is appended to the list `errors`.  If there are no errors,
        an entry (sis_id -> track_name) should be added to the `students` dict """
    # Skip empty rows
    if not row:
        return
    if len(row) < 4 or row[3] == "" or row[3] is None:
        errors.append("Row %s: missing track name" % (row_number))
    elif row[3] not in tracks:
        errors.append("Row %s: invalid track name '%s'" % (row_number, row[3]))
    elif row[2] != experiment.name:
        errors.append("Row %s: wrong experiment name '%s'" % (row_number, row[2]))
    elif row[1] not in unassigned_students:
        errors.append("Row %s: student '%s' not available for assignment" % (row_number, row[0]))
    else:
        students[row[1]] = tracks[row[3]] class InterventionPointUrl(TimestampedModel):
    """ This model stores the URL of a single intervention """
    url = models.URLField(max_length=URL_CHAR_LIMIT)
    track = models.ForeignKey(Track)
    intervention_point = models.ForeignKey(InterventionPoint)
    open_as_tab = models.BooleanField(default=False)
    is_canvas_page = models.BooleanField(default=False)
    
    class Meta:
        unique_together = (('track', 'intervention_point'),) class ExperimentStudent(CourseObject):
    """ This model stores which track a student is in for a given experiment
        within a given course.  A real-world student can be represented by
        multiple ExperimentStudent objects, and will have a separate object for
        each experiment they are in. """
    student_id = models.CharField(max_length=128, db_index=True)
    student_name = models.CharField(max_length=256, db_index=True, null=True)
    experiment = models.ForeignKey(Experiment, related_name="students")
    track = models.ForeignKey(Track)
    
    class Meta:
        unique_together = (('experiment', 'student_id'),) class Experiment(CourseObject):
    UNIFORM_RANDOM = 1
    WEIGHTED_PROBABILITY_RANDOM = 2
    CSV_UPLOAD = 3
    REVERSE_API = 4
    
    ASSIGNMENT_ENUM_TYPES = (
        (UNIFORM_RANDOM, "uniform_random"),
        (WEIGHTED_PROBABILITY_RANDOM, "weighted_probability_random"),
        (CSV_UPLOAD, "csv_upload"),
        (REVERSE_API, "reverse_api"),
    )
    
    name = models.CharField(max_length=NAME_CHAR_LIMIT)
    notes = models.TextField()
    tracks_finalized = models.BooleanField(default=False)
    assignment_method = models.IntegerField(max_length=1, default=1,
                                            choices=ASSIGNMENT_ENUM_TYPES,)
    
    class Meta:
        unique_together = (('course_id', 'name'),)
    
    @classmethod
    def get_all_started_csv(cls):
        return cls.objects.filter(tracks_finalized=True,
                                  assignment_method=Experiment.CSV_UPLOAD)
    
    def get_course_notification(self):
        """ The CourseNotification object should exist, at the point at which
            an experiment has been created under the same course, so this method
            will raise an exception if the CourseNotification object doesn't
            exist yet."""
        return CourseNotification.objects.get(course_id=self.course_id)
    
    def assert_not_finalized(self):
        if self.tracks_finalized:
            raise EXPERIMENT_TRACKS_ALREADY_FINALIZED
    
    def new_track(self, track_name):
        return Track.objects.create(course_id=self.course_id, experiment=self,
                                    name=track_name)
    
    def get_incomplete_intervention_point_names(self):
        return [point.name for point in self.intervention_points.all() if point.is_missing_urls()]
    
    def to_json(self):
        """ Converts the experiment and its associated tracks to json,
            in the form expected by the edit_experiment.html template and
            by edit_experiment.js """
        experiment_dict = {
            "id": self.id,
            "name": self.name,
            "notes": self.notes,
            "uniformRandom": bool(self.assignment_method == self.UNIFORM_RANDOM),
            "csvUpload": bool(self.assignment_method == self.CSV_UPLOAD),
            "tracks": [{"id": t.id,
                        "weighting": t.get_weighting(),
                        "name": t.name,
                        # databaseName stores the track name in the database for collision comparison
                        "databaseName": t.name,
                        # newName and editing are used to hold temporary values for track name editing
                        "newName" : t.name,
                        "editing": False,
                        "deleteURL": reverse('ab_testing_tool_delete_track', args=(t.id,))}
                       for t in self.tracks.all()],
        }
        return json.dumps(experiment_dict)
    
    def copy(self, new_name):
        tracks = self.tracks.all()
        intervention_points = self.intervention_points.all()
        # Copy Experiment
        self.save_as_new_object(name=new_name, tracks_finalized=False)
        
        # Copy Tracks
        track_id_mapping = {}
        for track in tracks:
            original_track_id = track.id
            track.save_as_new_object(experiment=self)
            track_id_mapping[original_track_id] = track
            # Copy TrackProbabilityWeight, if any
            try:
                weight = TrackProbabilityWeight.objects.get(track_id=original_track_id)
                weight.save_as_new_object(track=track)
            except TrackProbabilityWeight.DoesNotExist:
                pass
        
        # Copy InterventionPoints
        for intervention_point in intervention_points:
            orig_ip_id = intervention_point.id
            intervention_point.save_as_new_object(experiment=self)
            # Copy InterventionPointUrls, if any
            for ip_url in InterventionPointUrl.objects.filter(intervention_point_id=orig_ip_id):
                ip_url.save_as_new_object(track=track_id_mapping[ip_url.track.id],
                                          intervention_point=intervention_point)
    
    @classmethod
    def get_placeholder_course_experiment(cls, course_id):
        """ Gets or creates a single experiment for the course.  Placeholder
            method until interface supports multiple experiments.
            TODO: Remove once multiple experiments are supported """
        return Experiment.objects.get_or_create(course_id=course_id, name="Experiment 1")[0] class InterventionPointInteraction(CourseObject):
    """ This model logs every time an intervention point is deployed for a
        student.  Consider moving this out of the database and into
        flat file storage. """
    student = models.ForeignKey(ExperimentStudent)
    intervention_point = models.ForeignKey(InterventionPoint)
    experiment = models.ForeignKey(Experiment, related_name="intervention_point_interactions")
    track = models.ForeignKey(Track)
    url = models.URLField(max_length=URL_CHAR_LIMIT) class SessionTestCase(TestCase):
    def setUp(self):
        # http://code.djangoproject.com/ticket/10899
        settings.SESSION_ENGINE = "django.contrib.sessions.backends.file"
        engine = import_module(settings.SESSION_ENGINE)
        store = engine.SessionStore()
        store.save()
        self.session = store
        self.client.cookies[settings.SESSION_COOKIE_NAME] = store.session_key
        
        settings.LOGGING = {}
        
        lti_launch = {}
        lti_launch["roles"] = ADMINS
        lti_launch["custom_canvas_course_id"] = TEST_COURSE_ID
        lti_launch["lis_course_offering_sourcedid"] = "92345"
        lti_launch["custom_canvas_api_domain"] = TEST_DOMAIN
        lti_launch["launch_presentation_return_url"] = TEST_DOMAIN
        lti_launch["custom_canvas_user_login_id"] = TEST_STUDENT_ID
        lti_launch["lis_person_name_full"] = TEST_STUDENT_NAME
        lti_launch["context_title"] = "Course title"
        lti_launch["lis_person_contact_email_primary"] = TEST_EMAIL
        session = self.client.session
        session["LTI_LAUNCH"] = lti_launch
        session.save()
        
        self.request = RequestMock()
        self.request.session = session
        
        # Patches api functions for all tests; can be overridden by re-patching
        # the particular api call for a particular test
        patchers = [
            patch(LIST_MODULES, return_value=APIReturn([])),
            patch(LIST_ITEMS, return_value=APIReturn([])),
            patch(GET_TOKEN, return_value="MOCK_TOKEN"),
        ]
        for patcher in patchers:
            patcher.start()
            self.addCleanup(patcher.stop)
    
    def set_lti_param(self, param_name, param_value):
        """ Convenience method to set an LIT_LAUNCH parameter in the mock session """
        session = self.client.session
        session["LTI_LAUNCH"][param_name] = param_value
        session.save()
    
    def set_roles(self, roles):
        self.set_lti_param("roles", roles)
    
    def assertOkay(self, response):
        self.assertEqual(response.status_code, 200)
    
    def assertError(self, response, exception_instance):
        self.assertTemplateUsed(response, ERROR_TEMPLATE)
        if isinstance(exception_instance, RenderableError):
            self.assertEqual(response.status_code, exception_instance.status_code)
        else:
            self.assertEqual(response.status_code, DEFAULT_ERROR_STATUS)
        self.assertIn("message", response.context)
        self.assertEqual(response.context["message"], str(exception_instance))
    
    def assertSameIds(self, iterable_1, iterable_2, duplicates_allowed=False):
        """ Checks that the two iterables have the same set of .id attributes
            among their items; if duplicates_allowed is False, the test
            will pass if either iterable contains duplicate_ids within itself """
        ids_1 = [i.id for i in iterable_1]
        ids_2 = [i.id for i in iterable_2]
        if not duplicates_allowed:
            self.assertEqual(len(set(ids_1)), len(ids_1))
            self.assertEqual(len(set(ids_2)), len(ids_2))
        self.assertEqual(set(ids_1), set(ids_2))
    
    def assertRaisesSpecific(self, exception_instance, func, *args, **kwargs):
        """ Asserts that the function with the passed args and kwargs,
            raises an exception that matches the exception_instance.
            Match means has the same message, and is either the same class as
            exception_instance or a child of exception_instance's class. """
        error = False
        self.assertRaises(exception_instance.__class__, func, *args, **kwargs)
        try:
            func(*args, **kwargs)
            error = True
            
        except Exception as e:
            self.assertEquals(str(e), str(exception_instance))
        if error:
            raise Exception("Function either raises no exception or is "
                            "non-deterministic")
    
    def create_test_experiment(self, course_id=TEST_COURSE_ID, name="testexperiment",
                               assignment_method=Experiment.UNIFORM_RANDOM, **kwargs):
        return Experiment.objects.create(name=name, course_id=course_id,
                                         assignment_method=assignment_method,
                                         **kwargs)
    
    def create_test_track(self, course_id=TEST_COURSE_ID, name="testtrack", experiment=None):
        if not experiment:
            experiment = Experiment.get_placeholder_course_experiment(course_id)
        return Track.objects.create(name=name, course_id=course_id,
                                    experiment=experiment)
    
    def create_test_track_weight(self, course_id=TEST_COURSE_ID, weighting=1,
                                 track=None, experiment=None):
        if not experiment:
            experiment = Experiment.get_placeholder_course_experiment(course_id)
        if not track:
            track = Track.objects.create(name="testtrack", course_id=course_id,
                                         experiment=experiment)
        return TrackProbabilityWeight.objects.create(
                course_id=course_id, weighting=weighting, track=track,
                experiment=experiment)
    
    def create_test_intervention_point(self, course_id=TEST_COURSE_ID, name="testip",
                                       experiment=None):
        if not experiment:
            experiment = Experiment.get_placeholder_course_experiment(course_id)
        return InterventionPoint.objects.create(name=name, course_id=course_id,
                                                experiment=experiment) TEST_COURSE_ID = "12345" TEST_STUDENT_ID = "70707707" class Renderable404(RenderableError, Http404):
    status_code = 404
from django.test import RequestFactory
from ab_tool.spreadsheets import (get_student_list_csv, get_intervention_point_interactions_csv,
                                  get_track_selection_xlsx, get_track_selection_csv,
                                  parse_uploaded_file, parse_row)
from ab_tool.models import (InterventionPointUrl, ExperimentStudent, Experiment,
                            InterventionPointInteraction)
from ab_tool.tests.common import (SessionTestCase, TEST_COURSE_ID, TEST_STUDENT_ID)
from mock import patch, Mock
from error_middleware.exceptions import Renderable404




'''
Setup some test data
'''
TEST_XLS_FILE_NAME = 'test.xls'
TEST_XLSX_FILE_NAME = 'test.xlsx'
TEST_CSV_FILE_NAME = 'test.csv'
TEST_DOMAIN = 'http://www.example.com'
TEST_STUDENT_EMAIL = 'StudentB@example.com'
TEST_STUDENT_DICT = {'10123478': 'StudentA',
                     '20123278': 'StudentB',
                     '34512278': 'StudentC',
                     '40212478': 'StudentD',
}
TEST_CONTENT_TYPE = [('Content-Type', 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'),
                     ('Content-Disposition', 'attachment; filename=test.xlsx')]

TEST_TRACK_SELECTION_RESPONSE = ['Student Name,Student ID,Experiment,Assigned Track\r\n',
                                 'StudentD,40212478,Experiment 1,\r\n',
                                 'StudentB,20123278,Experiment 1,\r\n',
                                 'StudentA,10123478,Experiment 1,\r\n',
                                 'StudentC,34512278,Experiment 1,\r\n']

TEST_TRACK_SELECTION_UPLOAD = ['Student Name,Student ID,Experiment,Assigned Track\r\n',
                                 'StudentD,40212478,Experiment 1,track1\r\n',
                                 'StudentB,20123278,Experiment 1,track1\r\n',
                                 'StudentA,10123478,Experiment 1,track1\r\n',
                                 'StudentC,34512278,Experiment 1,track1\r\n']

TEST_ROW = [TEST_STUDENT_DICT['20123278'], '20123278', 'Experiment 1', 'track1']

TEST_ROW_NUMBER = 1


class TestSpreadsheets(SessionTestCase):
    def setUp(self):
        self.request = RequestFactory().get('/fake-path')
        self.request.user = Mock(name='user_mock')
        self.request.session = {
            'oauth_return_uri': TEST_DOMAIN,
            'LTI_LAUNCH': {
                'custom_canvas_api_domain': TEST_DOMAIN,
                'lis_person_contact_email_primary': TEST_STUDENT_EMAIL,
                'custom_canvas_course_id': TEST_COURSE_ID,
            },
        }


        self.experiment = Experiment.get_placeholder_course_experiment(TEST_COURSE_ID)