test_flow.py

        self.assertIsNone(flow_session.result_comment)

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()
        self.assertEqual(current_grade_changes.count(), 1)

    def test_not_append_comments_when_no_points(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule(credit_percent=110)
        answer_visits = mock.MagicMock()

        grade_info = self.get_test_grade_info(points=None)
        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 not None, assemble_answer_visits should not be
        # called
        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)

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

        flow_session.refresh_from_db()
        self.assertEqual(flow_session.points, None)
        self.assertEqual(flow_session.max_points, 10)
        self.assertIsNone(flow_session.result_comment)

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()

        # no points won't result in grade change objects creation.
        self.assertEqual(current_grade_changes.count(), 0)

    def test_not_append_comments_when_no_credit_percent(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule(credit_percent=None)
        answer_visits = mock.MagicMock()

        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 not None, assemble_answer_visits should not be
        # called
        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)

        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)
        self.assertIsNone(flow_session.result_comment)

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()
        self.assertEqual(current_grade_changes.count(), 1)

    def test_not_append_comments(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule(credit_percent=110)
        answer_visits = mock.MagicMock()

        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 not None, assemble_answer_visits should not be
        # called
        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)

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

        flow_session.refresh_from_db()
        self.assertEqual(flow_session.points, 5.5)
        self.assertEqual(flow_session.max_points, 10)
        self.assertEqual(
            flow_session.result_comment, "Counted at 110.0% of 5.0 points")

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()
        self.assertEqual(current_grade_changes.count(), 1)

    def test_no_grade_identifier(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule(grade_identifier=None)
        answer_visits = mock.MagicMock()

        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 not None, assemble_answer_visits should not be
        # called
        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)

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

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

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()
        self.assertEqual(current_grade_changes.count(), 0)

    def test_not_generates_grade(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule(generates_grade=False)
        answer_visits = mock.MagicMock()

        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 not None, assemble_answer_visits should not be
        # called
        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)

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

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

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()
        self.assertEqual(current_grade_changes.count(), 0)

    def test_session_has_no_participation(self):
        flow_session = self.get_default_test_session(
            course=self.course,
            participation=None, user=None)
        grading_rule = self.get_test_grading_rule()
        answer_visits = mock.MagicMock()

        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 not None, assemble_answer_visits should not be
        # called
        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)

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

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

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()
        self.assertEqual(current_grade_changes.count(), 0)

    def create_grade_change(self, flow_session, **kwargs):
        from course.flow import get_flow_session_attempt_id
        defaults = {
            "flow_session": flow_session,
            "opportunity": self.gopp,
            "participation": self.student_participation,
            "state": constants.grade_state_change_types.graded,
            "attempt_id": get_flow_session_attempt_id(flow_session),
            "points": 5,
            "max_points": 10,
            "comment": None,
            "grade_time": now()
        }
        defaults.update(kwargs)
        factories.GradeChangeFactory(**defaults)

    def test_has_identical_previous_grade_changes(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule()
        answer_visits = mock.MagicMock()

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

        # create an identical previous_grade_changes
        grade_time1 = now() - timedelta(days=1)
        grade_time2 = grade_time1 + timedelta(hours=1)

        self.create_grade_change(
            flow_session=flow_session,
            grade_time=grade_time1,
            comment="no comments")

        # to ensure the second one is used, it has a different comment with
        # the above one
        self.create_grade_change(
            flow_session=flow_session,
            grade_time=grade_time2,
            comment=None)

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

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

        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)
        self.assertIsNone(flow_session.result_comment)

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()

        # no new grade change objects is created
        self.assertEqual(current_grade_changes.count(), 2)

    def test_previous_grade_change_points_changed(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule()
        answer_visits = mock.MagicMock()

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

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

        self.create_grade_change(
            flow_session=flow_session,
            grade_time=grade_time1,
            points=4)

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

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

        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)
        self.assertIsNone(flow_session.result_comment)

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()

        # a new grade change objects is created
        self.assertEqual(current_grade_changes.count(), 2)
        self.assertEqual(current_grade_changes.last().points, 5)

    def test_previous_grade_change_max_points_changed(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule()
        answer_visits = mock.MagicMock()

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

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

        self.create_grade_change(
            flow_session=flow_session,
            grade_time=grade_time1,
            max_points=12)

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

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

        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)
        self.assertIsNone(flow_session.result_comment)

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()

        # a new grade change objects is created
        self.assertEqual(current_grade_changes.count(), 2)
        self.assertEqual(current_grade_changes.last().max_points, 10)

    def test_previous_grade_change_gchange_state_changed(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule()
        answer_visits = mock.MagicMock()

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

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

        self.create_grade_change(
            flow_session=flow_session,
            grade_time=grade_time1,
            state="other state")

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

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

        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)
        self.assertIsNone(flow_session.result_comment)

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()

        # a new grade change objects is created
        self.assertEqual(current_grade_changes.count(), 2)
        self.assertEqual(current_grade_changes.last().state,
                         constants.grade_state_change_types.graded)

    def test_previous_grade_change_comment_different(self):
        flow_session = self.get_default_test_session()
        grading_rule = self.get_test_grading_rule()
        answer_visits = mock.MagicMock()

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

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

        self.create_grade_change(
            flow_session=flow_session,
            grade_time=grade_time1,
            comment="no comments")

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

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

        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)
        self.assertIsNone(flow_session.result_comment)

        # it should return the grade_info calculated
        self.assertEqual(result, grade_info)

        current_grade_changes = models.GradeChange.objects.all()

        # a new grade change objects is created
        self.assertEqual(current_grade_changes.count(), 2)
        self.assertEqual(current_grade_changes.last().comment, None)


