test_flow.py

        self.mock_will_receive_feedback.return_value = True
        flow.add_buttons_to_form(form, self.fpctx, self.flow_session, frozenset())

        names, values = self.get_form_submit_inputs(form)
        self.assertNotIn("save", names)
        self.assertIn("submit", names)
        self.assertIn("Submit final answer", values)

    def test_add_save_button_by_default(self):
        class MyForm(StyledForm):
            pass

        form = MyForm()

        self.mock_will_receive_feedback.return_value = True
        flow.add_buttons_to_form(form, self.fpctx, self.flow_session, frozenset())

        names, values = self.get_form_submit_inputs(form)
        self.assertIn("save", names)
        self.assertIn("submit", names)
        self.assertIn("Submit final answer", values)

    def test_add_submit_answer_for_feedback_button(self):
        form = StyledForm()

        self.mock_will_receive_feedback.return_value = True
        flow.add_buttons_to_form(
            form, self.fpctx, self.flow_session, frozenset([fperm.change_answer]))

        names, values = self.get_form_submit_inputs(form)
        self.assertIn("submit", names)
        self.assertIn("Submit answer for feedback", values)

    def test_not_add_submit_answer_for_feedback_button(self):

        self.mock_will_receive_feedback.return_value = True
        combinations = [(frozenset([fp]), False) for fp in
                        get_flow_permissions_list(excluded=fperm.change_answer)]
        combinations.append(([], False))

        form = StyledForm()
        for permissions, _show in combinations:
            with self.subTest(
                    permissions=permissions):
                flow.add_buttons_to_form(
                    form, self.fpctx, self.flow_session, permissions)

            names, values = self.get_form_submit_inputs(form)
            self.assertNotIn("Submit answer for feedback", values)

    def test_add_save_and_next(self):

        self.mock_will_receive_feedback.return_value = False

        combinations = [(frozenset([fp]), True) for fp in
                        get_flow_permissions_list()]
        combinations.append(([], True))

        self.fake_flow_session_page_count(3)
        self.fake_fpctx_page_data_page_ordinal(1)

        form = StyledForm()
        for permissions, _show in combinations:
            with self.subTest(
                    permissions=permissions):
                flow.add_buttons_to_form(
                    form, self.fpctx, self.flow_session, permissions)

            names, values = self.get_form_submit_inputs(form)
            self.assertIn("save_and_next", names)
            self.assertNotIn("submit", names)
            self.assertNotIn("save_and_finish", names)

    def test_add_save_and_finish(self):

        self.mock_will_receive_feedback.return_value = False

        combinations = [(frozenset([fp]), True) for fp in
                        get_flow_permissions_list()]
        combinations.append(([], True))

        self.fake_flow_session_page_count(3)
        self.fake_fpctx_page_data_page_ordinal(2)

        form = StyledForm()
        for permissions, _show in combinations:
            with self.subTest(
                    permissions=permissions):
                flow.add_buttons_to_form(
                    form, self.fpctx, self.flow_session, permissions)

            names, values = self.get_form_submit_inputs(form)
            self.assertNotIn("save_and_next", names)
            self.assertNotIn("submit", names)
            self.assertIn("save_and_finish", names)


class CreateFlowPageVisitTest(SingleCourseTestMixin, TestCase):
    # test flow.create_flow_page_visit
    def setUp(self):
        super().setUp()
        rf = RequestFactory()
        self.request = rf.get(self.get_course_page_url())

    def test_anonymous_visit(self):
        self.request.user = AnonymousUser()
        fs = factories.FlowSessionFactory(
            course=self.course, participation=None, user=None)
        page_data = factories.FlowPageDataFactory(flow_session=fs)

        flow.create_flow_page_visit(self.request, fs, page_data)

        fpvs = models.FlowPageVisit.objects.all()
        self.assertEqual(fpvs.count(), 1)
        fpv = fpvs[0]
        self.assertEqual(fpv.user, None)
        self.assertEqual(fpv.is_submitted_answer, None)
        self.assertIsNotNone(fpv.remote_address)

    def test_impersonate(self):
        fs = factories.FlowSessionFactory(
            course=self.course, participation=self.student_participation)
        page_data = factories.FlowPageDataFactory(flow_session=fs)

        def dummy_get_response(*args):
            raise AssertionError("never called")

        self.request.user = self.student_participation.user
        middleware = SessionMiddleware(dummy_get_response)

        middleware.process_request(self.request)
        self.request.session.save()

        self.request.relate_impersonate_original_user = \
                self.instructor_participation.user

        flow.create_flow_page_visit(self.request, fs, page_data)

        fpvs = models.FlowPageVisit.objects.all()
        self.assertEqual(fpvs.count(), 1)
        fpv = fpvs[0]
        self.assertEqual(fpv.user, self.student_participation.user)
        self.assertEqual(fpv.is_submitted_answer, None)
        self.assertIsNotNone(fpv.remote_address)

        self.assertEqual(fpv.impersonated_by, self.instructor_participation.user)


