test_flow.py

            unknown_count=0
        )
        self.assertEqual(g_info.points_percent(), 0)

    def test_unreachable_points_percent_max_points_none(self):
        g_info = flow.GradeInfo(
            points=1,
            provisional_points=20,
            max_points=None,
            max_reachable_points=20,
            fully_correct_count=5,
            partially_correct_count=0,
            incorrect_count=0,
            unknown_count=0
        )
        self.assertEqual(g_info.unreachable_points_percent(), 0)

    def test_unreachable_points_percent_max_points_zero(self):
        g_info = flow.GradeInfo(
            points=1,
            provisional_points=20,
            max_points=0,
            max_reachable_points=20,
            fully_correct_count=5,
            partially_correct_count=0,
            incorrect_count=0,
            unknown_count=0
        )
        self.assertEqual(g_info.unreachable_points_percent(), 0)

    def test_unreachable_points_percent_max_reachable_points_zero(self):
        g_info = flow.GradeInfo(
            points=1,
            provisional_points=20,
            max_points=20,
            max_reachable_points=None,
            fully_correct_count=5,
            partially_correct_count=0,
            incorrect_count=0,
            unknown_count=0
        )
        self.assertEqual(g_info.unreachable_points_percent(), 0)

    def test_unreachable_points_percent_full(self):
        g_info = flow.GradeInfo(
            points=1,
            provisional_points=20,
            max_points=20,
            max_reachable_points=0,
            fully_correct_count=5,
            partially_correct_count=0,
            incorrect_count=0,
            unknown_count=0
        )

        # for visualization purposes
        self.assertTrue(
            99 < g_info.unreachable_points_percent() < 100)