@pytest.mark.django_db
class UnsubmitPageTest(unittest.TestCase):
    # test flow.unsubmit_page

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

        self.addCleanup(remove_all_course)

    def test(self):
        now_datetime = now() + timedelta(days=1)
        fpv = factories.FlowPageVisitFactory(is_submitted_answer=True,
                                             remote_address="127.0.0.1",
                                             is_synthetic=False)

        exist_fpv_count = models.FlowPageVisit.objects.count()
        self.assertEqual(exist_fpv_count, 1)
        flow.unsubmit_page(fpv, now_datetime)

        fpvs = models.FlowPageVisit.objects.all()
        self.assertEqual(fpvs.count(), 2)

        new_fpv = fpvs.last()
        self.assertEqual(new_fpv.visit_time, now_datetime)
        self.assertIsNone(new_fpv.remote_address)
        self.assertIsNone(new_fpv.user)
        self.assertTrue(new_fpv.is_synthetic)
        self.assertFalse(new_fpv.is_submitted_answer)


class ReopenSessionTest(SingleCourseTestMixin, TestCase):
    # test flow.reopen_session
    def setUp(self):
        super().setUp()

        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_unsubmit_page = mock.patch(
            "course.flow.unsubmit_page")
        self.mock_unsubmit_page = fake_unsubmit_page.start()
        self.addCleanup(fake_unsubmit_page.stop)

        self.default_now_datetime = now()

    def get_test_flow_session(self, **kwargs):
        defaults = {"course": self.course,
                    "participation": self.student_participation,
                    "points": 5,
                    "max_points": 10,
                    "result_comment": "Hi blahblah",
                    "completion_time": now() - timedelta(days=1)
                    }
        defaults.update(kwargs)
        return factories.FlowSessionFactory(**defaults)

    def test_in_progress(self):
        flow_session = self.get_test_flow_session(in_progress=True)
        expected_error_msg = "Cannot reopen a session that's already in progress"
        with self.assertRaises(RuntimeError) as cm:
            flow.reopen_session(now(), flow_session)
        self.assertIn(expected_error_msg, str(cm.exception))

        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)
        self.assertEqual(self.mock_unsubmit_page.call_count, 0)

    def test_session_without_participation(self):
        flow_session = self.get_test_flow_session(participation=None, user=None)
        expected_error_msg = "Cannot reopen anonymous sessions"
        with self.assertRaises(RuntimeError) as cm:
            flow.reopen_session(self.default_now_datetime, flow_session)
        self.assertIn(expected_error_msg, str(cm.exception))

        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)
        self.assertEqual(self.mock_unsubmit_page.call_count, 0)

    def test_not_suppress_log(self):
        flow_session = self.get_test_flow_session()

        original_comment = flow_session.result_comment

        flow.reopen_session(self.default_now_datetime, flow_session)

        flow_session.refresh_from_db()
        self.assertTrue(flow_session.in_progress)
        self.assertIsNone(flow_session.points)
        self.assertIsNone(flow_session.max_points)
        self.assertIsNone(flow_session.completion_time)

        self.assertIn(original_comment, flow_session.result_comment)
        self.assertIn("Session reopened at", flow_session.result_comment)
        self.assertIn("previous completion time was", flow_session.result_comment)

        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)
        self.assertEqual(self.mock_unsubmit_page.call_count, 0)

    def test_suppress_log(self):
        flow_session = self.get_test_flow_session()

        original_comment = flow_session.result_comment

        flow.reopen_session(
            self.default_now_datetime, flow_session, suppress_log=True)

        flow_session.refresh_from_db()
        self.assertTrue(flow_session.in_progress)
        self.assertIsNone(flow_session.points)
        self.assertIsNone(flow_session.max_points)
        self.assertIsNone(flow_session.completion_time)

        self.assertEqual(flow_session.result_comment, original_comment)

        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)
        self.assertEqual(self.mock_unsubmit_page.call_count, 0)

    def test_unsubmit_pages(self):
        flow_session = self.get_test_flow_session()
        original_comment = flow_session.result_comment

        # three not none faked answer visits
        self.mock_assemble_answer_visits.return_value = [
            None, mock.MagicMock(), None, mock.MagicMock(), mock.MagicMock()
        ]
        flow.reopen_session(
            self.default_now_datetime, flow_session, unsubmit_pages=True)

        flow_session.refresh_from_db()
        self.assertTrue(flow_session.in_progress)
        self.assertIsNone(flow_session.points)
        self.assertIsNone(flow_session.max_points)
        self.assertIsNone(flow_session.completion_time)
        self.assertNotEqual(flow_session.result_comment, original_comment)

        self.assertEqual(self.mock_assemble_answer_visits.call_count, 1)
        self.assertEqual(self.mock_unsubmit_page.call_count, 3)


