base_test_mixins.py

    flow_id = QUIZ_FLOW_ID

    @classmethod
    def update_default_flow_session_id(cls, course_identifier):
        cls._default_session_id = cls.default_flow_params["flow_session_id"]

    @classmethod
    def get_default_flow_session_id(cls, course_identifier):
        if cls._default_session_id is not None:
            return cls._default_session_id
        cls._default_session_id = cls.get_latest_session_id(course_identifier)
        return cls._default_session_id


class TwoCoursePageTestMixin(TwoCourseTestMixin):
    _course1_default_session_id = None
    _course2_default_session_id = None

    @property
    def flow_id(self):
        raise NotImplementedError

    @classmethod
    def get_default_flow_session_id(cls, course_identifier):
        if course_identifier == cls.course1.identifier:
            if cls._course1_default_session_id is not None:
                return cls._course1_default_session_id
            cls._course1_default_session_id = (
                cls.get_last_session_id(course_identifier))
            return cls._course1_default_session_id
        if course_identifier == cls.course2.identifier:
            if cls._course2_default_session_id is not None:
                return cls._course2_default_session_id
            cls._course2_default_session_id = (
                cls.get_last_session_id(course_identifier))
            return cls._course2_default_session_id

    @classmethod
    def update_default_flow_session_id(cls, course_identifier):
        new_session_id = cls.default_flow_params["flow_session_id"]
        if course_identifier == cls.course1.identifier:
            cls._course1_default_session_id = new_session_id
        elif course_identifier == cls.course2.identifier:
            cls._course2_default_session_id = new_session_id


