test_versioning.py

            ValidationError(my_validation_error_msg))

        self.check_command_message_result(
            command="update",
            add_message_expected_call_count=1,
            expected_add_message_literals=[
                VALIDATE_FAILURE_LITERAL, my_validation_error_msg],
            not_expected_add_message_literals=[PREVIEW_END_LITERAL]
        )
        self.assertIsNone(self.participation.preview_git_commit_sha)
        self.assertEqual(self.course.active_git_commit_sha, self.default_old_sha)

    def test_update_with_validation_error_previewing(self):
        from course.validation import ValidationError
        my_validation_error_msg = "my validation error."
        self.mock_validate_course_content.side_effect = (
            ValidationError(my_validation_error_msg))

        self.check_command_message_result(
            command="update",
            add_message_expected_call_count=1,
            is_previewing=True,
            expected_add_message_literals=[
                VALIDATE_FAILURE_LITERAL, my_validation_error_msg],
            not_expected_add_message_literals=[PREVIEW_END_LITERAL]
        )
        self.assertIsNotNone(self.participation.preview_git_commit_sha)
        self.assertEqual(
            self.participation.preview_git_commit_sha, self.default_preview_sha)
        self.assertEqual(self.course.active_git_commit_sha, self.default_old_sha)

    def test_fetch_not_may_update(self):
        self.mock_client.fetch.return_value = {
            b"HEAD": self.default_lastest_sha.encode()}

        command_tup = (
            ("fetch", 1, [FETCHED_LITERAL], [UPDATE_APPLIED_LITERAL],
             self.default_old_sha),
            ("fetch_update", 2, [FETCHED_LITERAL, VALIDATE_SUCCESS_LITERAL],
             [UPDATE_APPLIED_LITERAL],
             self.default_old_sha)
        )

        for (command, add_message_call_count, expected, not_expected,
             expected_course_sha) in command_tup:
            with self.subTest(command=command):
                self.check_command_message_result(
                    command=command,
                    add_message_expected_call_count=add_message_call_count,
                    expected_add_message_literals=expected,
                    not_expected_add_message_literals=not_expected,
                    may_update=False
                )

                self.check_command_message_result(
                    command=command,
                    add_message_expected_call_count=add_message_call_count,
                    expected_add_message_literals=expected,
                    not_expected_add_message_literals=not_expected,
                    may_update=False
                )

                self.assertEqual(
                    self.course.active_git_commit_sha, expected_course_sha)


class VersioningRepoMixin(object):
    @classmethod
    def setUpTestData(cls):  # noqa
        super(VersioningRepoMixin, cls).setUpTestData()
        cls.rf = RequestFactory()
        request = cls.rf.get(cls.get_update_course_url())
        request.user = cls.instructor_participation.user

        from course.utils import CoursePageContext
        pctx = CoursePageContext(request, cls.course.identifier)
        cls.repo = pctx.repo


class GitUpdateFormTest(VersioningRepoMixin, SingleCourseTestMixin, TestCase):

    # Todo: test inner format_commit
    def test_may_not_update(self):
        form = versioning.GitUpdateForm(
            may_update=False, previewing=True, repo=self.repo)
        from crispy_forms.layout import Submit
        submit_input_names = [
            input.name for input in form.helper.inputs
            if isinstance(input, Submit)
        ]

        self.assertNotIn("fetch_update", submit_input_names)
        self.assertNotIn("update", submit_input_names)

    def test_may_update(self):
        form = versioning.GitUpdateForm(
            may_update=True, previewing=True, repo=self.repo)
        from crispy_forms.layout import Submit
        submit_input_names = [
            input.name for input in form.helper.inputs
            if isinstance(input, Submit)
        ]

        self.assertIn("fetch_update", submit_input_names)
        self.assertIn("update", submit_input_names)

    def test_not_previewing(self):
        form = versioning.GitUpdateForm(
            may_update=True, previewing=False, repo=self.repo)
        from crispy_forms.layout import Submit
        submit_input_names = [
            input.name for input in form.helper.inputs
            if isinstance(input, Submit)
        ]

        self.assertNotIn("end_preview", submit_input_names)

    def test_previewing(self):
        form = versioning.GitUpdateForm(
            may_update=True, previewing=True, repo=self.repo)
        from crispy_forms.layout import Submit
        submit_input_names = [
            input.name for input in form.helper.inputs
            if isinstance(input, Submit)
        ]

        self.assertIn("end_preview", submit_input_names)