class ViewFlowPageTest(SingleCourseQuizPageTestMixin, HackRepoMixin, TestCase):
    # test flow.view_flow_page for not covered part by other tests

    def setUp(self):
        super().setUp()

        fake_lock_down_if_needed = mock.patch(
            "course.flow.lock_down_if_needed")
        self.mock_lock_down_if_needed = fake_lock_down_if_needed.start()
        self.addCleanup(fake_lock_down_if_needed.stop)

        fake_create_flow_page_visit = mock.patch(
            "course.flow.create_flow_page_visit")
        self.mock_create_flow_page_visit = fake_create_flow_page_visit.start()
        self.addCleanup(fake_create_flow_page_visit.stop)

    def test_invalid_flow_session_id(self):
        with mock.patch(
                "course.flow.adjust_flow_session_page_data") as mock_adjust_data:
            kwargs = {
                "course_identifier": self.course.identifier,
                "flow_session_id": 100,  # invalid
                "page_ordinal": 0
            }
            resp = self.client.get(
                reverse("relate-view_flow_page", kwargs=kwargs))
            self.assertEqual(resp.status_code, 404)

            # check should happen before adjust session data
            self.assertEqual(mock_adjust_data.call_count, 0)

            self.assertEqual(self.mock_create_flow_page_visit.call_count, 0)

    def test_fpctx_page_is_none(self):
        self.start_flow(self.flow_id)
        with mock.patch("course.content.get_flow_page_desc") as mock_get_page_desc:
            from django.core.exceptions import ObjectDoesNotExist
            mock_get_page_desc.side_effect = ObjectDoesNotExist

            resp = self.client.get(self.get_page_url_by_ordinal(0))
            self.assertEqual(resp.status_code, 404)

            self.assertEqual(self.mock_create_flow_page_visit.call_count, 0)

    default_session_access_rule = {"permissions": []}

    def test_may_not_view(self):
        self.start_flow(self.flow_id)
        with mock.patch(
                "course.page.base.PageBase.get_modified_permissions_for_page"
        ) as mock_perms:
            mock_perms.return_value = []
            resp = self.client.get(self.get_page_url_by_ordinal(0))
            self.assertEqual(resp.status_code, 403)
            self.assertEqual(mock_perms.call_count, 1)

            self.assertTrue(self.mock_lock_down_if_needed.call_count > 0)
            self.assertEqual(self.mock_create_flow_page_visit.call_count, 0)

    def test_post_finish(self):
        self.start_flow(self.flow_id)
        resp = self.client.post(
            self.get_page_url_by_ordinal(0), data={"finish": ""})
        self.assertRedirects(resp, self.get_finish_flow_session_view_url(),
                             fetch_redirect_response=False)
        self.assertEqual(self.mock_create_flow_page_visit.call_count, 0)

    def test_post_result_not_tuple(self):
        self.start_flow(self.flow_id)
        with mock.patch("course.flow.post_flow_page") as mock_post_flow_page:
            mock_post_flow_page.return_value = http.HttpResponse("hello world")
            resp = self.post_answer_by_page_id(
                "half", answer_data={"answer": "ok"})
            self.assertEqual(resp.status_code, 200)

    def test_prev_visit_id_after_post(self):
        self.start_flow(self.flow_id)
        resp = self.post_answer_by_page_id(
            "half", answer_data={"answer": "ok"})

        fpvs = models.FlowPageVisit.objects.all()
        self.assertEqual(fpvs.count(), 1)
        fpv = fpvs[0]

        self.assertResponseContextEqual(resp, "prev_visit_id", fpv.id)

        resp = self.post_answer_by_page_id(
            "half", answer_data={"answer": "1/2"})
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "prev_visit_id", fpv.id)

    def test_get_prev_visit_id_not_number(self):
        self.start_flow(self.flow_id)
        resp = self.client.get(self.get_page_url_by_ordinal(1, visit_id="foo"))
        self.assertEqual(resp.status_code, 400)

    def test_get_prev_visit_id_not_exists(self):
        # no prev_answer_visits
        self.start_flow(self.flow_id)
        resp = self.client.get(self.get_page_url_by_ordinal(1, visit_id=100))
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "viewing_prior_version", False)

    def test_get_prev_visit_id_not_exists_with_prev_answer_visit(self):
        self.start_flow(self.flow_id)
        page_id = "half"
        page_data = models.FlowPageData.objects.get(page_id=page_id)
        factories.FlowPageVisitFactory(
            page_data=page_data, answer={"answer": "hi"})

        resp = self.client.get(self.get_page_url_by_page_id(page_id, visit_id=100))
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "viewing_prior_version", False)

    def switch_to_fake_commit_sha(
            self, commit_sha="my_fake_commit_sha_for_view_flow_page"):
        self.course.active_git_commit_sha = commit_sha
        self.course.save()

    def test_get_prev_visit_id_exists(self):
        self.switch_to_fake_commit_sha()

        self.start_flow(self.flow_id)
        page_id = "half"
        page_data = models.FlowPageData.objects.get(page_id=page_id)

        visit_time = now() - timedelta(days=1)

        visit0 = factories.FlowPageVisitFactory(
            page_data=page_data, answer=None, visit_time=visit_time)

        visit_time = visit_time + timedelta(hours=1)

        visit1 = factories.FlowPageVisitFactory(
            page_data=page_data, answer={"answer": "hi"}, visit_time=visit_time)

        visit_time = visit_time + timedelta(hours=1)

        visit2 = factories.FlowPageVisitFactory(
            page_data=page_data, answer={"answer": "hello"}, visit_time=visit_time)

        with mock.patch("course.flow.messages.add_message") as mock_add_msg:
            # viewing current visit
            resp = self.client.get(
                self.get_page_url_by_page_id(page_id, visit_id=visit2.id))
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextEqual(resp, "viewing_prior_version", False)
            self.assertEqual(mock_add_msg.call_count, 0, mock_add_msg.call_args)
            mock_add_msg.reset_mock()

            # viewing non-answer visit
            resp = self.client.get(
                self.get_page_url_by_page_id(page_id, visit_id=visit0.id))
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextEqual(resp, "viewing_prior_version", False)
            self.assertEqual(mock_add_msg.call_count, 0, mock_add_msg.call_args)
            mock_add_msg.reset_mock()

            # viewing previous visit
            resp = self.client.get(
                self.get_page_url_by_page_id(page_id, visit_id=visit1.id))
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextEqual(resp, "viewing_prior_version", True)
            self.assertEqual(mock_add_msg.call_count, 1)
            expected_warn_message_partial = "Viewing prior submission dated"
            self.assertIn(
                expected_warn_message_partial, mock_add_msg.call_args[0][2])

    def test_see_session_time_not_in_progress(self):
        completion_time = now() - timedelta(days=1)
        start_time = completion_time - timedelta(hours=1)
        self.student_participation.time_factor = 1.11
        self.student_participation.save()

        fs = factories.FlowSessionFactory(
            course=self.course,
            participation=self.student_participation,
            start_time=start_time,
            completion_time=completion_time, in_progress=False)

        resp = self.client.get(
                self.get_page_url_by_ordinal(0, flow_session_id=fs.id))
        self.assertResponseContextEqual(resp, "session_minutes", None)
        self.assertResponseContextEqual(resp, "time_factor", 1)

        self.switch_to_fake_commit_sha()
        resp = self.client.get(
                self.get_page_url_by_ordinal(0, flow_session_id=fs.id))
        self.assertResponseContextEqual(resp, "session_minutes", 60)
        self.assertResponseContextEqual(resp, "time_factor", 1.11)

    def test_see_session_time_in_progress(self):
        self.switch_to_fake_commit_sha()

        start_time = now() - timedelta(minutes=62)

        fs = factories.FlowSessionFactory(
            course=self.course,
            participation=self.student_participation,
            start_time=start_time,
            in_progress=True)

        resp = self.client.get(
                self.get_page_url_by_ordinal(0, flow_session_id=fs.id))
        self.assertTrue(resp.context["session_minutes"] > 61)
        self.assertResponseContextEqual(resp, "time_factor", 1)

    def test_warning_for_anonymous_flow_session(self):
        # switch to a fake flow which allow anonymous to start a flow
        self.switch_to_fake_commit_sha()
        self.client.logout()
        self.start_flow(self.flow_id)

        with mock.patch("course.flow.messages.add_message") as mock_add_msg:
            resp = self.client.get(self.get_page_url_by_ordinal(1))
            expected_warn_msg = (
                "Changes to this session are being prevented "
                "because this session yields a permanent grade, but "
                "you have not completed your enrollment process in "
                "this course.")
            self.assertIn(expected_warn_msg, mock_add_msg.call_args[0])
            self.assertResponseContextIsNotNone(resp, "session_minutes")
            self.assertResponseContextEqual(resp, "time_factor", 1)

    def test_viewing_optional_page(self):
        self.switch_to_fake_commit_sha()
        self.start_flow(self.flow_id)

        resp = self.client.get(self.get_page_url_by_page_id("half"))
        self.assertResponseHasNoContext(resp, "is_optional_page")

        resp = self.client.get(self.get_page_url_by_page_id("half2"))
        self.assertResponseContextEqual(resp, "is_optional_page", True)