class FinishFlowSessionViewTest(HackRepoMixin,
                                SingleCourseQuizPageTestMixin, TestCase):
    # test flow.finish_flow_session_view

    @classmethod
    def setUpTestData(cls):  # noqa
        super(FinishFlowSessionViewTest, cls).setUpTestData()
        cls.course.active_git_commit_sha = (
            "my_fake_commit_sha_for_finish_flow_session")
        cls.course.save()
        cls.student = cls.student_participation.user

    def setUp(self):
        super(FinishFlowSessionViewTest, self).setUp()
        self.c.force_login(self.student)

        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 test_submit_all_correct(self):
        # with human graded questions not graded

        self.start_flow(self.flow_id)

        page_ids = self.get_current_page_ids()
        for page_id in page_ids:
            self.submit_page_answer_by_page_id_and_test(page_id)

        resp = self.end_flow()

        expected_grade_info_dict = {
            "fully_correct_count": 2,
            "incorrect_count": 0,
            "max_points": 12.0,
            "max_reachable_points": 7.0,
            "optional_fully_correct_count": 1,
            "optional_incorrect_count": 0,
            "optional_partially_correct_count": 0,
            "optional_unknown_count": 1,
            "partially_correct_count": 0,
            "points": None,
            "provisional_points": 7.0,
            "unknown_count": 1
        }

        self.assertGradeInfoEqual(resp, expected_grade_info_dict)

        resp = self.c.get(self.get_finish_flow_session_view_url())
        self.assertGradeInfoEqual(resp, expected_grade_info_dict)

    def test_submit_all_correct_all_graded(self):
        # with human graded questions graded

        self.start_flow(self.flow_id)

        page_ids = self.get_current_page_ids()
        for page_id in page_ids:
            self.submit_page_answer_by_page_id_and_test(page_id)

        self.end_flow()

        for page_id in page_ids:
            self.submit_page_human_grading_by_page_id_and_test(
                page_id, do_session_score_equal_assersion=False)

        expected_grade_info_dict = {
            "fully_correct_count": 3,
            "incorrect_count": 0,
            "max_points": 12.0,
            "max_reachable_points": 12.0,
            "optional_fully_correct_count": 2,
            "optional_incorrect_count": 0,
            "optional_partially_correct_count": 0,
            "optional_unknown_count": 0,
            "partially_correct_count": 0,
            "points": 12.0,
            "provisional_points": 12.0,
            "unknown_count": 0
        }
        resp = self.c.get(self.get_finish_flow_session_view_url())
        self.assertEqual(resp.status_code, 200)
        self.assertGradeInfoEqual(resp, expected_grade_info_dict)

    def test_submit_with_partial_correct_or_incorrect_and_all_graded(self):
        self.start_flow(self.flow_id)

        page_ids = self.get_current_page_ids()
        for page_id in page_ids:
            if page_id not in ["matrix_props", "half"]:
                self.submit_page_answer_by_page_id_and_test(page_id)

        self.submit_page_answer_by_page_id_and_test(
            "matrix_props", answer_data={"choice": ['0']}, expected_grades=0.5
        )

        self.submit_page_answer_by_page_id_and_test(
            "half", answer_data={"answer": ['1']}, expected_grades=0
        )

        resp = self.end_flow()

        expected_grade_info_dict = {
            "fully_correct_count": 1,
            "incorrect_count": 1,
            "max_points": 12.0,
            "max_reachable_points": 7.0,
            "optional_fully_correct_count": 0,
            "optional_incorrect_count": 0,
            "optional_partially_correct_count": 1,
            "optional_unknown_count": 1,
            "partially_correct_count": 0,
            "points": None,
            "provisional_points": 2.0,
            "unknown_count": 1
        }
        self.assertGradeInfoEqual(resp, expected_grade_info_dict)

        for page_id in page_ids:
            if page_id == "anyup":
                self.submit_page_human_grading_by_page_id_and_test(
                    page_id, grade_data={"grade_percent": "0", "released": "on"},
                    do_session_score_equal_assersion=False)
            if page_id == "proof":
                self.submit_page_human_grading_by_page_id_and_test(
                    page_id,
                    grade_data={"grade_percent": "70", "released": "on"},
                    do_session_score_equal_assersion=False)
            else:
                self.submit_page_human_grading_by_page_id_and_test(
                    page_id, do_session_score_equal_assersion=False)

        expected_grade_info_dict = {
            "fully_correct_count": 1,
            "incorrect_count": 1,
            "max_points": 12.0,
            "max_reachable_points": 12.0,
            "optional_fully_correct_count": 1,
            "optional_incorrect_count": 0,
            "optional_partially_correct_count": 1,
            "optional_unknown_count": 0,
            "partially_correct_count": 1,
            "points": 5.5,
            "provisional_points": 5.5,
            "unknown_count": 0
        }
        resp = self.c.get(self.get_finish_flow_session_view_url())
        self.assertEqual(resp.status_code, 200)
        self.assertGradeInfoEqual(resp, expected_grade_info_dict)

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

    def test_submit_with_bonus(self):  # noqa
        with mock.patch(
                "course.flow.get_session_grading_rule") as mock_get_grule:
            mock_get_grule.return_value = \
                self.get_hacked_session_grading_rule(bonus_points=2)

            self.start_flow(self.flow_id)

            page_ids = self.get_current_page_ids()
            for page_id in page_ids:
                if page_id not in ["matrix_props", "half"]:
                    self.submit_page_answer_by_page_id_and_test(page_id)

            self.submit_page_answer_by_page_id_and_test(
                "matrix_props", answer_data={"choice": ['0']},
                expected_grades=0.5
            )

            self.submit_page_answer_by_page_id_and_test(
                "half", answer_data={"answer": ['1']}, expected_grades=0
            )

            self.end_flow()

            for page_id in page_ids:
                if page_id == "anyup":
                    self.submit_page_human_grading_by_page_id_and_test(
                        page_id,
                        grade_data={"grade_percent": "0", "released": "on"},
                        do_session_score_equal_assersion=False)
                if page_id == "proof":
                    self.submit_page_human_grading_by_page_id_and_test(
                        page_id,
                        grade_data={"grade_percent": "70", "released": "on"},
                        do_session_score_equal_assersion=False)
                else:
                    self.submit_page_human_grading_by_page_id_and_test(
                        page_id, do_session_score_equal_assersion=False)

            expected_grade_info_dict = {
                "fully_correct_count": 1,
                "incorrect_count": 1,
                "max_points": 14.0,
                "max_reachable_points": 14.0,
                "optional_fully_correct_count": 1,
                "optional_incorrect_count": 0,
                "optional_partially_correct_count": 1,
                "optional_unknown_count": 0,
                "partially_correct_count": 1,
                "points": 7.5,
                "provisional_points": 7.5,
                "unknown_count": 0
            }
            resp = self.c.get(self.get_finish_flow_session_view_url())
            self.assertEqual(resp.status_code, 200)
            self.assertGradeInfoEqual(resp, expected_grade_info_dict)

    def test_submit_with_max_points_enforced_cap_no_answers(self):
        with mock.patch(
                "course.flow.get_session_grading_rule") as mock_get_grule:
            mock_get_grule.return_value = mock_get_grule.return_value = \
                self.get_hacked_session_grading_rule(max_points_enforced_cap=10)

            self.start_flow(self.flow_id)

            # {{{ no answers
            resp = self.end_flow()

            expected_grade_info_dict = {
                "fully_correct_count": 0,
                "incorrect_count": 3,
                "max_points": 12.0,
                "max_reachable_points": 10,
                "optional_fully_correct_count": 0,
                "optional_incorrect_count": 2,
                "optional_partially_correct_count": 0,
                "optional_unknown_count": 0,
                "partially_correct_count": 0,
                "points": 0.0,
                "provisional_points": 0.0,
                "unknown_count": 0
            }
            self.assertGradeInfoEqual(resp, expected_grade_info_dict)
            # }}}

    def test_submit_with_max_points_enforced_cap(self):
        with mock.patch(
                "course.flow.get_session_grading_rule") as mock_get_grule:
            mock_get_grule.return_value = mock_get_grule.return_value = \
                self.get_hacked_session_grading_rule(
                    max_points_enforced_cap=10)

            self.start_flow(self.flow_id)

            # answer all questions
            page_ids = self.get_current_page_ids()
            for page_id in page_ids:
                self.submit_page_answer_by_page_id_and_test(page_id)

            resp = self.end_flow()

            expected_grade_info_dict = {
                "fully_correct_count": 2,
                "incorrect_count": 0,
                "max_points": 12.0,
                "max_reachable_points": 7.0,
                "optional_fully_correct_count": 1,
                "optional_incorrect_count": 0,
                "optional_partially_correct_count": 0,
                "optional_unknown_count": 1,
                "partially_correct_count": 0,
                "points": None,
                "provisional_points": 7.0,
                "unknown_count": 1
            }
            self.assertGradeInfoEqual(resp, expected_grade_info_dict)

    def test_submit_with_max_points_enforced_cap2(self):
        with mock.patch(
                "course.flow.get_session_grading_rule") as mock_get_grule:
            mock_get_grule.return_value = mock_get_grule.return_value = \
                self.get_hacked_session_grading_rule(
                    # lower than provisional_points
                    max_points_enforced_cap=6)

            self.start_flow(self.flow_id)

            # answer all questions
            page_ids = self.get_current_page_ids()
            for page_id in page_ids:
                self.submit_page_answer_by_page_id_and_test(page_id)
            #
            resp = self.end_flow()

            expected_grade_info_dict = {
                "fully_correct_count": 2,
                "incorrect_count": 0,
                "max_points": 12.0,
                "max_reachable_points": 6.0,
                "optional_fully_correct_count": 1,
                "optional_incorrect_count": 0,
                "optional_partially_correct_count": 0,
                "optional_unknown_count": 1,
                "partially_correct_count": 0,
                "points": None,
                "provisional_points": 6.0,
                "unknown_count": 1
            }
            self.assertGradeInfoEqual(resp, expected_grade_info_dict)

    def test_submit_with_max_points(self):
        with mock.patch(
                "course.flow.get_session_grading_rule") as mock_get_grule:
            mock_get_grule.return_value = mock_get_grule.return_value = \
                self.get_hacked_session_grading_rule(max_points=11)

            self.start_flow(self.flow_id)

            # no answers
            resp = self.end_flow()

            expected_grade_info_dict = {
                "fully_correct_count": 0,
                "incorrect_count": 3,
                "max_points": 11,
                "max_reachable_points": 12.0,
                "optional_fully_correct_count": 0,
                "optional_incorrect_count": 2,
                "optional_partially_correct_count": 0,
                "optional_unknown_count": 0,
                "partially_correct_count": 0,
                "points": 0.0,
                "provisional_points": 0.0,
                "unknown_count": 0
            }
            self.assertGradeInfoEqual(resp, expected_grade_info_dict)

    def get_hacked_session_access_rule(self, **kwargs):
        from course.utils import FlowSessionAccessRule
        defaults = {
            "permissions": [fperm.view, fperm.end_session],
        }
        defaults.update(kwargs)
        return FlowSessionAccessRule(**defaults)

    def test_no_view_fperm(self):
        self.start_flow(self.flow_id)
        with mock.patch(
                "course.flow.get_session_access_rule") as mock_get_arule:
            mock_get_arule.return_value = (
                self.get_hacked_session_access_rule(
                    permissions=[fperm.end_session]))

            # fail for get
            resp = self.c.get(self.get_finish_flow_session_view_url())
            self.assertEqual(resp.status_code, 403)

            # fail for post
            resp = self.end_flow()
            self.assertEqual(resp.status_code, 403)

    def test_no_end_session_fperm(self):
        self.start_flow(self.flow_id)
        with mock.patch(
                "course.flow.get_session_access_rule") as mock_get_arule:
            mock_get_arule.return_value = (
                self.get_hacked_session_access_rule(
                    permissions=[fperm.view]))

            resp = self.end_flow()
            self.assertEqual(resp.status_code, 403)

    def test_odd_post_parameter(self):
        self.start_flow(self.flow_id)
        resp = self.end_flow(post_parameter="unknown")
        self.assertEqual(resp.status_code, 400)

    def test_finish_non_in_progress_session(self):
        fs = factories.FlowSessionFactory(
            course=self.course, participation=self.student_participation,
            in_progress=False
        )
        # re-submit finish flow
        resp = self.end_flow(
            course_identifier=self.course.identifier,
            flow_session_id=fs.pk)
        self.assertEqual(resp.status_code, 403)
        self.assertEqual(self.mock_add_message.call_count, 1)
        self.assertIn(
            "Cannot end a session that's already ended",
            self.mock_add_message.call_args[0])

    def test_notify_on_submit_emtpy(self):
        with mock.patch("course.utils.get_flow_desc") as mock_get_flow_desc:
            mock_get_flow_desc.return_value = (
                self.get_hacked_flow_desc(
                    # no recepient
                    notify_on_submit=[])
            )
            self.start_flow(self.flow_id)
            self.end_flow()

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

    def test_notify_on_submit_no_grade_identifier(self):
        with mock.patch(
                "course.flow.get_session_grading_rule") as mock_get_grule:
            mock_get_grule.return_value = \
                self.get_hacked_session_grading_rule(grade_identifier=None)
            notify_on_submit_emails = ["test_notif@example.com"]
            with mock.patch("course.utils.get_flow_desc") as mock_get_flow_desc:
                mock_get_flow_desc.return_value = (
                    self.get_hacked_flow_desc(
                        notify_on_submit=notify_on_submit_emails)
                )
                self.start_flow(self.flow_id)

                fs = models.FlowSession.objects.first()
                fs.participation = None
                fs.user = None
                fs.save()

                self.end_flow()

            self.assertEqual(len(mail.outbox), 1)
            expected_review_uri = reverse("relate-view_flow_page",
                                          args=(
                                              self.course.identifier,
                                              fs.id, 0))

            self.assertIn(
                expected_review_uri,
                mail.outbox[0].body)

    def test_notify_on_submit_no_participation(self):
        notify_on_submit_emails = ["test_notif@example.com"]
        with mock.patch("course.utils.get_flow_desc") as mock_get_flow_desc:
            mock_get_flow_desc.return_value = (
                self.get_hacked_flow_desc(
                    notify_on_submit=notify_on_submit_emails)
            )
            self.start_flow(self.flow_id)

            fs = models.FlowSession.objects.first()
            fs.participation = None
            fs.user = None
            fs.save()

            self.end_flow()

        self.assertEqual(len(mail.outbox), 1)
        expected_review_uri = reverse("relate-view_flow_page",
                                      args=(
                                          self.course.identifier,
                                          fs.id, 0))

        self.assertIn(
            expected_review_uri,
            mail.outbox[0].body)

    def test_notify_on_submit(self):
        notify_on_submit_emails = ["test_notif@example.com"]
        with mock.patch("course.utils.get_flow_desc") as mock_get_flow_desc:
            mock_get_flow_desc.return_value = (
                self.get_hacked_flow_desc(
                    notify_on_submit=notify_on_submit_emails)
            )
            self.start_flow(self.flow_id)
            self.end_flow()

        gopp = models.GradingOpportunity.objects.first()

        expected_review_uri = reverse("relate-view_single_grade",
                                      args=(
                                          self.course.identifier,
                                          self.student_participation.pk,
                                          gopp.pk
                                      ))

        self.assertEqual(len(mail.outbox), 1)
        self.assertEqual(
            mail.outbox[0].recipients(),
            notify_on_submit_emails + [self.course.notify_email])

        self.assertIn(
            self.student_participation.user.username,
            mail.outbox[0].body)
        self.assertIn(
            expected_review_uri,
            mail.outbox[0].body)
        self.assertIn(
            self.student_participation.user.username,
            mail.outbox[0].subject)

    def test_notify_on_submit_use_masked_profile(self):
        self.mock_will_use_masked_profile_for_email.return_value = True
        notify_on_submit_emails = [
            "test_notif@example.com", self.ta_participation.user.email]
        with mock.patch("course.utils.get_flow_desc") as mock_get_flow_desc:
            mock_get_flow_desc.return_value = (
                self.get_hacked_flow_desc(
                    notify_on_submit=notify_on_submit_emails)
            )
            self.start_flow(self.flow_id)
            self.end_flow()

        gopp = models.GradingOpportunity.objects.first()
        expected_review_uri = reverse("relate-view_single_grade",
                                      args=(
                                          self.course.identifier,
                                          self.student_participation.pk,
                                          gopp.pk
                                      ))

        self.assertEqual(len(mail.outbox), 1)
        self.assertIn(
            expected_review_uri,
            mail.outbox[0].body)

        self.assertNotIn(
            self.student_participation.user.username,
            mail.outbox[0].body)
        self.assertNotIn(
            self.student_participation.user.username,
            mail.outbox[0].subject)

        self.assertNotIn(
            self.student_participation.user.get_full_name(),
            mail.outbox[0].body)
        self.assertNotIn(
            self.student_participation.user.get_full_name(),
            mail.outbox[0].subject)

    def test_get_finish_non_interactive_flow(self):
        resp = self.start_flow(flow_id="001-linalg-recap")
        self.assertEqual(resp.status_code, 302)
        resp = self.c.get(self.get_finish_flow_session_view_url())
        self.assertEqual(resp.status_code, 200)
        self.assertTemplateUsed(resp, "course/flow-completion.html")

    def test_get_finish_interactive_flow_with_unfinished_pages(self):
        self.start_flow(flow_id=self.flow_id)
        resp = self.c.get(self.get_finish_flow_session_view_url())
        self.assertEqual(resp.status_code, 200)
        self.assertTemplateUsed(resp, "course/flow-confirm-completion.html")
        self.assertResponseHasNoContext(resp, "grade_info")
        self.assertResponseContextEqual(resp, "answered_count", 3)

    def test_post_finish_non_interactive_flow(self):
        with mock.patch(
                "course.flow.get_session_access_rule") as mock_get_arule:
            # This has to be done, or we won't be able to end the flow session,
            # though the session doesn't need to be ended.
            mock_get_arule.return_value = (
                self.get_hacked_session_access_rule(
                    permissions=[fperm.view, fperm.end_session]))
            resp = self.start_flow(flow_id="001-linalg-recap")
            self.assertEqual(resp.status_code, 302)
            resp = self.end_flow()
            self.assertEqual(resp.status_code, 200)
            self.assertTemplateUsed(resp, "course/flow-completion.html")

    def test_post_finish_with_cannot_see_flow_result_access_rule(self):
        with mock.patch(
                "course.flow.get_session_access_rule") as mock_get_arule:
            # This has to be done, or we won't be able to end the flow session,
            # though the session doesn't need to be ended.
            mock_get_arule.return_value = (
                self.get_hacked_session_access_rule(
                    permissions=[
                        fperm.view, fperm.end_session,
                        fperm.cannot_see_flow_result]))
            resp = self.start_flow(self.flow_id)
            self.assertEqual(resp.status_code, 302)
            resp = self.end_flow()
            self.assertEqual(resp.status_code, 200)
            self.assertTemplateUsed(resp, "course/flow-completion-grade.html")
            self.assertResponseContextIsNone(resp, "grade_info")

            # Then we get the finish page
            resp = self.c.get(self.get_finish_flow_session_view_url())
            self.assertEqual(resp.status_code, 200)
            self.assertTemplateUsed(resp, "course/flow-completion-grade.html")
            self.assertResponseContextIsNone(resp, "grade_info")


