test_validation_tools.py

                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_identifier"
        ) as mock_vi, mock.patch(
            "course.validation.validate_session_access_rule"
        ) as mock_vsar, mock.patch(
            "course.validation.validate_session_grading_rule"
        ) as mock_vsgr:
            mock_vsgr.side_effect = [True, True, False]

            with self.assertRaises(ValidationError) as cm:
                validation.validate_flow_rules(
                    vctx, location,
                    self.get_updated_rule(no_grading_rules=True))

            expected_error_msg = (
                "'grading' block is required if grade_identifier "
                "is not null/None.")

            self.assertIn(expected_error_msg, str(cm.exception))

        self.assertEqual(mock_vs.call_count, 1)
        self.assertEqual(mock_vi.call_count, 1)
        self.assertEqual(mock_vsar.call_count, 2)
        self.assertEqual(mock_vsgr.call_count, 0)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_grading_rules_empty(self):

        kwargs = {"grading": []}

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_identifier"
        ) as mock_vi, mock.patch(
            "course.validation.validate_session_access_rule"
        ) as mock_vsar, mock.patch(
            "course.validation.validate_session_grading_rule"
        ) as mock_vsgr:
            mock_vsgr.side_effect = [True, True, False]

            with self.assertRaises(ValidationError) as cm:
                validation.validate_flow_rules(
                    vctx, location,
                    self.get_updated_rule(**kwargs))

            expected_error_msg = "rules/grading: may not be an empty list"

            self.assertIn(expected_error_msg, str(cm.exception))

        self.assertEqual(mock_vs.call_count, 1)
        self.assertEqual(mock_vi.call_count, 1)
        self.assertEqual(mock_vsar.call_count, 2)
        self.assertEqual(mock_vsgr.call_count, 0)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_last_grading_rules_conditional(self):

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_identifier"
        ) as mock_vi, mock.patch(
            "course.validation.validate_session_access_rule"
        ) as mock_vsar, mock.patch(
            "course.validation.validate_session_grading_rule"
        ) as mock_vsgr:
            mock_vsgr.side_effect = [True, True, True]

            with self.assertRaises(ValidationError) as cm:
                validation.validate_flow_rules(
                    vctx, location,
                    self.get_updated_rule())

            expected_error_msg = (
                "rules/grading: last grading rule must be unconditional")

            self.assertIn(expected_error_msg, str(cm.exception))

        self.assertEqual(mock_vs.call_count, 1)
        self.assertEqual(mock_vi.call_count, 1)
        self.assertEqual(mock_vsar.call_count, 2)
        self.assertEqual(mock_vsgr.call_count, 3)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)


class ValidateFlowPermission(ValidationTestMixin, unittest.TestCase):
    # test validation.validate_flow_permission

    def test_success(self):
        validation.validate_flow_permission(vctx, location, "submit_answer")
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_deprecated_modify(self):
        validation.validate_flow_permission(vctx, location, "modify")

        expected_warn_msg = (
            "Uses deprecated 'modify' permission--"
            "replace by 'submit_answer' and 'end_session'")
        self.assertIn(expected_warn_msg, vctx.add_warning.call_args[0])

    def test_deprecated_see_answer(self):
        validation.validate_flow_permission(vctx, location, "see_answer")

        expected_warn_msg = (
            "Uses deprecated 'see_answer' permission--"
            "replace by 'see_answer_after_submission'")
        self.assertIn(expected_warn_msg, vctx.add_warning.call_args[0])

    def test_unknown_flow_permission(self):
        with self.assertRaises(ValidationError) as cm:
            validation.validate_flow_permission(vctx, location, "unknown")

        expected_error_msg = "invalid flow permission 'unknown'"
        self.assertIn(expected_error_msg, str(cm.exception))