class GetPressedButtonTest(unittest.TestCase):
    def test_success(self):
        buttons = ["save", "save_and_next", "save_and_finish", "submit"]
        for button in buttons:
            with self.subTest(button=button):
                form = StyledForm(data={button: ""})
                self.assertEqual(flow.get_pressed_button(form), button)

    def test_failure(self):
        form = StyledForm(data={"unknown": ""})
        from django.core.exceptions import SuspiciousOperation
        with self.assertRaises(SuspiciousOperation) as cm:
            flow.get_pressed_button(form)

        expected_error_msg = "could not find which button was pressed"
        self.assertIn(expected_error_msg, str(cm.exception))


class PostFlowPageTest(HackRepoMixin, SingleCourseQuizPageTestMixin, TestCase):
    # test flow.post_flow_page for not covered part by other tests

    initial_commit_sha = "my_fake_commit_sha_for_view_flow_page"
    page_id = "half"

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

        # We only concern one page, so it can be put here to speed up
        client = Client()
        client.force_login(cls.student_participation.user)
        cls.start_flow(client, cls.flow_id)

        # Because they change between test, we need to refer to them
        # to do refresh_from_db when setUp.
        cls.flow_session = models.FlowSession.objects.first()
        cls.page_data = models.FlowPageData.objects.get(page_id=cls.page_id)

    def setUp(self):
        super().setUp()
        self.flow_session.refresh_from_db()
        self.page_data.refresh_from_db()
        self.fpctx = mock.MagicMock()

        rf = RequestFactory()
        self.request = rf.get(self.get_course_page_url())
        self.request.user = self.student_participation.user

        def dummy_get_response(*args):
            raise AssertionError("never called")

        middleware = SessionMiddleware(dummy_get_response)
        middleware.process_request(self.request)
        self.request.session.save()

        self.fpctx.course = self.course
        self.fpctx.page_data = self.page_data
        self.fpctx.page.answer_data.return_value = {"answer": "hello"}
        self.fpctx.page_ordinal = self.page_data.page_ordinal

        from course.page.base import AnswerFeedback
        self.fpctx.page.grade.return_value = AnswerFeedback(correctness=1)

        fake_create_flow_page_visit = mock.patch(
            "course.flow.create_flow_page_visit")
        self.mock_create_flow_page_visit = fake_create_flow_page_visit.start()
        self.addCleanup(fake_create_flow_page_visit.stop)

        fake_get_prev_answer_visits_qset = mock.patch(
            "course.flow.get_prev_answer_visits_qset")
        self.mock_get_prev_answer_visits_qset = (
            fake_get_prev_answer_visits_qset.start())
        self.addCleanup(fake_get_prev_answer_visits_qset.stop)

        fake_get_pressed_button = mock.patch(
            "course.flow.get_pressed_button")
        self.mock_get_pressed_button = fake_get_pressed_button.start()
        self.addCleanup(fake_get_pressed_button.stop)

        fake_add_message = mock.patch("course.flow.messages.add_message")
        self.mock_add_message = fake_add_message.start()
        self.addCleanup(fake_add_message.stop)

        fake_will_receive_feedback = mock.patch(
            "course.flow.will_receive_feedback")
        self.mock_will_receive_feedback = fake_will_receive_feedback.start()
        self.addCleanup(fake_will_receive_feedback.stop)

    def test_no_submit_answer_fperm(self):
        flow.post_flow_page(
            self.flow_session, self.fpctx, self.request,
            permissions=frozenset(), generates_grade=True)

        self.assertEqual(
            self.mock_create_flow_page_visit.call_count, 1)
        self.assertEqual(self.mock_add_message.call_count, 2)
        msgs = ["Answer submission not allowed.",
                "Failed to submit answer."]

        used = []
        for calls in self.mock_add_message.call_args_list:
            args, _ = calls
            for msg in msgs:
                if msg in args:
                    used.append(msg)

        for msg in msgs:
            if msg not in used:
                self.fail("%s is unexpectedly not used in adding message")

    def test_impersonated(self):
        self.request.relate_impersonate_original_user = \
                self.instructor_participation.user
        self.mock_get_pressed_button.return_value = "save"
        flow.post_flow_page(
            self.flow_session, self.fpctx, self.request,
            permissions=frozenset([fperm.submit_answer, fperm.change_answer]),
            generates_grade=True)

        self.assertEqual(self.mock_add_message.call_count, 1)
        self.assertIn("Answer saved.",
                      self.mock_add_message.call_args[0])

        self.assertEqual(
            self.mock_create_flow_page_visit.call_count, 0)

        latest_answer_visit = models.FlowPageVisit.objects.last()
        self.assertEqual(latest_answer_visit.impersonated_by,
                         self.instructor_participation.user)

    def test_save_and_next(self):
        self.mock_get_pressed_button.return_value = "save_and_next"
        self.mock_will_receive_feedback.return_value = False
        resp = flow.post_flow_page(
            self.flow_session, self.fpctx, self.request,
            permissions=frozenset([fperm.submit_answer, fperm.change_answer]),
            generates_grade=True)

        self.assertIsInstance(resp, http.HttpResponse)

        next_page_ordinal = self.page_data.page_ordinal + 1

        self.assertRedirects(
            resp, self.get_page_url_by_ordinal(next_page_ordinal),
            fetch_redirect_response=False)

        self.assertEqual(self.mock_add_message.call_count, 1)
        self.assertIn("Answer saved.",
                      self.mock_add_message.call_args[0])

        self.assertEqual(
            self.mock_create_flow_page_visit.call_count, 0)

        latest_answer_visit = models.FlowPageVisit.objects.last()
        self.assertIsNone(latest_answer_visit.impersonated_by)

    def test_save_and_finish(self):
        self.mock_get_pressed_button.return_value = "save_and_finish"
        self.mock_will_receive_feedback.return_value = False
        resp = flow.post_flow_page(
            self.flow_session, self.fpctx, self.request,
            permissions=frozenset([fperm.submit_answer, fperm.change_answer]),
            generates_grade=True)

        self.assertRedirects(
            resp, self.get_finish_flow_session_view_url(),
            fetch_redirect_response=False)

        self.assertIsInstance(resp, http.HttpResponse)
        self.assertEqual(resp.status_code, 302)

        self.assertEqual(self.mock_add_message.call_count, 1)
        self.assertIn("Answer saved.",
                      self.mock_add_message.call_args[0])

        self.assertEqual(
            self.mock_create_flow_page_visit.call_count, 0)