class FinishFlowSessionStandaloneTest(SingleCourseTestMixin, TestCase):
    # test flow.finish_flow_session_standalone

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

        fake_get_session_grading_rule = mock.patch(
            "course.flow.get_session_grading_rule")
        self.mock_get_session_grading_rule = fake_get_session_grading_rule.start()
        self.addCleanup(fake_get_session_grading_rule.stop)

        fake_flow_context = mock.patch("course.flow.FlowContext")
        self.mock_flow_context = fake_flow_context.start()
        self.fctx = mock.MagicMock()
        self.mock_flow_context.return_value = self.fctx
        self.addCleanup(fake_flow_context.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)

        self.default_now_datetime = now()
        self.repo = mock.MagicMock()

    def get_test_flow_session(self, **kwargs):
        defaults = {"course": self.course,
                    "participation": self.student_participation,
                    "points": 5,
                    "max_points": 10,
                    "result_comment": "Hi blahblah",
                    "completion_time": now() - timedelta(days=1),
                    "in_progress": True
                    }
        defaults.update(kwargs)
        return factories.FlowSessionFactory(**defaults)

    def test_no_now_datetime(self):
        flow_session = self.get_test_flow_session()
        fake_grading_rule = self.get_hacked_session_grading_rule()
        self.mock_get_session_grading_rule.return_value = fake_grading_rule

        with mock.patch(
                "django.utils.timezone.now") as mock_now:
            fake_now = mock.MagicMock()
            mock_now.return_value = fake_now
            self.assertTrue(flow.finish_flow_session_standalone(
                self.repo, self.course, flow_session
            ))

        self.assertEqual(self.mock_finish_flow_session.call_count, 1)
        for arg in [self.fctx, fake_grading_rule]:
            self.assertIn(arg, self.mock_finish_flow_session.call_args[0])

        self.assertEqual(
            self.mock_finish_flow_session.call_args[1]["now_datetime"],
            fake_now
        )

    def test_has_now_datetime(self):
        flow_session = self.get_test_flow_session()
        fake_grading_rule = self.get_hacked_session_grading_rule()
        self.mock_get_session_grading_rule.return_value = fake_grading_rule

        self.assertTrue(flow.finish_flow_session_standalone(
            self.repo, self.course, flow_session,
            now_datetime=self.default_now_datetime,
        ))

        self.assertEqual(self.mock_finish_flow_session.call_count, 1)
        for arg in [self.fctx, fake_grading_rule]:
            self.assertIn(arg, self.mock_finish_flow_session.call_args[0])

        self.assertEqual(
            self.mock_finish_flow_session.call_args[1]["now_datetime"],
            self.default_now_datetime
        )

    def test_past_due_only_due_is_none(self):
        flow_session = self.get_test_flow_session()
        fake_grading_rule = self.get_hacked_session_grading_rule()
        self.mock_get_session_grading_rule.return_value = fake_grading_rule

        self.assertFalse(flow.finish_flow_session_standalone(
            self.repo, self.course, flow_session,
            now_datetime=self.default_now_datetime,
            past_due_only=True
        ))

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

    def test_past_due_only_not_due(self):
        flow_session = self.get_test_flow_session()
        fake_grading_rule = self.get_hacked_session_grading_rule(
            due=now() + timedelta(days=1))
        self.mock_get_session_grading_rule.return_value = fake_grading_rule

        self.assertFalse(flow.finish_flow_session_standalone(
            self.repo, self.course, flow_session,
            now_datetime=self.default_now_datetime,
            past_due_only=True
        ))

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

    def test_past_due_only_due(self):
        flow_session = self.get_test_flow_session()
        fake_grading_rule = self.get_hacked_session_grading_rule(
            due=now() - timedelta(days=1))
        self.mock_get_session_grading_rule.return_value = fake_grading_rule

        self.assertTrue(flow.finish_flow_session_standalone(
            self.repo, self.course, flow_session,
            now_datetime=self.default_now_datetime,
            past_due_only=True
        ))

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

    def test_other_kwargs_used_for_call_finish_flow_session(self):
        flow_session = self.get_test_flow_session()
        fake_grading_rule = self.get_hacked_session_grading_rule(
            due=now() - timedelta(days=1))
        self.mock_get_session_grading_rule.return_value = fake_grading_rule

        force_regrade = mock.MagicMock()
        respect_preview = mock.MagicMock()

        self.assertTrue(flow.finish_flow_session_standalone(
            self.repo, self.course, flow_session,
            force_regrade=force_regrade,
            respect_preview=respect_preview
        ))

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

        self.assertEqual(
            self.mock_finish_flow_session.call_args[1]["force_regrade"],
            force_regrade
        )

        self.assertEqual(
            self.mock_finish_flow_session.call_args[1]["respect_preview"],
            respect_preview
        )


