test_utils.py

                      self.mock_eval_generic_conditions.call_args[1])

        self.assertEqual(
            self.mock_eval_generic_conditions.call_args[1]["login_exam_ticket"],
            fake_login_exam_ticket
        )

    def test_not_passing_eval_participation_tags_conditions(self):
        self.mock_get_flow_rules.return_value = [mock.MagicMock()]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = False
        result = self.get_result()
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_not_for_rollover_and_if_in_facility(self):
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_in_facility": "f1"})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(facilities=frozenset(["f2"]))
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_not_for_rollover_and_if_has_in_progress_session(self):
        factories.FlowSessionFactory(
            participation=self.student_participation, flow_id=self.flow_id)
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_in_facility": "f1",
                            "if_has_in_progress_session": 2})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(facilities=frozenset(["f1", "f2"]))
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_not_for_rollover_and_if_has_session_tagged(self):
        factories.FlowSessionFactory(
            participation=self.student_participation, flow_id=self.flow_id,
            in_progress=True)
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_has_in_progress_session": 1,
                            "if_has_session_tagged": "atag1"})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result()
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_not_for_rollover_and_if_has_fewer_sessions_than(self):
        factories.FlowSessionFactory(
            participation=self.student_participation, flow_id=self.flow_id,
            access_rules_tag="atag1"
        )
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_has_session_tagged": "atag1",
                            "if_has_fewer_sessions_than": 1})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result()
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_not_for_rollover_and_if_has_fewer_tagged_sessions_than(self):
        factories.FlowSessionFactory.create_batch(size=2,
            participation=self.student_participation, flow_id=self.flow_id,
            access_rules_tag="atag1")
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_has_fewer_sessions_than": 3,
                            "if_has_fewer_tagged_sessions_than": 1})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result()
        self.assertRuleEqual(self.fallback_rule, result)

    def test_passing_not_for_rollover_and_if_has_fewer_tagged_sessions_than(self):
        factories.FlowSessionFactory.create_batch(size=2,
            participation=self.student_participation, flow_id=self.flow_id,
            access_rules_tag="atag1")
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_has_fewer_tagged_sessions_than": 3})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result()
        self.assertRuleEqual(
            result,
            {"tag_session": None,
            "may_start_new_session": True,
            "may_list_existing_sessions": True,
            "default_expiration_mode": None}
        )

    def test_passing_not_for_rollover(self):
        factories.FlowSessionFactory.create_batch(size=2,
            participation=self.student_participation, flow_id=self.flow_id)
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result()
        self.assertRuleEqual(
            result,
            {"tag_session": None,
            "may_start_new_session": True,
            "may_list_existing_sessions": True,
            "default_expiration_mode": None}
        )

    def test_passing_for_rollover(self):
        factories.FlowSessionFactory.create_batch(size=2,
            participation=self.student_participation, flow_id=self.flow_id)
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(for_rollover=True)
        self.assertRuleEqual(
            result,
            {"tag_session": None,
            "may_start_new_session": True,
            "may_list_existing_sessions": True,
            "default_expiration_mode": None}
        )

    def test_get_expected_rule(self):
        tag_session = mock.MagicMock()
        default_expiration_mode = mock.MagicMock()
        may_start_new_session = mock.MagicMock()
        may_list_existing_sessions = mock.MagicMock()
        self.mock_get_flow_rules.return_value = [
            dict_to_struct(
                {"tag_session": tag_session,
                 "default_expiration_mode": default_expiration_mode,
                 "may_start_new_session": may_start_new_session,
                 "may_list_existing_sessions": may_list_existing_sessions
                 })]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True

        # simplified
        result = self.get_result(for_rollover=True)
        self.assertRuleEqual(
            result,
            {"tag_session": tag_session,
             "may_start_new_session": may_start_new_session,
             "may_list_existing_sessions": may_list_existing_sessions,
             "default_expiration_mode": default_expiration_mode}
        )