class ValidateFlowDescTest(ValidationTestMixin, unittest.TestCase):
    # test validation.validate_flow_desc

    default_flow_title = "test flow title"
    default_pages = ["page1", "page2", "page3"]
    default_groups = (
        dict_to_struct(
            {"id": "flow_group1",
             "pages": ["page1", "page2", "page3"]
             }),
        dict_to_struct(
            {"id": "flow_group2",
             "pages": ["page4"]
             })
    )

    def get_updated_flow_desc(
            self, no_title=False, no_description=False, no_groups_pages=False,
            use_groups=True, use_pages=False,
            **kwargs):
        flow_desc = {}

        if not no_groups_pages:
            assert not (use_pages and use_groups)
            if use_groups:
                flow_desc["groups"] = self.default_groups
            if use_pages:
                flow_desc["pages"] = self.default_pages

        if not no_title:
            flow_desc["title"] = self.default_flow_title
        if not no_description:
            flow_desc["description"] = "hello"

        flow_desc.update(kwargs)
        return dict_to_struct(flow_desc)

    def test_success(self):

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.validation.validate_flow_group"
        ) as mock_vfg, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:

            validation.validate_flow_desc(
                vctx, location,
                self.get_updated_flow_desc())

        self.assertEqual(mock_vs.call_count, 1)
        self.assertEqual(mock_vfr.call_count, 0)
        self.assertEqual(mock_vfg.call_count, 2)
        self.assertEqual(mock_nfd.call_count, 0)
        self.assertEqual(mock_mk.call_count, 1)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_success_use_pages(self):

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.validation.validate_flow_group"
        ) as mock_vfg, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:

            mock_nfd.return_value = dict_to_struct(
                {"description": "hi",
                 "groups": [dict_to_struct(
                     {"id": "my_group",
                      "pages": [
                          dict_to_struct({"id": "page1"}),
                          dict_to_struct({"id": "page2"})]
                      })]
                 })

            validation.validate_flow_desc(
                vctx, location,
                self.get_updated_flow_desc(use_groups=False, use_pages=True))

        self.assertEqual(mock_vs.call_count, 1)
        self.assertEqual(mock_vfr.call_count, 0)
        self.assertEqual(mock_vfg.call_count, 1)
        self.assertEqual(mock_nfd.call_count, 1)
        self.assertEqual(mock_mk.call_count, 1)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_success_with_rules(self):
        kwargs = {"rules": dict_to_struct({"start": "start_rule1"})}

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.validation.validate_flow_group"
        ) as mock_vfg, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:

            validation.validate_flow_desc(
                vctx, location,
                self.get_updated_flow_desc(**kwargs))

        self.assertEqual(mock_vs.call_count, 1)
        self.assertEqual(mock_vfr.call_count, 1)
        self.assertEqual(mock_vfg.call_count, 2)
        self.assertEqual(mock_nfd.call_count, 0)
        self.assertEqual(mock_mk.call_count, 1)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_neither_groups_nor_pages(self):

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.validation.validate_flow_group"
        ) as mock_vfg, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:
            with self.assertRaises(ValidationError) as cm:
                validation.validate_flow_desc(
                    vctx, location,
                    self.get_updated_flow_desc(no_groups_pages=True))

            expected_error_msg = "must have either 'groups' or 'pages'"
            self.assertIn(expected_error_msg, str(cm.exception))

        self.assertEqual(mock_vs.call_count, 1)
        self.assertEqual(mock_vfr.call_count, 0)
        self.assertEqual(mock_vfg.call_count, 0)
        self.assertEqual(mock_nfd.call_count, 0)
        self.assertEqual(mock_mk.call_count, 0)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_both_groups_and_pages(self):

        kwargs = {"pages": self.default_pages}

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.validation.validate_flow_group"
        ) as mock_vfg, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:
            with self.assertRaises(ValidationError) as cm:
                validation.validate_flow_desc(
                    vctx, location,
                    self.get_updated_flow_desc(**kwargs))

            expected_error_msg = "must have either 'groups' or 'pages'"
            self.assertIn(expected_error_msg, str(cm.exception))

        self.assertEqual(mock_vs.call_count, 1)
        self.assertEqual(mock_vfr.call_count, 0)
        self.assertEqual(mock_vfg.call_count, 0)
        self.assertEqual(mock_nfd.call_count, 0)
        self.assertEqual(mock_mk.call_count, 0)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_group_pages_not_list(self):
        kwargs = {"groups": [
            dict_to_struct(
                {"id": "flow_group1",
                 "pages": "not pages"
                 }),
            dict_to_struct(
                {"id": "flow_group2",
                 "pages": ["page4"]
                 })]}

        with mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:
            with self.assertRaises(ValidationError) as cm:
                validation.validate_flow_desc(
                    vctx, location,
                    self.get_updated_flow_desc(**kwargs))

            expected_error_msg = (
                "'pages' has wrong type")
            self.assertIn(expected_error_msg, str(cm.exception))

        self.assertEqual(mock_vfr.call_count, 0)
        self.assertEqual(mock_nfd.call_count, 0)
        self.assertEqual(mock_mk.call_count, 1)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_group_had_no_page(self):
        kwargs = {"groups": [
            dict_to_struct(
                {"id": "flow_group1",
                 "pages": []
                 }),
            dict_to_struct(
                {"id": "flow_group2",
                 "pages": ["page4"]
                 })]}

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.validation.validate_flow_group"
        ) as mock_vfg, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:
            with self.assertRaises(ValidationError) as cm:
                validation.validate_flow_desc(
                    vctx, location,
                    self.get_updated_flow_desc(**kwargs))

            expected_error_msg = (
                "group 1 ('flow_group1'): "
                "no pages found")
            self.assertIn(expected_error_msg, str(cm.exception))

        self.assertEqual(mock_vs.call_count, 1)
        self.assertEqual(mock_vfr.call_count, 0)
        self.assertEqual(mock_vfg.call_count, 2)
        self.assertEqual(mock_nfd.call_count, 0)
        self.assertEqual(mock_mk.call_count, 0)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_flow_has_no_page(self):
        kwargs = {"groups": []}

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.validation.validate_flow_group"
        ) as mock_vfg, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:
            with self.assertRaises(ValidationError) as cm:
                validation.validate_flow_desc(
                    vctx, location,
                    self.get_updated_flow_desc(**kwargs))

            expected_error_msg = (
                "%s: no pages found" % location)
            self.assertIn(expected_error_msg, str(cm.exception))

        self.assertEqual(mock_vs.call_count, 1)
        self.assertEqual(mock_vfr.call_count, 0)
        self.assertEqual(mock_vfg.call_count, 0)
        self.assertEqual(mock_nfd.call_count, 0)
        self.assertEqual(mock_mk.call_count, 0)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_group_id_not_unique(self):
        kwargs = {"groups": [dict_to_struct(
            {"id": "flow_group1",
             "pages": ["page1", "page2", "page3"]
             }),
            dict_to_struct(
                {"id": "flow_group2",
                 "pages": ["page4"]
                 }),
            dict_to_struct(
                {"id": "flow_group2",
                 "pages": ["page5"]
                 })]}

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.validation.validate_flow_group"
        ) as mock_vfg, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:
            with self.assertRaises(ValidationError) as cm:
                validation.validate_flow_desc(
                    vctx, location,
                    self.get_updated_flow_desc(**kwargs))

            expected_error_msg = (
                "group id 'flow_group2' not unique")
            self.assertIn(expected_error_msg, str(cm.exception))

            self.assertEqual(mock_vs.call_count, 1)
            self.assertEqual(mock_vfr.call_count, 0)
            self.assertEqual(mock_vfg.call_count, 3)
            self.assertEqual(mock_nfd.call_count, 0)
            self.assertEqual(mock_mk.call_count, 0)

        # no warnings
        self.assertEqual(vctx.add_warning.call_count, 0)

    def test_completion_text(self):
        kwargs = {"completion_text": "some completion text"}

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.validation.validate_flow_group"
        ) as mock_vfg, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:

            validation.validate_flow_desc(
                vctx, location,
                self.get_updated_flow_desc(**kwargs))

            self.assertEqual(mock_vs.call_count, 1)
            self.assertEqual(mock_vfr.call_count, 0)
            self.assertEqual(mock_vfg.call_count, 2)
            self.assertEqual(mock_nfd.call_count, 0)
            self.assertEqual(mock_mk.call_count, 2)

            # no warnings
            self.assertEqual(vctx.add_warning.call_count, 0)

    def test_notify_on_submit(self):
        kwargs0 = {"notify_on_submit": []}
        kwargs = {"notify_on_submit": ["email1", "email2", ("email3",)]}

        with mock.patch(
                "course.validation.validate_struct"
        ) as mock_vs, mock.patch(
            "course.validation.validate_flow_rules"
        ) as mock_vfr, mock.patch(
            "course.validation.validate_flow_group"
        ) as mock_vfg, mock.patch(
            "course.content.normalize_flow_desc"
        ) as mock_nfd, mock.patch(
            "course.validation.validate_markup"
        ) as mock_mk:

            validation.validate_flow_desc(
                vctx, location,
                self.get_updated_flow_desc(**kwargs0))

            self.assertEqual(mock_vs.call_count, 1)
            mock_vs.reset_mock()

            self.assertEqual(mock_vfr.call_count, 0)
            mock_vfr.reset_mock()

            self.assertEqual(mock_vfg.call_count, 2)
            mock_vfg.reset_mock()

            self.assertEqual(mock_nfd.call_count, 0)
            mock_nfd.reset_mock()

            self.assertEqual(mock_mk.call_count, 1)
            mock_mk.reset_mock()

            with self.assertRaises(ValidationError) as cm:
                validation.validate_flow_desc(
                    vctx, location,
                    self.get_updated_flow_desc(**kwargs))

            expected_error_msg = "notify_on_submit: item 3 is not a string"

            self.assertIn(expected_error_msg, str(cm.exception))
            self.assertEqual(mock_vs.call_count, 1)
            self.assertEqual(mock_vfr.call_count, 0)
            self.assertEqual(mock_vfg.call_count, 2)
            self.assertEqual(mock_nfd.call_count, 0)
            self.assertEqual(mock_mk.call_count, 1)

            # no warnings
            self.assertEqual(vctx.add_warning.call_count, 0)

    def test_deprecated_attr(self):
        deprecated_attrs = [
            "max_points", "max_points_enforced_cap", "bonus_points"]

        for attr in deprecated_attrs:
            kwargs = {attr: 10}

            with mock.patch(
                    "course.validation.validate_struct"
            ) as mock_vs, mock.patch(
                "course.validation.validate_flow_rules"
            ) as mock_vfr, mock.patch(
                "course.validation.validate_flow_group"
            ) as mock_vfg, mock.patch(
                "course.content.normalize_flow_desc"
            ) as mock_nfd, mock.patch(
                "course.validation.validate_markup"
            ) as mock_mk:

                validation.validate_flow_desc(
                    vctx, location,
                    self.get_updated_flow_desc(**kwargs))

                self.assertEqual(mock_vs.call_count, 1)
                mock_vs.reset_mock()

                self.assertEqual(mock_vfr.call_count, 0)
                mock_vfr.reset_mock()

                self.assertEqual(mock_vfg.call_count, 2)
                mock_vfg.reset_mock()

                self.assertEqual(mock_nfd.call_count, 0)
                mock_nfd.reset_mock()

                self.assertEqual(mock_mk.call_count, 1)
                mock_mk.reset_mock()

                # no warnings

                expected_warn_msg = (
                    "Attribute '%s' is deprecated as part of a flow. "
                    "Specify it as part of a grading rule instead." % attr)

                self.assertIn(expected_warn_msg, vctx.add_warning.call_args[0])
                self.assertEqual(vctx.add_warning.call_count, 1)
                vctx.reset_mock()