class ExpireFlowSessionStandaloneTest(SingleCourseTestMixin, TestCase):
    # test flow.expire_flow_session_standalone

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

        fake_get_session_grading_rule = mock.patch(
            "course.flow.get_session_grading_rule")
        self.mock_get_session_grading_rule = fake_get_session_grading_rule.start()
        self.addCleanup(fake_get_session_grading_rule.stop)

        fake_flow_context = mock.patch("course.flow.FlowContext")
        self.mock_flow_context = fake_flow_context.start()
        self.fctx = mock.MagicMock()
        self.mock_flow_context.return_value = self.fctx
        self.addCleanup(fake_flow_context.stop)

        fake_expire_flow_session = mock.patch("course.flow.expire_flow_session")
        self.mock_expire_flow_session = fake_expire_flow_session.start()
        self.addCleanup(fake_expire_flow_session.stop)

        self.default_now_datetime = now()
        self.repo = mock.MagicMock()

    def get_test_flow_session(self, **kwargs):
        defaults = {"course": self.course,
                    "participation": self.student_participation,
                    "points": 5,
                    "max_points": 10,
                    "result_comment": "Hi blahblah",
                    "completion_time": now() - timedelta(days=1),
                    "in_progress": True
                    }
        defaults.update(kwargs)
        return factories.FlowSessionFactory(**defaults)

    def test_past_due_only_default_false(self):
        flow_session = self.get_test_flow_session()
        fake_grading_rule = self.get_hacked_session_grading_rule()
        self.mock_get_session_grading_rule.return_value = fake_grading_rule

        self.assertTrue(flow.expire_flow_session_standalone(
            self.repo, self.course, flow_session,
            self.default_now_datetime
        ))

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

        for arg in [fake_grading_rule, self.default_now_datetime]:
            self.assertIn(arg, self.mock_expire_flow_session.call_args[0])

        self.assertEqual(
            self.mock_expire_flow_session.call_args[1]["past_due_only"],
            False
        )

    def test_args_kwargs_used_for_call_expire_flow_session(self):
        flow_session = self.get_test_flow_session()
        fake_grading_rule = self.get_hacked_session_grading_rule(
            due=now() - timedelta(days=1))
        self.mock_get_session_grading_rule.return_value = fake_grading_rule

        past_due_only = mock.MagicMock()

        self.assertTrue(flow.expire_flow_session_standalone(
            self.repo, self.course, flow_session,
            self.default_now_datetime,
            past_due_only=past_due_only
        ))

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

        for arg in [fake_grading_rule, self.default_now_datetime]:
            self.assertIn(arg, self.mock_expire_flow_session.call_args[0])

        self.assertEqual(
            self.mock_expire_flow_session.call_args[1]["past_due_only"],
            past_due_only
        )