class GetCommitMessageAsHtmlTest(VersioningRepoMixin, SingleCourseTestMixin,
                                 TestCase):
    # test versioning._get_commit_message_as_html
    def test_result_text(self):
        commit_sha = "593a1cdcecc6f4759fd5cadaacec0ba9dd0715a7"
        expected_msg = ("Added an optional_page, another "
                        "PythonCodeQuestionWithHumanTextFeedback page.")
        self.assertEqual(
            versioning._get_commit_message_as_html(self.repo, commit_sha),
            expected_msg)

    def test_result_byte(self):
        commit_sha = b"593a1cdcecc6f4759fd5cadaacec0ba9dd0715a7"
        expected_msg = ("Added an optional_page, another "
                        "PythonCodeQuestionWithHumanTextFeedback page.")
        self.assertEqual(
            versioning._get_commit_message_as_html(self.repo, commit_sha),
            expected_msg)

    def test_non_exist_commit(self):
        commit_sha = b"unknown"
        expected_msg = "- not found -"

        self.assertEqual(
            versioning._get_commit_message_as_html(self.repo, commit_sha),
            expected_msg)

    def test_escape_html(self):
        commit_sha_1 = b"a_commit"
        commit_sha_2 = b"another_commit"
        commit_sha_3 = b"yet_another_commit"
        repo_dict = {
            commit_sha_1: FakeCommit("a_commit", message=b"test a > b  "),
            commit_sha_2: FakeCommit("another_commit", message=b"  <p>test</p>"),
            commit_sha_3: FakeCommit("another_commit", message=b"abc\\uDC80"),
        }
        repo = mock.MagicMock()
        repo.__getitem__.side_effect = repo_dict.__getitem__
        repo.__setitem__.side_effect = repo_dict.__setitem__

        expected_msg = "test a &gt; b"
        self.assertEqual(
            versioning._get_commit_message_as_html(repo, commit_sha_1),
            expected_msg)

        expected_msg = "&lt;p&gt;test&lt;/p&gt;"
        self.assertEqual(
            versioning._get_commit_message_as_html(repo, commit_sha_2),
            expected_msg)

        expected_msg = "abc\\uDC80"
        self.assertEqual(
            versioning._get_commit_message_as_html(repo, commit_sha_3),
            expected_msg)


class UpdateCourseTest(SingleCourseTestMixin, MockAddMessageMixing, TestCase):
    def test_no_permission(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.c.get(self.get_update_course_url())
            self.assertEqual(resp.status_code, 403)

            for command in versioning.ALLOWED_COURSE_REVISIOIN_COMMANDS:
                resp = self.post_update_course_content(
                    "some_commit_sha", command=command,
                    force_login_instructor=False)
                self.assertEqual(resp.status_code, 403)

    def test_participation_with_preview_permission(self):
        # Just to make sure it won't fail, Todo: assersion on form kwargs
        from course.models import ParticipationPermission
        pp = ParticipationPermission(
            participation=self.student_participation,
            permission=pperm.preview_content)
        pp.save()
        self.student_participation.individual_permissions.set([pp])

        with self.temporarily_switch_to_user(self.student_participation.user):
            for command in versioning.ALLOWED_COURSE_REVISIOIN_COMMANDS:
                resp = self.post_update_course_content(
                    "some_commit_sha", command=command,
                    force_login_instructor=False)
                self.assertEqual(resp.status_code, 200, command)

    def test_participation_with_update_permission(self):
        # Just to make sure it won't fail, Todo: assersion on form kwargs
        from course.models import ParticipationPermission
        pp = ParticipationPermission(
            participation=self.student_participation,
            permission=pperm.update_content)
        pp.save()
        self.student_participation.individual_permissions.set([pp])

        with self.temporarily_switch_to_user(self.student_participation.user):
            for command in versioning.ALLOWED_COURSE_REVISIOIN_COMMANDS:
                resp = self.post_update_course_content(
                    "some_commit_sha", command=command,
                    force_login_instructor=False)
                self.assertEqual(resp.status_code, 200, command)

    def test_get(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.c.get(self.get_update_course_url())
            self.assertEqual(resp.status_code, 200)

    def test_unknown_command(self):
        resp = self.post_update_course_content(
            "some_commit_sha", command="unknown")
        self.assertEqual(resp.status_code, 400)

    @suppress_stdout_decorator(suppress_stderr=True)
    def test_run_course_update_command_failure(self):
        with mock.patch(
            "course.versioning.run_course_update_command"
        ) as mock_run_update:
            error_msg = "my runtime error"
            mock_run_update.side_effect = RuntimeError(error_msg)
            resp = self.post_update_course_content(
                self.course.active_git_commit_sha, command="update")
            self.assertEqual(resp.status_code, 200)
            self.assertAddMessageCallCount(1)
            expected_error_msg = "Error: RuntimeError %s" % error_msg
            self.assertAddMessageCalledWith(expected_error_msg)

    def test_form_not_valid(self):
        with mock.patch(
                "course.versioning.GitUpdateForm.is_valid"
        ) as mock_form_valid, mock.patch(
            "course.versioning.run_course_update_command"
        ) as mock_run_update:
            mock_form_valid.return_value = False
            resp = self.post_update_course_content(
                "some_commit_sha", command="update")
            self.assertEqual(resp.status_code, 200)
            self.assertEqual(mock_run_update.call_count, 0)

    def test_repo_is_in_subdir(self):
        self.course.course_root_path = "/subdir"
        self.course.save()

        from course.content import get_course_repo, SubdirRepoWrapper
        self.assertIsInstance(get_course_repo(self.course), SubdirRepoWrapper)

        with mock.patch(
            "course.versioning.run_course_update_command"
        ) as mock_run_update:
            self.post_update_course_content(
                self.course.active_git_commit_sha, command="update")

            self.assertEqual(mock_run_update.call_count, 1)

            from course.content import SubdirRepoWrapper
            from dulwich.repo import Repo
            self.assertIsInstance(mock_run_update.call_args[0][1], Repo)