class SingleCourseQuizPageTestMixin(SingleCoursePageTestMixin):

    skip_code_question = True

    @classmethod
    def ensure_grading_ui_get(cls, page_id):
        with cls.temporarily_switch_to_user(cls.instructor_participation.user):
            url = cls.get_page_grading_url_by_page_id(page_id)
            resp = cls.c.get(url)
            assert resp.status_code == 200

    @classmethod
    def ensure_analytic_page_get(cls, group_id, page_id):
        with cls.temporarily_switch_to_user(cls.instructor_participation.user):
            resp = cls.get_flow_page_analytics(
                flow_id=cls.flow_id, group_id=group_id,
                page_id=page_id)
            assert resp.status_code == 200

    @classmethod
    def ensure_download_submission(
            cls, group_id, page_id, dl_file_extension=None,
            file_with_ext_count=None):
        with cls.temporarily_switch_to_user(cls.instructor_participation.user):
            group_page_id = "%s/%s" % (group_id, page_id)
            resp = cls.post_download_all_submissions_by_group_page_id(
                group_page_id=group_page_id, flow_id=cls.flow_id)
            assert resp.status_code == 200
            prefix, zip_file = resp["Content-Disposition"].split('=')
            assert prefix == "attachment; filename"
            assert resp.get('Content-Type') == "application/zip"
            if dl_file_extension:
                buf = six.BytesIO(resp.content)
                import zipfile
                with zipfile.ZipFile(buf, 'r') as zf:
                    assert zf.testzip() is None
                    # todo: make more assertions in terms of file content

                    for f in zf.filelist:
                        assert f.file_size > 0

                    if file_with_ext_count is None:
                        assert len([f for f in zf.filelist if
                                    f.filename.endswith(dl_file_extension)]) > 0, \
                            ("The zipped file unexpectedly didn't contain "
                             "file with extension '%s', the actual file list "
                             "is %s" % (
                                 dl_file_extension,
                                 repr([f.filename for f in zf.filelist])))
                    else:
                        assert (
                                len([f for f in zf.filelist if
                                     f.filename.endswith(dl_file_extension)])
                                == file_with_ext_count), \
                            ("The zipped file unexpectedly didn't contain "
                             "%d files with extension '%s', the actual file list "
                             "is %s" % (
                                 file_with_ext_count,
                                 dl_file_extension,
                                 repr([f.filename for f in zf.filelist])))

    @classmethod
    def submit_page_answer_by_ordinal_and_test(
            cls, page_ordinal, use_correct_answer=True, answer_data=None,
            skip_code_question=True,
            expected_grades=None, expected_post_answer_status_code=200,
            do_grading=False, do_human_grade=False, grade_data=None,
            grade_data_extra_kwargs=None,
            dl_file_extension=None,
            ensure_grading_ui_get_before_grading=False,
            ensure_grading_ui_get_after_grading=False,
            ensure_analytic_page_get_before_submission=False,
            ensure_analytic_page_get_after_submission=False,
            ensure_analytic_page_get_before_grading=False,
            ensure_analytic_page_get_after_grading=False,
            ensure_download_before_submission=False,
            ensure_download_after_submission=False,
            ensure_download_before_grading=False,
            ensure_download_after_grading=False,
            dl_file_with_ext_count=None):
        page_id = cls.get_page_id_via_page_oridnal(page_ordinal)

        return cls.submit_page_answer_by_page_id_and_test(
            page_id, use_correct_answer,
            answer_data, skip_code_question, expected_grades,
            expected_post_answer_status_code,
            do_grading, do_human_grade,
            grade_data, grade_data_extra_kwargs, dl_file_extension,
            ensure_grading_ui_get_before_grading,
            ensure_grading_ui_get_after_grading,
            ensure_analytic_page_get_before_submission,
            ensure_analytic_page_get_after_submission,
            ensure_analytic_page_get_before_grading,
            ensure_analytic_page_get_after_grading,
            ensure_download_before_submission,
            ensure_download_after_submission,
            ensure_download_before_grading,
            ensure_download_after_grading,
            dl_file_with_ext_count)

    @classmethod
    def submit_page_answer_by_page_id_and_test(
            cls, page_id, use_correct_answer=True, answer_data=None,
            skip_code_question=True,
            expected_grades=None, expected_post_answer_status_code=200,
            do_grading=False, do_human_grade=False, grade_data=None,
            grade_data_extra_kwargs=None,
            dl_file_extension=None,
            ensure_grading_ui_get_before_grading=False,
            ensure_grading_ui_get_after_grading=False,
            ensure_analytic_page_get_before_submission=False,
            ensure_analytic_page_get_after_submission=False,
            ensure_analytic_page_get_before_grading=False,
            ensure_analytic_page_get_after_grading=False,
            ensure_download_before_submission=False,
            ensure_download_after_submission=False,
            ensure_download_before_grading=False,
            ensure_download_after_grading=False,
            dl_file_with_ext_count=None):

        if answer_data is not None:
            assert isinstance(answer_data, dict)
            use_correct_answer = False

        submit_answer_response = None
        post_grade_response = None

        for page_tuple in TEST_PAGE_TUPLE:
            if skip_code_question and page_tuple.need_runpy:
                continue
            if page_id == page_tuple.page_id:
                group_id = page_tuple.group_id
                if ensure_grading_ui_get_before_grading:
                    cls.ensure_grading_ui_get(page_id)

                if ensure_analytic_page_get_before_submission:
                    cls.ensure_analytic_page_get(group_id, page_id)

                if ensure_download_before_submission:
                    cls.ensure_download_submission(group_id, page_id)

                if page_tuple.correct_answer is not None:

                    if answer_data is None:
                        answer_data = page_tuple.correct_answer

                    if page_id in ["anyup", "proof"]:
                        file_path = answer_data["uploaded_file"]
                        if not file_path:
                            # submitting an empty answer
                            submit_answer_response = (
                                cls.post_answer_by_page_id(page_id, answer_data))
                        else:
                            if isinstance(file_path, list):
                                file_path, = file_path

                            file_path = file_path.strip()
                            with open(file_path, 'rb') as fp:
                                answer_data = {"uploaded_file": fp}
                                submit_answer_response = (
                                    cls.post_answer_by_page_id(
                                        page_id, answer_data))
                    else:
                        submit_answer_response = (
                            cls.post_answer_by_page_id(page_id, answer_data))

                    # Fixed #514
                    # https://github.com/inducer/relate/issues/514
                    submit_answer_response.context["form"].as_p()

                    assert (submit_answer_response.status_code
                            == expected_post_answer_status_code), (
                            "%s != %s" % (submit_answer_response.status_code,
                                          expected_post_answer_status_code))

                    if ensure_analytic_page_get_after_submission:
                        cls.ensure_analytic_page_get(group_id, page_id)

                    if ensure_download_after_submission:
                        cls.ensure_download_submission(group_id, page_id)

                if not do_grading:
                    break

                assert cls.end_flow().status_code == 200

                if ensure_analytic_page_get_before_grading:
                    cls.ensure_analytic_page_get(group_id, page_id)

                if ensure_download_before_grading:
                    cls.ensure_download_submission(group_id, page_id)

                if page_tuple.correct_answer is not None:
                    if use_correct_answer:
                        expected_grades = page_tuple.full_points

                    if page_tuple.need_human_grade:
                        if not do_human_grade:
                            cls.assertSessionScoreEqual(None)
                            break
                        if grade_data is not None:
                            assert isinstance(grade_data, dict)
                        else:
                            grade_data = page_tuple.grade_data.copy()

                        if grade_data_extra_kwargs:
                            assert isinstance(grade_data_extra_kwargs, dict)
                            grade_data.update(grade_data_extra_kwargs)

                        post_grade_response = cls.post_grade_by_page_id(
                            page_id, grade_data)
                    cls.assertSessionScoreEqual(expected_grades)

                    if not dl_file_extension:
                        dl_file_extension = page_tuple.dl_file_extension

                    if ensure_download_after_grading:
                        cls.ensure_download_submission(
                            group_id, page_id,
                            dl_file_extension=dl_file_extension,
                            file_with_ext_count=dl_file_with_ext_count)

                if ensure_analytic_page_get_after_grading:
                    cls.ensure_analytic_page_get(group_id, page_id)

                if ensure_grading_ui_get_after_grading:
                    cls.ensure_grading_ui_get(page_id)

        return submit_answer_response, post_grade_response

    def default_submit_page_answer_by_page_id_and_test(self, page_id,
                                                       answer_data=None,
                                                       expected_grade=None,
                                                       do_grading=True,
                                                       grade_data=None,
                                                       grade_data_extra_kwargs=None,
                                                       ):
        return self.submit_page_answer_by_page_id_and_test(
            page_id, answer_data=answer_data,
            skip_code_question=self.skip_code_question,
            expected_grades=expected_grade, expected_post_answer_status_code=200,
            do_grading=do_grading, do_human_grade=True, grade_data=grade_data,
            grade_data_extra_kwargs=grade_data_extra_kwargs,
            ensure_grading_ui_get_before_grading=True,
            ensure_grading_ui_get_after_grading=True,
            ensure_analytic_page_get_before_submission=True,
            ensure_analytic_page_get_after_submission=True,
            ensure_analytic_page_get_before_grading=True,
            ensure_analytic_page_get_after_grading=True,
            ensure_download_before_submission=True,
            ensure_download_after_submission=True,
            ensure_download_before_grading=True,
            ensure_download_after_grading=True)

    @classmethod
    def submit_page_human_grading_by_page_id_and_test(
            cls, page_id,
            expected_post_grading_status_code=200,
            grade_data=None,
            expected_grades=None,
            do_session_score_equal_assersion=True,
            grade_data_extra_kwargs=None,
            force_login_instructor=True,
            ensure_grading_ui_get_before_grading=False,
            ensure_grading_ui_get_after_grading=False,
            ensure_analytic_page_get_before_grading=False,
            ensure_analytic_page_get_after_grading=False,
            ensure_download_before_grading=False,
            ensure_download_after_grading=False):

        # this helper is expected to be used when the session is finished

        post_grade_response = None

        for page_tuple in TEST_PAGE_TUPLE:
            if page_id == page_tuple.page_id:
                group_id = page_tuple.group_id
                if ensure_grading_ui_get_before_grading:
                    cls.ensure_grading_ui_get(page_id)

                if ensure_analytic_page_get_before_grading:
                    cls.ensure_analytic_page_get(group_id, page_id)

                if ensure_download_before_grading:
                    cls.ensure_download_submission(group_id, page_id)

                if not page_tuple.need_human_grade:
                    break

                assign_full_grades = True

                if grade_data is not None:
                    assert isinstance(grade_data, dict)
                    assign_full_grades = False
                else:
                    grade_data = page_tuple.grade_data.copy()

                if assign_full_grades:
                    expected_grades = page_tuple.full_points

                if grade_data_extra_kwargs:
                    assert isinstance(grade_data_extra_kwargs, dict)
                    grade_data.update(grade_data_extra_kwargs)

                post_grade_response = cls.post_grade_by_page_id(
                    page_id, grade_data,
                    force_login_instructor=force_login_instructor)

                assert (post_grade_response.status_code
                        == expected_post_grading_status_code)

                if post_grade_response.status_code == 200:
                    if do_session_score_equal_assersion:
                        cls.assertSessionScoreEqual(expected_grades)

                if ensure_download_after_grading:
                    cls.ensure_download_submission(group_id, page_id)

                if ensure_analytic_page_get_after_grading:
                    cls.ensure_analytic_page_get(group_id, page_id)

                if ensure_grading_ui_get_after_grading:
                    cls.ensure_grading_ui_get(page_id)

        return post_grade_response