class RegradeSessionTest(SingleCourseTestMixin, TestCase):
    # test flow.regrade_session
    def setUp(self):
        super().setUp()

        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_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_finish_flow_session_standalone = mock.patch(
            "course.flow.finish_flow_session_standalone")
        self.mock_finish_flow_session_standalone = (
            fake_finish_flow_session_standalone.start())
        self.addCleanup(fake_finish_flow_session_standalone.stop)

        fake_reopen_session = mock.patch(
            "course.flow.reopen_session")
        self.mock_reopen_session = (
            fake_reopen_session.start())
        self.addCleanup(fake_reopen_session.stop)

        fake_grade_page_visit = mock.patch(
            "course.flow.grade_page_visit")
        self.mock_grade_page_visit = (
            fake_grade_page_visit.start())
        self.addCleanup(fake_grade_page_visit.stop)

        self.repo = mock.MagicMock()

    def get_test_flow_session(self, **kwargs):
        defaults = {"course": self.course,
                    "participation": self.student_participation,
                    "points": 5,
                    "max_points": 10,
                    "completion_time": None,
                    "in_progress": True
                    }
        defaults.update(kwargs)
        return factories.FlowSessionFactory(**defaults)

    def test_session_in_progress(self):
        flow_session = self.get_test_flow_session()

        answer_visit1 = mock.MagicMock()
        answer_visit1.get_most_recent_grade.return_value = mock.MagicMock()

        answer_visit2 = mock.MagicMock()
        answer_visit2.get_most_recent_grade.return_value = None

        self.mock_assemble_answer_visits.return_value = [
            None, answer_visit1, None, answer_visit2]

        flow.regrade_session(self.repo, self.course, flow_session)

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 1)
        self.assertFalse(
            self.mock_adjust_flow_session_page_data.call_args[1]["respect_preview"])

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

        self.assertEqual(self.mock_grade_page_visit.call_count, 1)
        self.assertFalse(
            self.mock_grade_page_visit.call_args[1]["respect_preview"])

        self.assertEqual(self.mock_reopen_session.call_count, 0)
        self.assertEqual(self.mock_finish_flow_session_standalone.call_count, 0)

    def test_session_not_in_progress(self):
        flow_session = self.get_test_flow_session(
            in_progress=False, completion_time=now() - timedelta(days=1))

        flow.regrade_session(self.repo, self.course, flow_session)

        flow_session.refresh_from_db()
        self.assertIn("Session regraded at", flow_session.result_comment)

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 1)
        self.assertFalse(
            self.mock_adjust_flow_session_page_data.call_args[1]["respect_preview"])

        self.assertEqual(self.mock_reopen_session.call_count, 1)
        self.assertTrue(
            self.mock_reopen_session.call_args[1]["force"])
        self.assertTrue(
            self.mock_reopen_session.call_args[1]["suppress_log"])

        self.assertEqual(self.mock_finish_flow_session_standalone.call_count, 1)
        self.assertFalse(
            self.mock_finish_flow_session_standalone.call_args[1][
                "respect_preview"])

        self.assertEqual(self.mock_assemble_answer_visits.call_count, 0)
        self.assertEqual(self.mock_grade_page_visit.call_count, 0)