class ValidateCalendarDescStructTest(ValidationTestMixin, unittest.TestCase):
    # test validation.validate_calendar_desc_struct

    def get_updated_event_desc(self, **kwargs):
        event_desc = {}
        event_desc.update(kwargs)
        return dict_to_struct(event_desc)

    def test_success(self):
        validation.validate_calendar_desc_struct(
            vctx, location, self.get_updated_event_desc())

    def test_has_event_kinds(self):
        validation.validate_calendar_desc_struct(
            vctx, location,
            self.get_updated_event_desc(
                event_kinds=dict_to_struct({
                    "lecture": dict_to_struct({
                        "title": "Lecture {nr}",
                        "color": "blue"
                    })
                }),
                events=dict_to_struct({
                    "lecture 1": dict_to_struct({
                        "title": "l1"})
                })
            ))

    def test_show_description_from(self):
        datespec1 = "some_date"
        validation.validate_calendar_desc_struct(
            vctx, location,
            self.get_updated_event_desc(
                events=dict_to_struct({
                    "lecture 1": dict_to_struct({
                        "title": "l1",
                        "description": "blabla",
                        "show_description_from": datespec1,
                    })
                })
            ))

        self.assertEqual(vctx.encounter_datespec.call_count, 1)
        self.assertIn(datespec1, vctx.encounter_datespec.call_args[0])
        vctx.reset_mock()

        datespec2 = "another_date"
        validation.validate_calendar_desc_struct(
            vctx, location,
            self.get_updated_event_desc(
                events=dict_to_struct({
                    "lecture 2": dict_to_struct({
                        "title": "l2",
                        "description": "blablabla",
                        "show_description_until": datespec2
                    })
                })
            ))
        self.assertEqual(vctx.encounter_datespec.call_count, 1)
        self.assertIn(datespec2, vctx.encounter_datespec.call_args[0])