class FinishFlowSessionTest(SingleCourseTestMixin, TestCase):
    # test flow.finish_flow_session
    def setUp(self):
        super(FinishFlowSessionTest, self).setUp()
        self.fctx = mock.MagicMock()

    def test_finish_non_in_progress_session(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=False)
        grading_rule = mock.MagicMock()

        with mock.patch(
                "course.flow.assemble_answer_visits") as mock_asv, mock.patch(
                "course.flow.grade_page_visits") as mock_gpv, mock.patch(
                "course.flow.grade_flow_session") as mock_gfs:
            expected_error_msg = "Can't end a session that's already ended"
            with self.assertRaises(RuntimeError) as cm:
                flow.finish_flow_session(self.fctx, flow_session, grading_rule)

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

            self.assertEqual(mock_asv.call_count, 0)
            self.assertEqual(mock_gpv.call_count, 0)
            self.assertEqual(mock_gfs.call_count, 0)

    def test_now_datetime(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=True)
        grading_rule = FlowSessionGradingRule(
            grade_identifier="la_quiz",
            grade_aggregation_strategy=g_strategy.use_latest,
            due=None,
            generates_grade=True,
            use_last_activity_as_completion_time=False
        )
        force_regrade = mock.MagicMock()
        respect_preview = mock.MagicMock

        now_datetime = now() - timedelta(days=1)
        with mock.patch(
                "course.flow.assemble_answer_visits") as mock_asv, mock.patch(
                "course.flow.grade_page_visits") as mock_gpv, mock.patch(
                "course.flow.grade_flow_session") as mock_gfs:
            flow.finish_flow_session(
                self.fctx, flow_session, grading_rule, force_regrade,
                now_datetime, respect_preview)

            self.assertEqual(flow_session.completion_time, now_datetime)
            self.assertFalse(flow_session.in_progress)

            self.assertEqual(mock_asv.call_count, 1)
            self.assertEqual(mock_gpv.call_count, 1)
            self.assertEqual(mock_gfs.call_count, 1)

            # make sure "answer_visits" is not None when calling
            # grade_flow_session
            self.assertIn(mock_asv.return_value, mock_gfs.call_args[0])

    def test_now_datetime_none(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=True)
        grading_rule = FlowSessionGradingRule(
            grade_identifier="la_quiz",
            grade_aggregation_strategy=g_strategy.use_latest,  # noqa
            due=None,
            generates_grade=True,
            use_last_activity_as_completion_time=False
        )
        force_regrade = mock.MagicMock()
        respect_preview = mock.MagicMock

        now_datetime = None

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

        with mock.patch(
                "course.flow.assemble_answer_visits") as mock_asv, mock.patch(
                "course.flow.grade_page_visits") as mock_gpv, mock.patch(
                "course.flow.grade_flow_session") as mock_gfs, mock.patch(
                "django.utils.timezone.now") as mock_now:
            mock_now.return_value = faked_now
            flow.finish_flow_session(
                self.fctx, flow_session, grading_rule, force_regrade,
                now_datetime, respect_preview)

            self.assertEqual(flow_session.completion_time, faked_now)
            self.assertFalse(flow_session.in_progress)

            self.assertEqual(mock_asv.call_count, 1)
            self.assertEqual(mock_gpv.call_count, 1)
            self.assertEqual(mock_gfs.call_count, 1)

            # make sure "answer_visits" is not None when calling
            # grade_flow_session
            self.assertIn(mock_asv.return_value, mock_gfs.call_args[0])

    def test_rule_use_last_activity_as_completion_time_no_last_activity(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=True)

        grading_rule = FlowSessionGradingRule(
            grade_identifier="la_quiz",
            grade_aggregation_strategy=g_strategy.use_latest,  # noqa
            due=None,
            generates_grade=True,
            use_last_activity_as_completion_time=True
        )
        force_regrade = mock.MagicMock()
        respect_preview = mock.MagicMock

        now_datetime = None
        faked_now = now() - timedelta(days=1)

        with mock.patch(
                "course.flow.assemble_answer_visits") as mock_asv, mock.patch(
                "course.flow.grade_page_visits") as mock_gpv, mock.patch(
                "course.flow.grade_flow_session") as mock_gfs, mock.patch(
                "django.utils.timezone.now") as mock_now:
            mock_now.return_value = faked_now

            flow.finish_flow_session(
                self.fctx, flow_session, grading_rule, force_regrade,
                now_datetime, respect_preview)

            self.assertIsNotNone(flow_session.completion_time)
            self.assertFalse(flow_session.in_progress)

            self.assertEqual(mock_asv.call_count, 1)
            self.assertEqual(mock_gpv.call_count, 1)
            self.assertEqual(mock_gfs.call_count, 1)

            # make sure "answer_visits" is not None when calling
            # grade_flow_session
            self.assertIn(mock_asv.return_value, mock_gfs.call_args[0])

            self.assertEqual(flow_session.completion_time, faked_now)

    def test_rule_use_last_activity_as_completion_time(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=True)
        page_data = factories.FlowPageDataFactory(
            flow_session=flow_session
        )

        answer_visit_time = now() - timedelta(days=1)
        factories.FlowPageVisitFactory(
            page_data=page_data, answer={"answer": "hi"},
            visit_time=answer_visit_time)

        null_answer_visit_time = now() + timedelta(minutes=60)

        # this visit happened after the answer visit
        factories.FlowPageVisitFactory(
            page_data=page_data, answer=None,
            visit_time=null_answer_visit_time)

        grading_rule = FlowSessionGradingRule(
            grade_identifier="la_quiz",
            grade_aggregation_strategy=g_strategy.use_latest,  # noqa
            due=None,
            generates_grade=True,
            use_last_activity_as_completion_time=True
        )
        force_regrade = mock.MagicMock()
        respect_preview = mock.MagicMock

        now_datetime = now()

        with mock.patch(
                "course.flow.assemble_answer_visits") as mock_asv, mock.patch(
                "course.flow.grade_page_visits") as mock_gpv, mock.patch(
                "course.flow.grade_flow_session") as mock_gfs:
            flow.finish_flow_session(
                self.fctx, flow_session, grading_rule, force_regrade,
                now_datetime, respect_preview)

            self.assertIsNotNone(flow_session.completion_time)
            self.assertFalse(flow_session.in_progress)

            self.assertEqual(mock_asv.call_count, 1)
            self.assertEqual(mock_gpv.call_count, 1)
            self.assertEqual(mock_gfs.call_count, 1)

            # make sure "answer_visits" is not None when calling
            # grade_flow_session
            self.assertIn(mock_asv.return_value, mock_gfs.call_args[0])

            self.assertEqual(flow_session.completion_time, answer_visit_time)