class MockAddMessageMixing(object):
    """
    The mixing for testing django.contrib.messages.add_message
    """

    def setUp(self):
        super(MockAddMessageMixing, self).setUp()
        self._fake_add_message_path = "django.contrib.messages.add_message"
        fake_add_messag = mock.patch(self._fake_add_message_path)

        self._mock_add_message = fake_add_messag.start()
        self.addCleanup(fake_add_messag.stop)

    def _get_added_messages(self, join=True):
        try:
            msgs = [
                "'%s'" % str(arg[2])
                for arg, _ in self._mock_add_message.call_args_list]
        except IndexError:
            self.fail("%s is unexpectedly not called." % self._fake_add_message_path)
        else:
            if join:
                return "; ".join(msgs)
            return msgs

    def assertAddMessageCallCount(self, expected_call_count, reset=False):  # noqa
        fail_msg = (
            "%s is unexpectedly called %d times, instead of %d times." %
            (self._fake_add_message_path, self._mock_add_message.call_count,
             expected_call_count))
        if self._mock_add_message.call_count > 0:
            fail_msg += ("The called messages are: %s"
                         % repr(self._get_added_messages(join=False)))
        self.assertEqual(
            self._mock_add_message.call_count, expected_call_count, msg=fail_msg)
        if reset:
            self._mock_add_message.reset_mock()

    def assertAddMessageCalledWith(self, expected_messages, reset=True):  # noqa
        joined_msgs = self._get_added_messages()

        if not isinstance(expected_messages, list):
            expected_messages = [expected_messages]

        not_called = []
        for msg in expected_messages:
            if msg not in joined_msgs:
                not_called.append(msg)

        if not_called:
            fail_msg = "%s unexpectedly not added in messages. " % repr(not_called)
            if joined_msgs:
                fail_msg += "the actual message are \"%s\"" % joined_msgs
            self.fail(fail_msg)
        if reset:
            self._mock_add_message.reset_mock()

    def assertAddMessageNotCalledWith(self, expected_messages, reset=False):  # noqa
        joined_msgs = self._get_added_messages()

        if not isinstance(expected_messages, list):
            expected_messages = [expected_messages]

        called = []
        for msg in expected_messages:
            if msg in joined_msgs:
                called.append(msg)

        if called:
            fail_msg = "%s unexpectedly added in messages. " % repr(called)
            fail_msg += "the actual message are \"%s\"" % joined_msgs
            self.fail(fail_msg)
        if reset:
            self._mock_add_message.reset_mock()

    def reset_add_message_mock(self):
        self._mock_add_message.reset_mock()