class GetSessionAccessRuleTest(GetSessionRuleMixin, SingleCourseTestMixin, TestCase):
    # test utils.get_session_access_rule
    call_func = utils.get_session_access_rule
    rule_klass = utils.FlowSessionAccessRule

    fallback_rule = utils.FlowSessionAccessRule(permissions=frozenset())
    default_permissions = [fperm.view]

    @property
    def default_kwargs(self):
        return {
            "session": self.fs1,
            "flow_desc": mock.MagicMock(),
            "now_datetime": self.now,
            "facilities": None,
            "login_exam_ticket": None,
        }

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

        cls.now = now() - timedelta(days=1)

        start_time = cls.now - timedelta(minutes=60)

        cls.ta_participation.time_factor = 1.1
        cls.ta_participation.save()

        cls.fs1 = factories.FlowSessionFactory(
            participation=cls.student_participation, in_progress=False,
            expiration_mode=constants.flow_session_expiration_mode.end,
            start_time=start_time
        )
        cls.fs2 = factories.FlowSessionFactory(
            participation=cls.ta_participation, in_progress=True,
            expiration_mode=constants.flow_session_expiration_mode.roll_over,
            start_time=start_time
        )
        cls.fs3 = factories.FlowSessionFactory(
            course=cls.course,
            participation=None, in_progress=True, user=None,
            expiration_mode=constants.flow_session_expiration_mode.roll_over,
            start_time=start_time
        )

    def get_result(self, **extra_kwargs):
        kwargs = self.get_updated_kwargs(**extra_kwargs)
        return utils.get_session_access_rule(**kwargs)

    def get_default_rule(self, **kwargs):
        defaults = {
            "permissions": self.default_permissions[:],
            "message": None,
        }
        defaults.update(kwargs)
        return utils.FlowSessionAccessRule(**defaults)

    def test_no_rules(self):
        self.mock_get_flow_rules.return_value = []
        result = self.get_result()
        self.assertRuleEqual(self.fallback_rule, result)

        # make sure get_flow_rules is called with expected default_rules_desc
        self.assertEqual(self.mock_get_flow_rules.call_count, 1)
        self.assertIn("default_rules_desc", self.mock_get_flow_rules.call_args[1])

        default_rules_desc = (
            self.mock_get_flow_rules.call_args[1]["default_rules_desc"])
        self.assertEqual(
            default_rules_desc[0].permissions, self.default_permissions)

    def test_not_passing_eval_generic_conditions(self):
        self.mock_get_flow_rules.return_value = [mock.MagicMock()]
        self.mock_eval_generic_conditions.return_value = False

        fake_login_exam_ticket = mock.MagicMock()
        result = self.get_result(login_exam_ticket=fake_login_exam_ticket)
        self.assertRuleEqual(self.fallback_rule, result)

        # make sure _eval_generic_conditions is called with expected
        # login_exam_ticket
        self.assertEqual(self.mock_eval_generic_conditions.call_count, 1)
        self.assertIn("login_exam_ticket",
                      self.mock_eval_generic_conditions.call_args[1])

        self.assertEqual(
            self.mock_eval_generic_conditions.call_args[1]["login_exam_ticket"],
            fake_login_exam_ticket
        )

    def test_not_passing_eval_participation_tags_conditions(self):
        self.mock_get_flow_rules.return_value = [mock.MagicMock()]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = False
        result = self.get_result()
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_eval_generic_session_conditions(self):
        self.mock_get_flow_rules.return_value = [mock.MagicMock()]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        self.mock_eval_generic_session_conditions.return_value = False
        result = self.get_result()
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_if_in_facility(self):
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_in_facility": "f1",
                            "permissions": mock.MagicMock()
                            })]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(facilities=frozenset(["f2"]))
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_if_in_progress(self):
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_in_facility": "f1",
                            "if_in_progress": True,
                            "permissions": mock.MagicMock()
                            })]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(facilities=frozenset(["f1", "f2"]))
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_if_expiration_mode(self):
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_in_progress": True,
                            "if_expiration_mode":
                                constants.flow_session_expiration_mode.end,
                            "permissions": mock.MagicMock()
                            })]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(session=self.fs2)
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_if_session_duration_shorter_than_minutes(self):
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_expiration_mode":
                                constants.flow_session_expiration_mode.end,
                            "if_session_duration_shorter_than_minutes": 59,
                            "permissions": mock.MagicMock()
                            })]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result()
        self.assertRuleEqual(self.fallback_rule, result)

    def test_not_passing_if_session_duration_shorter_than_minutes_anonymous(self):
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_session_duration_shorter_than_minutes": 59,
                            "permissions": mock.MagicMock()
                            })]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(session=self.fs3)
        self.assertRuleEqual(self.fallback_rule, result)

    def test_passed_session_duration_shorter_than_minutes(self):
        faked_permissions = frozenset([mock.MagicMock()])
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_session_duration_shorter_than_minutes": 59,
                            "permissions": faked_permissions})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(session=self.fs2)
        self.assertRuleEqual(
            result,
            {"message": None,
             "permissions": faked_permissions})

    def test_with_above_not_considiered(self):
        faked_permissions = frozenset([mock.MagicMock()])
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"permissions": faked_permissions})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(session=self.fs2)
        self.assertRuleEqual(
            result,
            {"message": None,
             "permissions": faked_permissions})

    def test_deal_with_deprecated_modify(self):
        faked_permission = mock.MagicMock()
        self.mock_get_flow_rules.return_value = [
            dict_to_struct(
                {"permissions": frozenset(["modify", faked_permission])})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(session=self.fs2)

        self.assertRuleEqual(
            result,
            {"message": None,
             "permissions": frozenset(
                 [fperm.submit_answer, fperm.end_session, faked_permission])})

    def test_deal_with_deprecated_see_answer(self):
        faked_permission = mock.MagicMock()
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({
                "permissions": frozenset([
                    "see_answer", faked_permission])})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result(session=self.fs2)
        self.assertRuleEqual(
            result,
            {"message": None,
             "permissions": frozenset(
                 [faked_permission, fperm.see_answer_after_submission])})

    def test_removing_access_permissions_for_non_in_progress_sessions(self):
        faked_permission = mock.MagicMock()
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({
                "permissions": frozenset([
                    "modify", faked_permission])})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result()
        self.assertRuleEqual(
            result,
            {"message": None,
             "permissions": frozenset(
                 [faked_permission])})

        self.mock_get_flow_rules.return_value = [
            dict_to_struct({
                "permissions": frozenset([
                    "end_session", faked_permission])})]
        self.mock_eval_generic_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = True
        result = self.get_result()
        self.assertRuleEqual(
            result,
            {"message": None,
             "permissions": frozenset(
                 [faked_permission])})


class GetSessionGradingRuleTest(GetSessionRuleMixin,
                                SingleCourseTestMixin, TestCase):
    # test utils.get_session_grading_rule
    call_func = utils.get_session_grading_rule
    rule_klass = utils.FlowSessionGradingRule

    no_g_rule_exception_msg = (
        "grading rule determination was unable to find a grading rule")

    fallback_rule = None
    default_rule = {"generates_grade": False}

    @property
    def default_kwargs(self):
        return {
            "session": self.fs1,
            "flow_desc": mock.MagicMock(),
            "now_datetime": self.now,
        }

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

        cls.now = now() - timedelta(days=1)

        start_time = cls.now - timedelta(minutes=60)

        cls.fs1 = factories.FlowSessionFactory(
            participation=cls.student_participation, in_progress=False,
            expiration_mode=constants.flow_session_expiration_mode.end,
            start_time=start_time, completion_time=cls.now
        )
        cls.fs2 = factories.FlowSessionFactory(
            participation=cls.ta_participation, in_progress=True,
            expiration_mode=constants.flow_session_expiration_mode.roll_over,
            start_time=start_time
        )
        cls.fs3 = factories.FlowSessionFactory(
            course=cls.course,
            participation=None, in_progress=True, user=None,
            expiration_mode=constants.flow_session_expiration_mode.roll_over,
            start_time=start_time, completion_time=cls.now
        )

    def get_result(self, **extra_kwargs):
        kwargs = self.get_updated_kwargs(**extra_kwargs)
        return utils.get_session_grading_rule(**kwargs)

    def get_default_rule(self, **kwargs):
        defaults = {
            "grade_identifier": "la_quiz",
            "grade_aggregation_strategy":
                constants.grade_aggregation_strategy.use_latest,
            "due": None,
            "generates_grade": True,
            "description": None,
            "credit_percent": 100,
            "use_last_activity_as_completion_time": False,
            "max_points": None,
            "max_points_enforced_cap": None,
            "bonus_points": 0
        }
        defaults.update(kwargs)
        return utils.FlowSessionGradingRule(**defaults)

    def test_no_rules(self):
        self.mock_get_flow_rules.return_value = []

        with self.assertRaises(RuntimeError) as cm:
            self.get_result()
        self.assertIn(self.no_g_rule_exception_msg, str(cm.exception))

        # make sure get_flow_rules is called with expected default_rules_desc
        self.assertEqual(self.mock_get_flow_rules.call_count, 1)
        self.assertIn("default_rules_desc", self.mock_get_flow_rules.call_args[1])

        default_rules_desc = (
            self.mock_get_flow_rules.call_args[1]["default_rules_desc"])
        self.assertFalse(default_rules_desc[0].generates_grade)

    def test_skip_if_has_role(self):
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_has_role": ["instructor", "ta"]})]

        with self.assertRaises(RuntimeError) as cm:
            self.get_result()
        self.assertIn(self.no_g_rule_exception_msg, str(cm.exception))

    def test_passed_if_has_role(self):
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_has_role": ["instructor", "ta"]}),
            dict_to_struct({"if_has_role": ["instructor", "ta", "student"]})]

        result = self.get_result(flow_desc=self.get_hacked_flow_desc())
        self.assertRuleEqual(result, self.get_default_rule())

    def test_not_passing_eval_generic_session_conditions(self):
        self.mock_get_flow_rules.return_value = [dict_to_struct({})]
        self.mock_eval_generic_session_conditions.return_value = False
        self.mock_eval_participation_tags_conditions.return_value = True
        with self.assertRaises(RuntimeError) as cm:
            self.get_result(flow_desc=self.get_hacked_flow_desc())
        self.assertIn(self.no_g_rule_exception_msg, str(cm.exception))

    def test_not_passing_eval_participation_tags_conditions(self):
        self.mock_get_flow_rules.return_value = [dict_to_struct({})]
        self.mock_eval_generic_session_conditions.return_value = True
        self.mock_eval_participation_tags_conditions.return_value = False
        with self.assertRaises(RuntimeError) as cm:
            self.get_result(flow_desc=self.get_hacked_flow_desc())
        self.assertIn(self.no_g_rule_exception_msg, str(cm.exception))

    def test_if_completed_before_skipped(self):
        self.mock_get_flow_rules.return_value = [dict_to_struct({
            "if_completed_before": "my_test_event 1"
        })]
        with self.assertRaises(RuntimeError) as cm:
            self.get_result(flow_desc=self.get_hacked_flow_desc())
        self.assertIn(self.no_g_rule_exception_msg, str(cm.exception))

    def test_if_completed_before_in_progress_session_skipped(self):
        self.mock_get_flow_rules.return_value = [dict_to_struct({
            "if_completed_before": "my_test_event 1"
        })]
        with self.assertRaises(RuntimeError) as cm:
            self.get_result(
                session=self.fs2,
                flow_desc=self.get_hacked_flow_desc())
        self.assertIn(self.no_g_rule_exception_msg, str(cm.exception))

    def test_if_completed_before_passed_not_using_last_activity(self):
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({"if_completed_before": "my_test_event 1"}),
            dict_to_struct({"if_completed_before": "my_test_event 2"}),
        ]
        result = self.get_result(flow_desc=self.get_hacked_flow_desc())
        self.assertRuleEqual(result, self.get_default_rule())

    def test_if_completed_before_using_last_activity_with_last_activity_none_skipped(self):  # noqa
        self.mock_get_flow_rules.return_value = [
            dict_to_struct(
                {"if_completed_before": "my_test_event 1",
                 "use_last_activity_as_completion_time": True
                 }),
        ]
        with self.assertRaises(RuntimeError) as cm:
            self.get_result(flow_desc=self.get_hacked_flow_desc())
        self.assertIn(self.no_g_rule_exception_msg, str(cm.exception))

    def test_if_completed_before_using_last_activity_exist_but_skipped(self):
        # create last_activity
        page_data = factories.FlowPageDataFactory(flow_session=self.fs1)
        factories.FlowPageVisitFactory(
            page_data=page_data,
            visit_time=my_test_event_1_time + timedelta(hours=1),
            answer={"answer": "hi"})

        self.mock_get_flow_rules.return_value = [
            dict_to_struct(
                {"if_completed_before": "my_test_event 1",
                 "use_last_activity_as_completion_time": True
                 }),
        ]
        with self.assertRaises(RuntimeError) as cm:
            self.get_result(flow_desc=self.get_hacked_flow_desc())
        self.assertIn(self.no_g_rule_exception_msg, str(cm.exception))

    def test_if_completed_before_using_last_activity_exist_passed(self):
        # create last_activity
        page_data = factories.FlowPageDataFactory(flow_session=self.fs1)
        factories.FlowPageVisitFactory(
            page_data=page_data,
            visit_time=my_test_event_1_time - timedelta(hours=1),
            answer={"answer": "hi"})

        self.mock_get_flow_rules.return_value = [
            dict_to_struct(
                {"if_completed_before": "my_test_event 1",
                 "use_last_activity_as_completion_time": True
                 }),
        ]
        result = self.get_result(flow_desc=self.get_hacked_flow_desc())
        self.assertRuleEqual(
            result, self.get_default_rule(
                use_last_activity_as_completion_time=True))

    def test_params_in_passed_to_result(self):
        # rule params
        mock_bonus_points = mock.MagicMock()
        mock_max_points = mock.MagicMock()
        mock_max_points_enforced_cap = mock.MagicMock()

        mock_generates_grade = mock.MagicMock()
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({
                "generates_grade": mock_generates_grade,
                "bonus_poinsts": mock_bonus_points,
                "max_points": mock_max_points,
                "max_points_enforced_cap": mock_max_points_enforced_cap,
                "due": "my_mock_event_time"
            })]

        # flow_desc_params
        mock_flow_desc_grade_identifier = mock.MagicMock()
        mock_flow_desc_grade_aggregation_strategy = mock.MagicMock()

        result = self.get_result(flow_desc=self.get_hacked_flow_desc(
            rules=dict_to_struct({
                "grade_identifier": mock_flow_desc_grade_identifier,
                "grade_aggregation_strategy":
                    mock_flow_desc_grade_aggregation_strategy})))

        self.assertRuleEqual(result, self.get_default_rule(
            max_points=mock_max_points,
            max_points_enforced_cap=mock_max_points_enforced_cap,
            generates_grade=mock_generates_grade,
            grade_identifier=mock_flow_desc_grade_identifier,
            grade_aggregation_strategy=mock_flow_desc_grade_aggregation_strategy,
            due=my_mock_event_time
        ))

    def test_no_flow_desc_rule(self):
        self.mock_get_flow_rules.return_value = [
            dict_to_struct({})]

        result = self.get_result(
            flow_desc=self.get_hacked_flow_desc(del_rules=True))
        self.assertRuleEqual(result, self.get_default_rule(
            grade_identifier=None,
            grade_aggregation_strategy=None,
            bonus_points=0,
            max_points=None,
            max_points_enforced_cap=None,
        ))