class RecalculateSessionGradeTest(SingleCourseTestMixin, TestCase):
    def setUp(self):
        super().setUp()

        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_reopen_session = mock.patch(
            "course.flow.reopen_session")
        self.mock_reopen_session = (
            fake_reopen_session.start())
        self.addCleanup(fake_reopen_session.stop)

        fake_finish_flow_session_standalone = mock.patch(
            "course.flow.finish_flow_session_standalone")
        self.mock_finish_flow_session_standalone = (
            fake_finish_flow_session_standalone.start())
        self.addCleanup(fake_finish_flow_session_standalone.stop)

        self.repo = mock.MagicMock()

    def get_test_flow_session(self, **kwargs):
        defaults = {"course": self.course,
                    "participation": self.student_participation,
                    "points": 5,
                    "max_points": 10,
                    "completion_time": now() - timedelta(days=1),
                    "in_progress": False
                    }
        defaults.update(kwargs)
        return factories.FlowSessionFactory(**defaults)

    def test_session_in_progress(self):
        flow_session = self.get_test_flow_session(
            in_progress=True, completion_time=None)
        expected_error_msg = "cannot recalculate grade on in-progress session"
        with self.assertRaises(RuntimeError) as cm:
            flow.recalculate_session_grade(self.repo, self.course, flow_session)
        self.assertIn(expected_error_msg, str(cm.exception))

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

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 0)
        self.assertEqual(self.mock_reopen_session.call_count, 0)
        self.assertEqual(self.mock_finish_flow_session_standalone.call_count, 0)

    def test_session_not_in_progress(self):
        flow_session = self.get_test_flow_session()
        flow.recalculate_session_grade(self.repo, self.course, flow_session)

        flow_session.refresh_from_db()
        self.assertIn("Session grade recomputed at", flow_session.result_comment)

        self.assertEqual(self.mock_adjust_flow_session_page_data.call_count, 1)
        self.assertFalse(
            self.mock_adjust_flow_session_page_data.call_args[1]["respect_preview"])

        self.assertEqual(self.mock_reopen_session.call_count, 1)
        self.assertTrue(
            self.mock_reopen_session.call_args[1]["force"])
        self.assertTrue(
            self.mock_reopen_session.call_args[1]["suppress_log"])

        self.assertEqual(self.mock_finish_flow_session_standalone.call_count, 1)
        self.assertFalse(
            self.mock_finish_flow_session_standalone.call_args[1][
                "respect_preview"])


@pytest.mark.django_db
class LockDownIfNeededTest(unittest.TestCase):
    # test flow.lock_down_if_needed
    def setUp(self):
        super().setUp()
        self.flow_session = mock.MagicMock()
        self.flow_session.id = 1
        self.flow_session.pk = 1
        self.request = mock.MagicMock()
        self.request.session = {}

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

        self.addCleanup(remove_all_course)

    def test_no_lock_down_as_exam_session_flow_permission(self):
        flow_permissions = ["other_flow_permission"]
        flow.lock_down_if_needed(self.request, flow_permissions, self.flow_session)

        self.assertIsNone(
            self.request.session.get(
                "relate_session_locked_to_exam_flow_session_pk"))

    def test_has_lock_down_as_exam_session_flow_permission(self):
        flow_permissions = [fperm.lock_down_as_exam_session,
                            "other_flow_permission"]
        flow.lock_down_if_needed(self.request, flow_permissions, self.flow_session)

        self.assertEqual(
            self.request.session.get(
                "relate_session_locked_to_exam_flow_session_pk"),
            self.flow_session.pk
        )


class ViewStartFlowTest(SingleCourseTestMixin, TestCase):
    # test flow.view_start_flow

    flow_id = QUIZ_FLOW_ID

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

        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_flow_context = mock.patch("course.flow.FlowContext")
        self.mock_flow_context = fake_flow_context.start()
        self.fctx = mock.MagicMock()
        self.fctx.flow_id = self.flow_id
        self.fctx.flow_desc = dict_to_struct(
            {"title": "test page title", "description_html": "foo bar"})
        self.mock_flow_context.return_value = self.fctx
        self.addCleanup(fake_flow_context.stop)

        fake_post_start_flow = mock.patch("course.flow.post_start_flow")