class SubprocessRunpyContainerMixin(object):
    """
    This mixin is used to fake a runpy container, only needed when
    the TestCase include test(s) for code questions
    """
    @classmethod
    def setUpClass(cls):  # noqa
        if not may_run_expensive_tests():
            from unittest import SkipTest
            raise SkipTest(SKIP_EXPENSIVE_TESTS_REASON)

        super(SubprocessRunpyContainerMixin, cls).setUpClass()

        python_executable = os.getenv("PY_EXE")

        if not python_executable:
            python_executable = sys.executable

        import subprocess
        args = [python_executable,
                os.path.abspath(
                    os.path.join(
                        os.path.dirname(__file__), os.pardir,
                        "docker-image-run-py", "runcode")),
                ]
        cls.faked_container_process = subprocess.Popen(
            args,
            stdout=subprocess.DEVNULL,

            # because runpy prints to stderr
            stderr=subprocess.DEVNULL
        )

    def setUp(self):
        super(SubprocessRunpyContainerMixin, self).setUp()
        self.faked_container_patch = mock.patch(
            "course.page.code.SPAWN_CONTAINERS", False)
        self.faked_container_patch.start()
        self.addCleanup(self.faked_container_patch.stop)

    @classmethod
    def tearDownClass(cls):  # noqa
        super(SubprocessRunpyContainerMixin, cls).tearDownClass()

        from course.page.code import SPAWN_CONTAINERS
        # Make sure SPAWN_CONTAINERS is reset to True
        assert SPAWN_CONTAINERS
        if sys.platform.startswith("win"):
            # Without these lines, tests on Appveyor hanged when all tests
            # finished.
            # However, On nix platforms, these lines resulted in test
            # failure when there were more than one TestCases which were using
            # this mixin. So we don't kill the subprocess, and it won't bring
            # bad side effects to remainder tests.
            cls.faked_container_process.kill()