class GetYamlFromRepoSafelyTest(unittest.TestCase):
    def test_success(self):
        repo = mock.MagicMock()
        full_name = "some_file"
        commit_sha = "some_commit_sha"
        with mock.patch("course.content.get_yaml_from_repo") as mock_gyfr:
            mock_gyfr.return_value = "some_string"
            validation.get_yaml_from_repo_safely(repo, full_name, commit_sha)

            _, kwargs = mock_gyfr.call_args
            self.assertEqual(kwargs["full_name"], full_name)
            self.assertEqual(kwargs["commit_sha"], commit_sha)
            self.assertEqual(kwargs["cached"], False)

    @suppress_stdout_decorator(suppress_stderr=True)
    def test_fail(self):
        repo = mock.MagicMock()
        full_name = "some_file"
        commit_sha = "some_commit_sha"
        with mock.patch("course.content.get_yaml_from_repo") as mock_gyfr:
            mock_gyfr.side_effect = RuntimeError("some error")
            with self.assertRaises(ValidationError) as cm:
                validation.get_yaml_from_repo_safely(repo, full_name, commit_sha)

            _, kwargs = mock_gyfr.call_args

            self.assertEqual(kwargs["full_name"], full_name)
            self.assertEqual(kwargs["commit_sha"], commit_sha)
            self.assertEqual(kwargs["cached"], False)

            expected_error_msg = "some_file: RuntimeError: some error"
            self.assertIn(expected_error_msg, str(cm.exception))