class ExpireFlowSessionTest(SingleCourseTestMixin, TestCase):
    # test flow.expire_flow_session
    def setUp(self):
        super(ExpireFlowSessionTest, self).setUp()
        self.fctx = mock.MagicMock()

        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_finish_flow_session = mock.patch("course.flow.finish_flow_session")
        self.mock_finish_flow_session = fake_finish_flow_session.start()
        self.addCleanup(fake_finish_flow_session.stop)

        fake_get_session_start_rule = mock.patch(
            "course.flow.get_session_start_rule")
        self.mock_get_session_start_rule = fake_get_session_start_rule.start()
        self.addCleanup(fake_get_session_start_rule.stop)

        self.now_datatime = now()

    def test_expire_non_in_progess_session(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=False)
        grading_rule = mock.MagicMock()

        expected_error_msg = "Can't expire a session that's not in progress"
        with self.assertRaises(RuntimeError) as cm:
            flow.expire_flow_session(
                self.fctx, flow_session, grading_rule, self.now_datatime)

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

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 0)
        self.assertEqual(self.mock_finish_flow_session.call_count, 0)
        self.assertEqual(self.mock_get_session_start_rule.call_count, 0)

    def test_expire_session_of_anonymous_user(self):
        flow_session = factories.FlowSessionFactory(
            course=self.course, participation=None, user=None, in_progress=True)
        grading_rule = mock.MagicMock()

        expected_error_msg = "Can't expire an anonymous flow session"
        with self.assertRaises(RuntimeError) as cm:
            flow.expire_flow_session(
                self.fctx, flow_session, grading_rule, self.now_datatime)

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

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 0)
        self.assertEqual(self.mock_finish_flow_session.call_count, 0)
        self.assertEqual(self.mock_get_session_start_rule.call_count, 0)

    def test_past_due_only_due_none(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=True)
        grading_rule = FlowSessionGradingRule(
            grade_identifier="la_quiz",
            grade_aggregation_strategy=g_strategy.use_latest,  # noqa
            due=None,
            generates_grade=True,
            use_last_activity_as_completion_time=False
        )

        self.assertFalse(flow.expire_flow_session(
            self.fctx, flow_session, grading_rule, self.now_datatime,
            past_due_only=True))

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 0)
        self.assertEqual(self.mock_finish_flow_session.call_count, 0)
        self.assertEqual(self.mock_get_session_start_rule.call_count, 0)

    def test_past_due_only_now_datetime_not_due(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=True)

        due = self.now_datatime + timedelta(hours=1)

        grading_rule = FlowSessionGradingRule(
            grade_identifier="la_quiz",
            grade_aggregation_strategy=g_strategy.use_latest,
            due=due,
            generates_grade=True,
            use_last_activity_as_completion_time=False
        )

        self.assertFalse(flow.expire_flow_session(
            self.fctx, flow_session, grading_rule, self.now_datatime,
            past_due_only=True))

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 0)
        self.assertEqual(self.mock_finish_flow_session.call_count, 0)
        self.assertEqual(self.mock_get_session_start_rule.call_count, 0)

    def test_past_due_only_now_datetime_other_case(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=True)

        due = self.now_datatime - timedelta(hours=1)

        grading_rule = FlowSessionGradingRule(
            grade_identifier="la_quiz",
            grade_aggregation_strategy=g_strategy.use_latest,
            due=due,
            generates_grade=True,
            use_last_activity_as_completion_time=False
        )

        flow.expire_flow_session(
            self.fctx, flow_session, grading_rule, self.now_datatime,
            past_due_only=True)

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 1)
        self.assertEqual(self.mock_finish_flow_session.call_count, 1)
        self.assertEqual(self.mock_get_session_start_rule.call_count, 0)

    def test_expiration_mode_rollover_not_may_start_new_session(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=True,
            expiration_mode=constants.flow_session_expiration_mode.roll_over
        )

        self.mock_get_session_start_rule.return_value = (
            FlowSessionStartRule(
                tag_session="roll_over_tag",
                may_start_new_session=False
            ))

        grading_rule = FlowSessionGradingRule(
            grade_identifier="la_quiz",
            grade_aggregation_strategy=g_strategy.use_latest,
            due=None,
            generates_grade=True,
            use_last_activity_as_completion_time=False
        )

        self.assertTrue(flow.expire_flow_session(
            self.fctx, flow_session, grading_rule, self.now_datatime))

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 1)
        self.assertEqual(self.mock_finish_flow_session.call_count, 1)
        self.assertEqual(self.mock_get_session_start_rule.call_count, 1)
        self.assertTrue(
            self.mock_get_session_start_rule.call_args[1]["for_rollover"])

    def test_expiration_mode_end(self):
        flow_session = factories.FlowSessionFactory(