def improperly_configured_cache_patch():
    # can be used as context manager or decorator
    if six.PY3:
        built_in_import_path = "builtins.__import__"
        import builtins  # noqa
    else:
        built_in_import_path = "__builtin__.__import__"
        import __builtin__ as builtins  # noqa
    built_in_import = builtins.__import__

    def my_disable_cache_import(name, globals=None, locals=None, fromlist=(),
                                level=0):
        if name == "django.core.cache":
            raise ImproperlyConfigured()
        return built_in_import(name, globals, locals, fromlist, level)

    return mock.patch(built_in_import_path, side_effect=my_disable_cache_import)


# {{{ admin

ADMIN_TWO_COURSE_SETUP_LIST = deepcopy(TWO_COURSE_SETUP_LIST)
# switch roles
ADMIN_TWO_COURSE_SETUP_LIST[1]["participations"][0]["role_identifier"] = "ta"
ADMIN_TWO_COURSE_SETUP_LIST[1]["participations"][1]["role_identifier"] = "instructor"  # noqa


class AdminTestMixin(TwoCourseTestMixin):
    courses_setup_list = ADMIN_TWO_COURSE_SETUP_LIST
    none_participation_user_create_kwarg_list = (
        NONE_PARTICIPATION_USER_CREATE_KWARG_LIST)

    @classmethod
    def setUpTestData(cls):  # noqa
        super(AdminTestMixin, cls).setUpTestData()  # noqa

        # create 2 participation (with new user) for course1
        from tests.factories import ParticipationFactory

        cls.course1_student_participation2 = (
            ParticipationFactory.create(course=cls.course1))
        cls.course1_student_participation3 = (
            ParticipationFactory.create(course=cls.course1))
        cls.instructor1 = cls.course1_instructor_participation.user
        cls.instructor2 = cls.course2_instructor_participation.user
        assert cls.instructor1 != cls.instructor2

        # grant all admin permissions to instructors
        from django.contrib.auth.models import Permission

        for user in [cls.instructor1, cls.instructor2]:
            user.is_staff = True
            user.save()
            for perm in Permission.objects.all():
                user.user_permissions.add(perm)

    @classmethod
    def get_admin_change_list_view_url(cls, app_name, model_name):
        return reverse("admin:%s_%s_changelist" % (app_name, model_name))

    @classmethod
    def get_admin_change_view_url(cls, app_name, model_name, args=None):
        if args is None:
            args = []
        return reverse("admin:%s_%s_change" % (app_name, model_name), args=args)

    @classmethod
    def get_admin_add_view_url(cls, app_name, model_name, args=None):
        if args is None:
            args = []
        return reverse("admin:%s_%s_add" % (app_name, model_name), args=args)

    def get_admin_form_fields(self, response):
        """
        Return a list of AdminFields for the AdminForm in the response.
        """
        admin_form = response.context['adminform']
        fieldsets = list(admin_form)

        field_lines = []
        for fieldset in fieldsets:
            field_lines += list(fieldset)

        fields = []
        for field_line in field_lines:
            fields += list(field_line)

        return fields

    def get_admin_form_fields_names(self, response):
        return [f.field.name for f in self.get_admin_form_fields(response)]

    def get_changelist(self, request, model, model_admin):
        from django.contrib.admin.views.main import ChangeList
        return ChangeList(
            request, model, model_admin.list_display,
            model_admin.list_display_links, model_admin.get_list_filter(request),
            model_admin.date_hierarchy, model_admin.search_fields,
            model_admin.list_select_related, model_admin.list_per_page,
            model_admin.list_max_show_all, model_admin.list_editable,
            model_admin=model_admin,
            sortable_by=model_admin.sortable_by
        )

    def get_filterspec_list(self, request, changelist=None, model=None,
                            model_admin=None):
        if changelist is None:
            assert request and model and model_admin
            changelist = self.get_changelist(request, model, model_admin)

        filterspecs = changelist.get_filters(request)[0]
        filterspec_list = []
        for filterspec in filterspecs:
            choices = tuple(c['display'] for c in filterspec.choices(changelist))
            filterspec_list.append(choices)

        return filterspec_list