class CoursePageContextTest(SingleCourseTestMixin, MockAddMessageMixing, TestCase):
    # test utils.CoursePageContext (for cases not covered by other tests)

    def setUp(self):
        super().setUp()
        rf = RequestFactory()
        self.request = rf.get(self.get_course_page_url())

    def test_preview_commit_sha(self):
        # commit_sha of https://github.com/inducer/relate-sample/pull/11
        commit_sha = "ec41a2de73a99e6022060518cb5c5c162b88cdf5"
        self.ta_participation.preview_git_commit_sha = commit_sha
        self.ta_participation.save()
        self.request.user = self.ta_participation.user
        pctx = utils.CoursePageContext(self.request, self.course.identifier)

        self.assertEqual(
            pctx.course_commit_sha,
            commit_sha.encode())

        self.assertAddMessageCallCount(0)

    def test_invalid_preview_commit_sha(self):
        commit_sha = "invalid_commit_sha"
        self.ta_participation.preview_git_commit_sha = commit_sha
        self.ta_participation.save()
        self.request.user = self.ta_participation.user
        pctx = utils.CoursePageContext(self.request, self.course.identifier)

        self.assertEqual(
            pctx.course_commit_sha,
            self.course.active_git_commit_sha.encode())

        self.assertAddMessageCallCount(1)
        expected_error_msg = (
                "Preview revision '%s' does not exist--"
                "showing active course content instead." % commit_sha)

        self.assertAddMessageCalledWith(expected_error_msg)

    def test_role_identifiers(self):
        self.request.user = self.ta_participation.user
        pctx = utils.CoursePageContext(self.request, self.course.identifier)
        self.assertEqual(pctx.role_identifiers(), ["ta"])

        with mock.patch(
                "course.enrollment.get_participation_role_identifiers"
        ) as mock_get_prole_identifiers:
            self.assertEqual(pctx.role_identifiers(), ["ta"])

            # This is to ensure _role_identifiers_cache is working
            self.assertEqual(mock_get_prole_identifiers.call_count, 0)

    def test_error_when_nestedly_use_pctx_as_context_manager(self):
        self.request.user = self.ta_participation.user
        pctx = utils.CoursePageContext(self.request, self.course.identifier)

        with self.assertRaises(RuntimeError) as cm:
            with pctx:
                with pctx:
                    pass

        expected_error_msg = (
            "Nested use of 'course_view' as context manager "
            "is not allowed.")

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