class SendEmailAboutFlowPageTest(HackRepoMixin,
                                 SingleCourseQuizPageTestMixin, TestCase):
    # test flow.send_email_about_flow_page
    initial_commit_sha = "my_fake_commit_sha_for_view_flow_page"
    page_id = "half"

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

        # We only concern one page, so it can be put here to speed up
        client = Client()
        client.force_login(cls.student_participation.user)
        cls.start_flow(client, cls.flow_id)

        # Because they change between test, we need to refer to them
        # to do refresh_from_db when setUp.
        cls.flow_session = models.FlowSession.objects.first()
        cls.page_data = models.FlowPageData.objects.get(page_id=cls.page_id)

    def setUp(self):
        super().setUp()

        fake_get_session_access_rule = mock.patch(
            "course.flow.get_session_access_rule")
        self.mock_get_session_access_rule = fake_get_session_access_rule.start()
        self.addCleanup(fake_get_session_access_rule.stop)

        fake_adjust_flow_session_page_data = mock.patch(
            "course.flow.adjust_flow_session_page_data")
        self.mock_adjust_flow_session_page_data = (
            fake_adjust_flow_session_page_data.start())
        self.mock_adjust_flow_session_page_data.return_value = None
        self.addCleanup(fake_adjust_flow_session_page_data.stop)

        fake_get_login_exam_ticket = mock.patch("course.flow.get_login_exam_ticket")
        self.mock_get_login_exam_ticket = fake_get_login_exam_ticket.start()
        self.addCleanup(fake_get_login_exam_ticket.stop)

        fake_get_modified_permissions_for_page = mock.patch(
            "course.page.base.PageBase.get_modified_permissions_for_page")
        self.mock_get_modified_permissions_for_page = (
            fake_get_modified_permissions_for_page.start()
        )
        self.mock_get_modified_permissions_for_page.return_value = [
            fperm.view, fperm.send_email_about_flow_page]
        self.addCleanup(fake_get_modified_permissions_for_page.stop)

        fake_get_and_check_flow_session = mock.patch(
            "course.flow.get_and_check_flow_session")
        self.mock_get_and_check_flow_session = (
            fake_get_and_check_flow_session.start())
        self.mock_get_and_check_flow_session.return_value = self.flow_session
        self.addCleanup(fake_get_and_check_flow_session.stop)

        fake_add_message = mock.patch("course.flow.messages.add_message")
        self.mock_add_message = fake_add_message.start()
        self.addCleanup(fake_add_message.stop)

        fake_will_use_masked_profile_for_email = mock.patch(
            "course.utils.will_use_masked_profile_for_email")
        self.mock_will_use_masked_profile_for_email = (
            fake_will_use_masked_profile_for_email.start())
        self.mock_will_use_masked_profile_for_email.return_value = False
        self.addCleanup(fake_will_use_masked_profile_for_email.stop)

    def get_send_email_about_flow_page_url(
            self, page_ordinal, course_identifier=None,
            flow_session_id=None):
        course_identifier = course_identifier or self.get_default_course_identifier()
        flow_session_id = (flow_session_id
                           or self.get_default_flow_session_id(course_identifier))
        return reverse(
            "relate-flow_page_interaction_email",
            kwargs={"course_identifier": course_identifier,
                    "flow_session_id": flow_session_id,
                    "page_ordinal": page_ordinal})

    def test_404(self):
        with mock.patch("course.content.get_flow_page_desc") as mock_get_page_desc:
            from django.core.exceptions import ObjectDoesNotExist
            mock_get_page_desc.side_effect = ObjectDoesNotExist
            resp = self.client.get(
                self.get_send_email_about_flow_page_url(page_ordinal=1))
            self.assertEqual(resp.status_code, 404)

            self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 1)
            self.assertEqual(self.mock_get_login_exam_ticket.call_count, 0)

    def test_no_permission_404(self):
        self.mock_get_modified_permissions_for_page.return_value = [fperm.view]

        resp = self.client.get(
            self.get_send_email_about_flow_page_url(page_ordinal=1))
        self.assertEqual(resp.status_code, 404)

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 1)
        self.assertEqual(self.mock_get_login_exam_ticket.call_count, 1)
        self.assertEqual(self.mock_get_session_access_rule.call_count, 1)
        self.assertEqual(self.mock_get_modified_permissions_for_page.call_count, 1)

    def test_may_send_email_about_flow_page_called(self):
        with mock.patch(
                "course.flow.may_send_email_about_flow_page") as mock_check:
            mock_check.return_value = False

            permissions = mock.MagicMock()
            self.mock_get_modified_permissions_for_page.return_value = permissions

            resp = self.client.get(
                self.get_send_email_about_flow_page_url(page_ordinal=1))
            self.assertEqual(resp.status_code, 404)
            self.assertEqual(mock_check.call_count, 1)
            self.assertIn(permissions, mock_check.call_args[0])

            mock_check.reset_mock()

            mock_check.return_value = True

            resp = self.client.get(
                self.get_send_email_about_flow_page_url(page_ordinal=1))
            self.assertEqual(resp.status_code, 200)
            self.assertEqual(mock_check.call_count, 1)
            self.assertIn(permissions, mock_check.call_args[0])

    def test_get(self):
        resp = self.client.get(
            self.get_send_email_about_flow_page_url(page_ordinal=1))
        self.assertEqual(resp.status_code, 200)

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 1)
        self.assertEqual(self.mock_get_login_exam_ticket.call_count, 1)
        self.assertEqual(self.mock_get_session_access_rule.call_count, 1)
        self.assertEqual(self.mock_get_modified_permissions_for_page.call_count, 1)

    def test_post(self):
        resp = self.client.post(
            self.get_send_email_about_flow_page_url(page_ordinal=1),
            data={"message": "foo bar" * 10}
        )
        self.assertRedirects(resp, self.get_page_url_by_ordinal(1),
                             fetch_redirect_response=False)

        self.assertEqual(self.mock_add_message.call_count, 1)
        expected_msg = ("Email sent, and notice that you will "
                        "also receive a copy of the email.")
        self.assertIn(expected_msg, self.mock_add_message.call_args[0])

        self.assertEqual(len(mail.outbox), 1)
        self.assertEqual(
            mail.outbox[0].recipients(),
            [self.ta_participation.user.email,
             self.student_participation.user.email])
        self.assertEqual(
            mail.outbox[0].reply_to, [self.student_participation.user.email])
        self.assertEqual(
            mail.outbox[0].bcc, [self.student_participation.user.email])

    def test_post_too_few_words(self):
        resp = self.client.post(
            self.get_send_email_about_flow_page_url(page_ordinal=1),
            data={"message": "foo bar"}
        )
        self.assertEqual(resp.status_code, 200)
        self.assertFormErrorLoose(
            resp, "At least 20 characters are required for submission.")
        self.assertEqual(len(mail.outbox), 0)

    def test_no_tas(self):
        self.ta_participation.status = constants.participation_status.dropped
        self.ta_participation.save()

        resp = self.client.post(
            self.get_send_email_about_flow_page_url(page_ordinal=1),
            data={"message": "foo bar" * 10}
        )
        self.assertRedirects(resp, self.get_page_url_by_ordinal(1),
                             fetch_redirect_response=False)

        self.assertEqual(self.mock_add_message.call_count, 1)
        expected_msg = ("Email sent, and notice that you will "
                        "also receive a copy of the email.")
        self.assertIn(expected_msg, self.mock_add_message.call_args[0])

        self.assertEqual(len(mail.outbox), 1)
        self.assertEqual(
            mail.outbox[0].recipients(),
            [self.instructor_participation.user.email,
             self.student_participation.user.email])
        self.assertEqual(
            mail.outbox[0].reply_to, [self.student_participation.user.email])
        self.assertEqual(
            mail.outbox[0].bcc, [self.student_participation.user.email])

        self.assertIn(
            self.student_participation.user.get_email_appellation(),
            mail.outbox[0].body)
        self.assertNotIn(
            self.student_participation.user.get_masked_profile(),
            mail.outbox[0].body)
        self.assertNotIn(
            "Dear user",
            mail.outbox[0].body)

    def test_mask_student_profile(self):
        self.mock_will_use_masked_profile_for_email.return_value = True

        resp = self.client.post(
            self.get_send_email_about_flow_page_url(page_ordinal=1),
            data={"message": "foo bar" * 10}
        )
        self.assertRedirects(resp, self.get_page_url_by_ordinal(1),
                             fetch_redirect_response=False)

        self.assertEqual(self.mock_add_message.call_count, 1)
        expected_msg = ("Email sent, and notice that you will "
                        "also receive a copy of the email.")
        self.assertIn(expected_msg, self.mock_add_message.call_args[0])

        self.assertEqual(len(mail.outbox), 1)

        self.assertNotIn(
            self.student_participation.user.get_email_appellation(),
            mail.outbox[0].body)
        self.assertIn(
            self.student_participation.user.get_masked_profile(),
            mail.outbox[0].body)