# }}}


class HackRepoMixin(object):

    # This is need to for correctly getting other blobs
    fallback_commit_sha = b"4124e0c23e369d6709a670398167cb9c2fe52d35"

    # This need to be configured when the module tested imported get_repo_blob
    # at module level
    get_repo_blob_patching_path = "course.content.get_repo_blob"

    @classmethod
    def setUpTestData(cls):  # noqa
        super(HackRepoMixin, cls).setUpTestData()

        class Blob(object):
            def __init__(self, yaml_file_name):
                with open(os.path.join(FAKED_YAML_PATH, yaml_file_name), "rb") as f:
                    data = f.read()
                self.data = data

        def get_repo_side_effect(repo, full_name, commit_sha, allow_tree=True):
            commit_sha_path_maps = COMMIT_SHA_MAP.get(full_name)
            if commit_sha_path_maps:
                assert isinstance(commit_sha_path_maps, list)
                for cs_map in commit_sha_path_maps:
                    if commit_sha.decode() in cs_map:
                        path = cs_map[commit_sha.decode()]["path"]
                        return Blob(path)

            return get_repo_blob(repo, full_name, cls.fallback_commit_sha,
                                 allow_tree=allow_tree)

        cls.batch_fake_get_repo_blob = mock.patch(cls.get_repo_blob_patching_path)
        cls.mock_get_repo_blob = cls.batch_fake_get_repo_blob.start()
        cls.mock_get_repo_blob.side_effect = get_repo_side_effect

    @classmethod
    def tearDownClass(cls):  # noqa
        # This must be done to avoid inconsistency
        super(HackRepoMixin, cls).tearDownClass()
        cls.batch_fake_get_repo_blob.stop()

    def get_current_page_ids(self):
        current_sha = self.course.active_git_commit_sha
        for commit_sha_path_maps in COMMIT_SHA_MAP.values():
            for cs_map in commit_sha_path_maps:
                if current_sha in cs_map:
                    return cs_map[current_sha]["page_ids"]

        raise ValueError("Page_ids for that commit_sha doesn't exist")

    def assertGradeInfoEqual(self, resp, expected_grade_info_dict=None):  # noqa
        grade_info = resp.context["grade_info"]

        assert isinstance(grade_info, GradeInfo)
        if not expected_grade_info_dict:
            import json
            error_msg = ("\n%s" % json.dumps(OrderedDict(
                sorted(
                    [(k, v) for (k, v) in six.iteritems(grade_info.__dict__)])),
                indent=4))
            error_msg = error_msg.replace("null", "None")
            self.fail(error_msg)

        assert isinstance(expected_grade_info_dict, dict)

        grade_info_dict = grade_info.__dict__
        not_match_infos = []
        for k in grade_info_dict.keys():
            if grade_info_dict[k] != expected_grade_info_dict[k]:
                not_match_infos.append(
                    "'%s' is expected to be %s, while got %s"
                    % (k, str(expected_grade_info_dict[k]),
                       str(grade_info_dict[k])))

        if not_match_infos:
            self.fail("\n".join(not_match_infos))

# vim: fdm=marker