class FlowContextTest(unittest.TestCase):
    # test utils.FlowContext (for cases not covered by other tests)
    def test_404(self):
        repo = mock.MagicMock()
        course = mock.MagicMock()
        flow_id = "some_id"
        participation = mock.MagicMock()

        with mock.patch(
                "course.utils.get_course_commit_sha"), mock.patch(
                "course.utils.get_flow_desc") as mock_get_flow_desc:
            from django.core.exceptions import ObjectDoesNotExist
            mock_get_flow_desc.side_effect = ObjectDoesNotExist

            from django import http
            with self.assertRaises(http.Http404):
                utils.FlowContext(repo, course, flow_id, participation)


class ParticipationPermissionWrapperTest(SingleCourseTestMixin, TestCase):
    # test utils.ParticipationPermissionWrapper (for cases not covered
    # by other tests)

    def setUp(self):
        super().setUp()
        rf = RequestFactory()
        request = rf.get(self.get_course_page_url())
        request.user = self.ta_participation.user
        self.pctx = utils.CoursePageContext(request, self.course.identifier)

    def test_get_invalid_permission(self):
        ppwraper = utils.ParticipationPermissionWrapper(self.pctx)
        invalid_perm = "invalid_perm"
        with self.assertRaises(ValueError) as cm:
            ppwraper[invalid_perm]

        expected_error_msg = (
            "permission name '%s' not valid" % invalid_perm)

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

    def test_not_iterale(self):
        ppwraper = utils.ParticipationPermissionWrapper(self.pctx)
        with self.assertRaises(TypeError) as cm:
            iter(ppwraper)

        expected_error_msg = (
            "ParticipationPermissionWrapper is not iterable.")

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