class CheckGradeIdentifierLinkTest(
        ValidationTestMixin, CoursesTestMixinBase, TestCase):
    # test validation.check_grade_identifier_link

    @classmethod
    def setUpTestData(cls):
        super().setUpTestData()

        cls.default_grade_identifier = "gopp1"

        cls.course1 = factories.CourseFactory(identifier="test-course1")
        cls.course2 = factories.CourseFactory(identifier="test-course2")
        cls.course1_gopp = factories.GradingOpportunityFactory(
            course=cls.course1, identifier=cls.default_grade_identifier,
            flow_id="flow1_id")

        # Ensure cross gopp independence
        factories.GradingOpportunityFactory(
            course=cls.course1, identifier="gopp2",
            flow_id="flow1_id")

        # Ensure cross course independence
        factories.GradingOpportunityFactory(
            course=cls.course2, identifier=cls.default_grade_identifier,
            flow_id="flow2_id")

    def test_success(self):
        validation.check_grade_identifier_link(
            vctx, location, self.course1, flow_id="flow1_id",
            flow_grade_identifier=self.default_grade_identifier)

    def test_fail(self):
        new_flow_id = "flow2_id"
        with self.assertRaises(ValidationError) as cm:
            validation.check_grade_identifier_link(
                vctx, location, self.course1, flow_id=new_flow_id,
                flow_grade_identifier=self.course1_gopp.identifier)

        expected_error_msg = (
            "{location}: existing grading opportunity with identifier "
            "'{grade_identifier}' refers to flow '{other_flow_id}', however "
            "flow code in this flow ('{new_flow_id}') specifies the same "
            "grade identifier. "
            "(Have you renamed the flow? If so, edit the grading "
            "opportunity to match.)".format(
                location=location,
                grade_identifier=self.course1_gopp.identifier,
                other_flow_id=self.course1_gopp.flow_id,
                new_flow_id=new_flow_id))
        self.assertIn(expected_error_msg, str(cm.exception))