class UpdatePageBookmarkStateTest(SingleCourseQuizPageTestMixin, TestCase):
    # test flow.update_page_bookmark_state

    def setUp(self):
        super().setUp()
        self.start_flow(self.flow_id)
        self.flow_session = models.FlowSession.objects.last()

    def get_update_page_bookmark_state_url(
            self, page_ordinal, course_identifier=None, flow_session_id=None):
        course_identifier = course_identifier or self.get_default_course_identifier()
        flow_session_id = (flow_session_id
                           or self.get_default_flow_session_id(course_identifier))
        return reverse(
            "relate-update_page_bookmark_state",
            kwargs={"course_identifier": course_identifier,
                    "flow_session_id": flow_session_id,
                    "page_ordinal": page_ordinal})

    def test_not_post(self):
        resp = self.client.get(self.get_update_page_bookmark_state_url(1))
        self.assertEqual(resp.status_code, 400)
        fpd = models.FlowPageData.objects.get(
            flow_session=self.flow_session, page_ordinal=1)
        self.assertEqual(fpd.bookmarked, False)

    def test_invalid_flow_session_id(self):
        resp = self.client.post(
            self.get_update_page_bookmark_state_url(1, flow_session_id=100),
            data={"bookmark_state": "1"})
        self.assertEqual(resp.status_code, 404)
        fpd = models.FlowPageData.objects.get(
            flow_session=self.flow_session, page_ordinal=1)
        self.assertEqual(fpd.bookmarked, False)

    def test_success(self):
        resp = self.client.post(
            self.get_update_page_bookmark_state_url(1),
            data={"bookmark_state": "1"})
        self.assertEqual(resp.status_code, 200)
        fpd = models.FlowPageData.objects.get(
            flow_session=self.flow_session, page_ordinal=1)
        self.assertEqual(fpd.bookmarked, True)

    def test_post_invalid_bookmark_state(self):
        resp = self.client.post(
            self.get_update_page_bookmark_state_url(1),
            data={"bookmark_state": "a"})
        self.assertEqual(resp.status_code, 400)
        fpd = models.FlowPageData.objects.get(
            flow_session=self.flow_session, page_ordinal=1)
        self.assertEqual(fpd.bookmarked, False)

    def test_not_you_session(self):
        with self.temporarily_switch_to_user(self.ta_participation.user):
            resp = self.client.post(
                self.get_update_page_bookmark_state_url(1),
                data={"bookmark_state": "1"})
            self.assertEqual(resp.status_code, 403)
            fpd = models.FlowPageData.objects.get(
                flow_session=self.flow_session, page_ordinal=1)
            self.assertEqual(fpd.bookmarked, False)


