test_flow.py

            "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.client.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):
        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.client.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_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)

            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.client.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)

            # {{{ 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)

            # 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)

            # 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)

            # 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 test_no_view_fperm(self):
        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.client.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):
        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):
        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)
            )
            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.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_notify_on_submit_no_participation_use_masked_profile(self):
        self.mock_will_use_masked_profile_for_email.return_value = True
        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)
            )
            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_get_finish_non_interactive_flow(self):
        resp = self.start_flow(flow_id="001-linalg-recap")
        self.assertEqual(resp.status_code, 302)
        resp = self.client.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.client.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.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.client.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().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().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(
            participation=self.student_participation, in_progress=True,
            expiration_mode=constants.flow_session_expiration_mode.end
        )

        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, 0)

    def test_invalid_expiration_mode(self):
        flow_session = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=True,
            expiration_mode="unknown"
        )

        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
        )

        expected_error_msg = ("invalid expiration mode 'unknown' "
                              "on flow session ID %i" % flow_session.pk)
        with self.assertRaises(ValueError) 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, 1)
        self.assertEqual(self.mock_finish_flow_session.call_count, 0)
        self.assertEqual(self.mock_get_session_start_rule.call_count, 0)


@pytest.mark.django_db
class GetFlowSessionAttemptIdTest(unittest.TestCase):
    # test flow.get_flow_session_attempt_id

    def setUp(self):
        def remove_all_course():
            for course in models.Course.objects.all():
                course.delete()

        self.addCleanup(remove_all_course)

    def test(self):
        course = factories.CourseFactory()
        participation = factories.ParticipationFactory(course=course)
        fs1 = factories.FlowSessionFactory(participation=participation)
        fs2 = factories.FlowSessionFactory(participation=participation)

        self.assertNotEqual(
            flow.get_flow_session_attempt_id(fs1),
            flow.get_flow_session_attempt_id(fs2))

        self.assertNotEqual(flow.get_flow_session_attempt_id(fs1), "main")
        self.assertNotEqual(flow.get_flow_session_attempt_id(fs2), "main")


class GradeFlowSessionTest(SingleCourseQuizPageTestMixin,
                           HackRepoMixin, TestCase):
    # test flow.grade_flow_session

    initial_commit_sha = "my_fake_commit_sha_for_grade_flow_session"

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

        self.fctx = mock.MagicMock()
        self.fctx.title = "my flow session title"

        fake_gather_grade_info = mock.patch("course.flow.gather_grade_info")
        self.mock_gather_grade_info = fake_gather_grade_info.start()
        self.addCleanup(fake_gather_grade_info.stop)

        fake_assemble_answer_visits = mock.patch(
            "course.flow.assemble_answer_visits")
        self.mock_assemble_answer_visits = fake_assemble_answer_visits.start()
        self.addCleanup(fake_assemble_answer_visits.stop)

        fake_get_flow_grading_opportunity = mock.patch(
            "course.models.get_flow_grading_opportunity")
        self.mock_get_flow_grading_opportunity = \
            fake_get_flow_grading_opportunity.start()
        self.gopp = factories.GradingOpportunityFactory(course=self.course)
        self.mock_get_flow_grading_opportunity.return_value = (self.gopp)
        self.addCleanup(fake_get_flow_grading_opportunity.stop)

    def get_test_grade_info(self, **kwargs):
        defaults = {
            "fully_correct_count": 0,
            "incorrect_count": 0,
            "max_points": 10.0,
            "max_reachable_points": 8.0,
            "optional_fully_correct_count": 0,
            "optional_incorrect_count": 0,
            "optional_partially_correct_count": 0,
            "optional_unknown_count": 0,
            "partially_correct_count": 1,
            "points": 5.0,
            "provisional_points": 8.0,
            "unknown_count": 0
        }
        defaults.update(kwargs)
        return flow.GradeInfo(**defaults)

    def get_test_grading_rule(self, **kwargs):
        defaults = {
            "grade_identifier": "la_quiz",
            "grade_aggregation_strategy": g_strategy.use_latest,
            "due": None,
            "generates_grade": True,
            "use_last_activity_as_completion_time": False,
            "credit_percent": 100
        }
        defaults.update(kwargs)
        return FlowSessionGradingRule(**defaults)

    def get_default_test_session(self, **kwargs):
        defaults = {"participation": self.student_participation,
                    "in_progress": False}
        defaults.update(kwargs)
        return factories.FlowSessionFactory(**defaults)

    def test_answer_visits_none(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule()
        answer_visits = None

        grade_info = self.get_test_grade_info()
        self.mock_gather_grade_info.return_value = grade_info

        result = flow.grade_flow_session(
            self.fctx, flow_session, grading_rule, answer_visits)

        # when answer_visits is None, assemble_answer_visits should be called
        self.assertEqual(self.mock_assemble_answer_visits.call_count, 1)

        self.assertEqual(self.mock_get_flow_grading_opportunity.call_count, 1)

        flow_session.refresh_from_db()
        self.assertEqual(flow_session.points, 5)
        self.assertEqual(flow_session.max_points, 10)