class CheckForPageTypeChangesTest(
        ValidationTestMixin, CoursesTestMixinBase, TestCase):
    # test validation.check_for_page_type_changes

    flow_id = "flow_id"

    def get_updated_flow_desc(
            self, **kwargs):
        flow_desc = {"access": ["access_rule1", "access_rule2"]}
        flow_desc["title"] = "test flow title"
        flow_desc["pages"] = [
            dict_to_struct({"id": "page1", "type": "PType1"}),
            dict_to_struct({"id": "page2", "type": "PType2"}),
            dict_to_struct({"id": "page3", "type": None}),
        ]

        flow_desc.update(kwargs)
        return dict_to_struct(flow_desc)

    @classmethod
    def setUpTestData(cls):
        super().setUpTestData()
        cls.course1 = factories.CourseFactory(identifier="test-course1")
        course1_participation = factories.ParticipationFactory(course=cls.course1)
        course1_session1 = factories.FlowSessionFactory(
            course=cls.course1, participation=course1_participation,
            flow_id=cls.flow_id)
        factories.FlowPageDataFactory(
            flow_session=course1_session1, group_id="main",
            page_id="page1", page_type="PType1")
        factories.FlowPageDataFactory(
            flow_session=course1_session1, group_id="main",
            page_id="page2", page_type="PType2")

        # this entry is used to ensure page_data with page_type None
        # be excluded
        factories.FlowPageDataFactory(
            flow_session=course1_session1, group_id="main",
            page_id="page2", page_type=None)

        # Ensure cross flow_id independence
        course1_session2 = factories.FlowSessionFactory(
            course=cls.course1, participation=course1_participation,
            flow_id="another_flow_id")

        factories.FlowPageDataFactory(
            flow_session=course1_session2, group_id="main",
            page_id="page2", page_type="PType10")

        # for failure test, also to ensure page_data from different
        # group is not included
        factories.FlowPageDataFactory(
            flow_session=course1_session1, group_id="grp1",
            page_id="page2", page_type="PType2")

        # this is to ensure course is called in filter
        cls.course2 = factories.CourseFactory(identifier="test-course2")
        course2_participation = factories.ParticipationFactory(course=cls.course2)
        course2_session1 = factories.FlowSessionFactory(
            course=cls.course2, participation=course2_participation,
            flow_id=cls.flow_id)
        factories.FlowPageDataFactory(
            flow_session=course2_session1, group_id="main",
            page_id="page1", page_type="PType3", page_ordinal=1)

    def test_success(self):
        validation.check_for_page_type_changes(
            vctx, location, self.course1, self.flow_id,
            self.get_updated_flow_desc())

    def test_fail(self):
        with self.assertRaises(ValidationError) as cm:
            validation.check_for_page_type_changes(
                vctx, location, self.course2, self.flow_id,
                self.get_updated_flow_desc())

        expected_error_msg = (
            "group 'main', page 'page1': page type ('PType1') "
            "differs from type used in database ('PType3')"
        )
        self.assertIn(expected_error_msg, str(cm.exception))


class ValidateFlowIdTest(ValidationTestMixin, unittest.TestCase):
    # test validation.validate_flow_id

    def test_success(self):
        flow_id = "abc-def"
        validation.validate_flow_id(vctx, location, flow_id)
        flow_id = "abc_def1"
        validation.validate_flow_id(vctx, location, flow_id)

    def test_fail(self):
        expected_error_msg = (
            "invalid flow name. Flow names may only contain (roman) "
            "letters, numbers, dashes and underscores.")

        flow_id = "abc def"
        with self.assertRaises(ValidationError) as cm:
            validation.validate_flow_id(vctx, location, flow_id)
        self.assertIn(expected_error_msg, str(cm.exception))

        flow_id = "abc/def"
        with self.assertRaises(ValidationError) as cm:
            validation.validate_flow_id(vctx, location, flow_id)
        self.assertIn(expected_error_msg, str(cm.exception))


class ValidateStaticPageNameTest(ValidationTestMixin, unittest.TestCase):
    # test validation.validate_static_page_name

    def test_success(self):
        page_name = "abc-def"
        validation.validate_static_page_name(vctx, location, page_name)

        page_name = "abc_def1"
        validation.validate_static_page_name(vctx, location, page_name)

    def test_fail(self):
        expected_error_msg = (
            "invalid page name. Page names may only contain alphanumeric "
            "characters (any language) and hyphens.")

        page_name = "abc/def"

        with self.assertRaises(ValidationError) as cm:
            validation.validate_static_page_name(vctx, location, page_name)
        self.assertIn(expected_error_msg, str(cm.exception))