class UpdateExpirationModeTest(SingleCourseQuizPageTestMixin, TestCase):
    # test flow.update_expiration_mode

    def setUp(self):
        super().setUp()
        self.start_flow(self.flow_id)
        self.flow_session = models.FlowSession.objects.last()

        fake_get_login_exam_ticket = mock.patch("course.flow.get_login_exam_ticket")
        self.mock_get_login_exam_ticket = fake_get_login_exam_ticket.start()
        self.addCleanup(fake_get_login_exam_ticket.stop)

        fake_is_expiration_mode_allowed = mock.patch(
            "course.flow.is_expiration_mode_allowed")
        self.mock_is_expiration_mode_allowed = (
            fake_is_expiration_mode_allowed.start())
        self.mock_is_expiration_mode_allowed.return_value = True
        self.addCleanup(fake_is_expiration_mode_allowed.stop)

    def get_relate_update_expiration_mode_url(
            self, course_identifier=None, flow_session_id=None):
        course_identifier = course_identifier or self.get_default_course_identifier()
        flow_session_id = (flow_session_id
                           or self.get_default_flow_session_id(course_identifier))
        return reverse(
            "relate-update_expiration_mode",
            kwargs={"course_identifier": course_identifier,
                    "flow_session_id": flow_session_id})

    def test_not_post(self):
        resp = self.client.get(self.get_relate_update_expiration_mode_url())
        self.assertEqual(resp.status_code, 400)
        fs = models.FlowSession.objects.last()
        self.assertEqual(fs.expiration_mode,
                         constants.flow_session_expiration_mode.end)

    def test_invalid_flow_session_id(self):
        resp = self.client.post(
            self.get_relate_update_expiration_mode_url(flow_session_id=100),
            data={"expiration_mode":
                      constants.flow_session_expiration_mode.roll_over})
        self.assertEqual(resp.status_code, 404)
        fs = models.FlowSession.objects.last()
        self.assertEqual(fs.expiration_mode,
                         constants.flow_session_expiration_mode.end)

    def test_session_not_in_progress(self):
        self.flow_session.in_progress = False
        self.flow_session.save()
        resp = self.client.post(
            self.get_relate_update_expiration_mode_url(),
            data={"expiration_mode":
                      constants.flow_session_expiration_mode.roll_over})
        self.assertEqual(resp.status_code, 403)
        fs = models.FlowSession.objects.last()
        self.assertEqual(fs.expiration_mode,
                         constants.flow_session_expiration_mode.end)

    def test_success(self):
        resp = self.client.post(
            self.get_relate_update_expiration_mode_url(),
            data={"expiration_mode":
                      constants.flow_session_expiration_mode.roll_over})
        self.assertEqual(resp.status_code, 200)
        fs = models.FlowSession.objects.last()
        self.assertEqual(fs.expiration_mode,
                         constants.flow_session_expiration_mode.roll_over)

    def test_not_is_expiration_mode_allowed(self):
        self.mock_is_expiration_mode_allowed.return_value = False
        resp = self.client.post(
            self.get_relate_update_expiration_mode_url(),
            data={"expiration_mode":
                      constants.flow_session_expiration_mode.roll_over})
        self.assertEqual(resp.status_code, 403)
        fs = models.FlowSession.objects.last()
        self.assertEqual(fs.expiration_mode,
                         constants.flow_session_expiration_mode.end)

    def test_post_invalid_bookmark_state(self):
        resp = self.client.post(
            self.get_relate_update_expiration_mode_url(),
            data={"expiration_mode": "unknown"})
        self.assertEqual(resp.status_code, 400)
        fs = models.FlowSession.objects.last()
        self.assertEqual(fs.expiration_mode,
                         constants.flow_session_expiration_mode.end)

    def test_not_you_session(self):
        with self.temporarily_switch_to_user(self.ta_participation.user):
            resp = self.client.post(
                self.get_relate_update_expiration_mode_url(),
                data={"expiration_mode":
                          constants.flow_session_expiration_mode.roll_over})
            self.assertEqual(resp.status_code, 403)
            fs = models.FlowSession.objects.last()
            self.assertEqual(fs.expiration_mode,
                             constants.flow_session_expiration_mode.end)