class WillUseMaskedProfileForEmailTest(SingleCourseTestMixin, TestCase):
    # test utils.will_use_masked_profile_for_email
    def test_no_recipient_email(self):
        self.assertFalse(utils.will_use_masked_profile_for_email(None))
        self.assertFalse(utils.will_use_masked_profile_for_email([]))

    def test_check_single_email(self):
        self.assertFalse(
            utils.will_use_masked_profile_for_email(
                "foo@bar.com"))
        self.assertFalse(
            utils.will_use_masked_profile_for_email(
                ["foo@bar.com"]))

    def test_any(self):
        from course.constants import participation_permission as pperm
        from course.models import ParticipationPermission

        pp = ParticipationPermission(
            participation=self.ta_participation,
            permission=pperm.view_participant_masked_profile)
        pp.save()
        self.assertTrue(
            utils.will_use_masked_profile_for_email(
                self.ta_participation.user.email))

        self.assertFalse(
            utils.will_use_masked_profile_for_email(
                self.instructor_participation.user.email))

        # any participation in the list have that permission, then True
        self.assertTrue(
            utils.will_use_masked_profile_for_email(
                [self.ta_participation.user.email,
                 self.instructor_participation.user.email]))


class GetFacilitiesConfigTest(unittest.TestCase):
    # utils.get_facilities_config (for cases not covered by other tests)
    def test_none(self):
        with override_settings():
            del settings.RELATE_FACILITIES
            self.assertIsNone(utils.get_facilities_config())

        with override_settings(RELATE_FACILITIES=None):
            self.assertIsNone(utils.get_facilities_config())


class FlowPageContextTest(SingleCourseQuizPageTestMixin, TestCase):

    flow_id = QUIZ_FLOW_ID

    def test_flow_page_context_uri_without_requests(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            self.start_flow(self.flow_id)
            from course.models import FlowSession
            flow_session = FlowSession.objects.first()

            from course.utils import FlowPageContext
            with self.get_course_page_context(
                    self.student_participation.user) as pctx:
                fpctx = FlowPageContext(
                    repo=pctx.repo,
                    course=self.course,
                    flow_id=self.flow_id,
                    page_ordinal=1,
                    participation=self.student_participation,
                    flow_session=flow_session
                )
        self.assertIsNone(fpctx.page_context.page_uri)

# vim: foldmethod=marker