VALID_ATTRIBUTES_YML = """
unenrolled:
    - test1.pdf
student:
    - test2.pdf
"""

INVALID_ATTRIBUTES_YML = """
unenrolled:
    - test1.pdf
student:
    - test2.pdf
    - 42
"""

VALID_ATTRIBUTES_WITH_PUBLIC_YML = """
public:
    - test1.pdf
student:
    - test2.pdf
"""

INVALID_ATTRIBUTES_WITH_PUBLIC_AND_UNENROLLED_YML = """
public:
    - test1.pdf
unenrolled:
    - test2.pdf
"""

GITIGNORE = b"""a_dir
another_dir/*"""

GITIGNORE_COMMENTED = b"""
# a_dir
"""

default_access_kinds = ["public", "in_exam", "student", "ta",
                        "unenrolled", "instructor"]


class FakeBlob:
    def __init__(self, data):
        self.data = data


class FakeRepo:
    def __setitem__(self, key, value):
        self.__dict__[key] = value

    def __getitem__(self, item):
        return self.__dict__[item]


class CheckAttributesYmlTest(ValidationTestMixin, unittest.TestCase):
    # test validation.check_attributes_yml

    def setUp(self):
        super().setUp()
        patch = mock.patch("course.content.get_true_repo_and_path")
        self.mock_get_true_repo_and_path = patch.start()

        self.mock_get_true_repo_and_path.return_value = (vctx.repo, "")
        self.addCleanup(patch.stop)

    def test_success_with_no_attributes_yml_and_no_gitignore(self):
        path = ""
        repo = vctx.repo
        tree = Tree()
        validation.check_attributes_yml(
            vctx, repo, path, tree, default_access_kinds)

    def test_success_with_no_attributes_yml_and_no_gitignore_root_only_contains_a_subfolder(self):  # noqa
        path = ""
        repo = vctx.repo
        tree = Tree()

        tree.add(b"a_dir", stat.S_IFDIR,
                 hashlib.sha224(b"a dir").hexdigest().encode())
        validation.check_attributes_yml(
            vctx, repo, path, tree, default_access_kinds)
        self.assertEqual(vctx.add_warning.call_count, 0)

        # make sure check_attributes_yml is called recursively
        self.assertEqual(
            self.mock_get_true_repo_and_path.call_count, 2,
            "check_attributes_yml is expected to be called recursively")
        self.assertEqual(
            self.mock_get_true_repo_and_path.call_args[0][1], "a_dir",
            "check_attributes_yml is expected call with subfolder'a_dir'"
        )

    def test_success_with_attributes_yml(self):
        path = ""
        repo = vctx.repo
        tree = Tree()
        tree.add(ATTRIBUTES_FILENAME.encode(), stat.S_IFREG,
                 b".attributes.yml_content")

        fake_repo = FakeRepo()
        fake_repo[b".attributes.yml_content"] = FakeBlob(VALID_ATTRIBUTES_YML)
        self.mock_get_true_repo_and_path.return_value = (fake_repo, "")

        validation.check_attributes_yml(
            vctx, repo, path, tree, default_access_kinds)

    def test_attributes_yml_public_deprecated(self):
        path = ""
        repo = vctx.repo
        tree = Tree()
        tree.add(ATTRIBUTES_FILENAME.encode(), stat.S_IFREG,
                 b".attributes.yml_content")

        fake_repo = FakeRepo()
        fake_repo[b".attributes.yml_content"] = FakeBlob(
            VALID_ATTRIBUTES_WITH_PUBLIC_YML)
        self.mock_get_true_repo_and_path.return_value = (fake_repo, "")

        validation.check_attributes_yml(
            vctx, repo, path, tree, default_access_kinds)
        self.assertEqual(vctx.add_warning.call_count, 1)

        expected_warn_msg = ("Access class 'public' is deprecated. "
                             "Use 'unenrolled' instead.")
        self.assertIn(expected_warn_msg, vctx.add_warning.call_args[0])

    def test_failure_with_invalid_attributes_yml(self):
        path = ""
        repo = vctx.repo