class RegradeFlowsViewTest(SingleCourseQuizPageTestMixin, TestCase):
    # test flow.regrade_flows_view

    def setUp(self):
        super().setUp()
        self.client.force_login(self.instructor_participation.user)

        fake_regrade_task = mock.patch(
                "course.tasks.regrade_flow_sessions.delay")
        self.mock_regrade_task = fake_regrade_task.start()
        self.addCleanup(fake_regrade_task.stop)

        fake_redirect = mock.patch("course.flow.redirect")
        self.mock_redirect = fake_redirect.start()
        self.addCleanup(fake_redirect.stop)

    def get_regrade_flows_view_url(self, course_identifier=None):
        course_identifier = course_identifier or self.get_default_course_identifier()
        return reverse("relate-regrade_flows_view", args=(course_identifier,))

    def test_no_pperm(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.client.get(self.get_regrade_flows_view_url())
            self.assertEqual(resp.status_code, 403)

            self.assertEqual(self.mock_regrade_task.call_count, 0)
            self.assertEqual(self.mock_redirect.call_count, 0)

            resp = self.client.post(
                self.get_regrade_flows_view_url(),
                data={
                    "regraded_session_in_progress": "yes",
                    "flow_id": QUIZ_FLOW_ID,
                    "access_rules_tag": "",
                }
            )
            self.assertEqual(resp.status_code, 403)

            self.assertEqual(self.mock_regrade_task.call_count, 0)
            self.assertEqual(self.mock_redirect.call_count, 0)

    def test_form_error(self):
        with mock.patch(
                "course.flow.RegradeFlowForm.is_valid") as mock_form_is_valid:
            mock_form_is_valid.return_value = False