def intervention_point_url(request, intervention_point_id):
    """ Builds a URL to deploy the intervention_point with the database id
        intervention_point_id """
    try:
        intervention_point_id = int(intervention_point_id)
    except (TypeError, ValueError):
        raise BAD_INTERVENTION_POINT_ID
    return request.build_absolute_uri(reverse("ab_testing_tool_deploy_intervention_point",
                                              args=(intervention_point_id,))) def validate_format_url(url):
    validator = URLValidator()
    """ Adds "http://" to the beginning of a url if it isn't there """
    url = url.strip()
    if not url.startswith("http://") and not url.startswith("https://"):
        url = "http://%s" % url
    try:
        #the validator enforces standard URL length requirement
        validator(url)
        return url
    except ValidationError:
        raise INVALID_URL_PARAM def post_param(request, param_name):
    if param_name in request.POST:
        return request.POST[param_name]
    raise missing_param_error(param_name) def assign_track_and_create_student(experiment, student_id, student_name):
    """ Method looks up assignment_mode to select track and creates student in selected track.
        Method returns student object """
    if experiment.assignment_method == Experiment.CSV_UPLOAD:
        # Raise error as student should have already been assigned a track if CSV upload
        #TODO: infrastructure needed notify course administrator about incomplete student-track mapping
        raise CSV_UPLOAD_NEEDED
    tracks = experiment.tracks.all()
    if not tracks:
        raise NO_TRACKS_FOR_EXPERIMENT
    if experiment.assignment_method == Experiment.UNIFORM_RANDOM:
        # If uniform, pick randomly from the set of tracks
        chosen_track = choice(tracks)
    if experiment.assignment_method == Experiment.WEIGHTED_PROBABILITY_RANDOM:
        # If weighted, generate a weighted list of tracks and pick one randomly from list
        weighted_tracks = []
        for track in tracks:
            try:
                track_weight = TrackProbabilityWeight.objects.get(track=track, experiment=experiment)
            except TrackProbabilityWeight.DoesNotExist:
                raise TRACK_WEIGHTS_NOT_SET
            weighted_tracks.extend([track] * track_weight.weighting)
        chosen_track = choice(weighted_tracks)
    # Create student with chosen track
    student = ExperimentStudent.objects.create(
            student_id=student_id, course_id=experiment.course_id,
            track=chosen_track, student_name=student_name,
            experiment=experiment
    )
    return student def validate_name(name):
    if not name:
        raise MISSING_NAME_PARAM
    if len(name) > NAME_CHAR_LIMIT:
        raise PARAM_LENGTH_EXCEEDS_LIMIT
    return name def validate_weighting(weighting):
    """ Track weights need to be an integer between 1 and 100 """
    weighting = int(weighting)
    if not 0 <= weighting <=100:
        raise INCORRECT_WEIGHTING_PARAM
    return weighting 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_STUDENT_ID = "70707707" TEST_STUDENT_NAME = "Test Student" BAD_INTERVENTION_POINT_ID = Renderable400(ADMIN_VISIBLE_ERROR) CSV_UPLOAD_NEEDED = Renderable403(STUDENT_VISIBLE_ERROR) NO_TRACKS_FOR_EXPERIMENT = Renderable403(STUDENT_VISIBLE_ERROR) TRACK_WEIGHTS_NOT_SET = Renderable403(STUDENT_VISIBLE_ERROR) INVALID_URL_PARAM = Renderable400("One of your URL inputs is invalid.") MISSING_NAME_PARAM = Renderable400("Names are required. Please go back and submit a name for all experiments, tracks, and intervention points.") PARAM_LENGTH_EXCEEDS_LIMIT = Renderable400("There are length limitations. At least one of your inputs is too long.") INCORRECT_WEIGHTING_PARAM = Renderable400("Weights (between 0 and 100) are required if you are randomizing with weights. Please go back and submit weights for all tracks.") 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 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'),)
from mock import MagicMock
from ab_tool.controllers import (intervention_point_url,
    validate_format_url, post_param, assign_track_and_create_student,
    validate_name, validate_weighting)
from ab_tool.tests.common import (SessionTestCase, TEST_STUDENT_ID, TEST_STUDENT_NAME)
from ab_tool.exceptions import (BAD_INTERVENTION_POINT_ID, CSV_UPLOAD_NEEDED,
    NO_TRACKS_FOR_EXPERIMENT, TRACK_WEIGHTS_NOT_SET,
    INVALID_URL_PARAM, MISSING_NAME_PARAM, PARAM_LENGTH_EXCEEDS_LIMIT,
    INCORRECT_WEIGHTING_PARAM)
from ab_tool.models import Experiment, ExperimentStudent



class TestControllers(SessionTestCase):
    def test_assign_track_and_create_student_creates_student(self):
        """ Tests function assign_track_and_create_student creates student object
            via DB count """
        experiment = self.create_test_experiment(assignment_method=Experiment.UNIFORM_RANDOM)
        self.create_test_track(experiment=experiment)
        student_count = ExperimentStudent.objects.count()
        student = assign_track_and_create_student(experiment, TEST_STUDENT_ID, TEST_STUDENT_ID)
        self.assertTrue(ExperimentStudent.objects.count() == student_count + 1 )
        self.assertTrue(student in ExperimentStudent.objects.all())
    
    def test_assign_track_and_create_student_with_weights(self):
        experiment = self.create_test_experiment(assignment_method=Experiment.WEIGHTED_PROBABILITY_RANDOM)
        track = self.create_test_track(experiment=experiment)
        self.create_test_track_weight(track=track, experiment=experiment)
        student_count = ExperimentStudent.objects.count()
        student = assign_track_and_create_student(experiment, TEST_STUDENT_ID, TEST_STUDENT_ID)
        self.assertTrue(ExperimentStudent.objects.count() == student_count + 1 )
        self.assertTrue(student in ExperimentStudent.objects.all())
        self.assertTrue(student.track == track)

    def test_assign_track_and_create_student_without_weights_raises_error(self):
        experiment = self.create_test_experiment(assignment_method=Experiment.WEIGHTED_PROBABILITY_RANDOM)
        self.create_test_track(experiment=experiment)
        data = {"experiment": experiment, "student_id":TEST_STUDENT_ID,
                "student_name": TEST_STUDENT_NAME}
        self.assertRaisesSpecific(TRACK_WEIGHTS_NOT_SET, assign_track_and_create_student, **data)
    
    def test_assign_track_and_create_student_raises_error_for_no_tracks(self):
        experiment = self.create_test_experiment()
        data = {"experiment": experiment, "student_id":TEST_STUDENT_ID,
                "student_name": TEST_STUDENT_NAME}
        self.assertRaisesSpecific(NO_TRACKS_FOR_EXPERIMENT, assign_track_and_create_student, **data)
    
    def test_assign_track_and_create_student_raises_error_for_no_csv(self):
        experiment = self.create_test_experiment(assignment_method=Experiment.CSV_UPLOAD)
        data = {"experiment": experiment, "student_id":TEST_STUDENT_ID,
                "student_name": TEST_STUDENT_NAME}
        self.assertRaisesSpecific(CSV_UPLOAD_NEEDED, assign_track_and_create_student, **data)
    
    def test_intervention_point_url_contains_intervention_point_id(self):
        """ Tests that intervention_point_url contains the string of the intervention_point_id """
        intervention_point_id = 999888777

        url = intervention_point_url(self.request, intervention_point_id)