test_grade_book.py

            participation=self.ta_participation, points=3)))

        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_single_grade(self.student_participation, self.gopp)
        self.assertResponseContextEqual(resp, "avg_grade_percentage", 6)
        self.assertResponseContextEqual(resp, "avg_grade_population", 1)

        temp_ptcp = factories.ParticipationFactory.create(
            course=self.course)

        factories.GradeChangeFactory.create(
            **(self.gc(participation=temp_ptcp, points=3)))
        with self.temporarily_switch_to_user(temp_ptcp.user):
            resp = self.get_view_single_grade(temp_ptcp, self.gopp)
        self.assertResponseContextEqual(resp, "avg_grade_percentage", 4.5)
        self.assertResponseContextEqual(resp, "avg_grade_population", 2)

    def test_append_gc(self):
        self.use_default_setup()
        self.append_gc(self.gc(points=8, flow_session=self.session2))
        self.assertGradeChangeMachineReadableStateEqual(8)
        self.assertGradeChangeStateEqual("8.0% (/3)")

        self.append_gc(self.gc(points=0, flow_session=self.session2))
        self.assertGradeChangeMachineReadableStateEqual(0)
        self.assertGradeChangeStateEqual("0.0% (/3)")

    def test_update_latest_gc_of_latest_finished_session(self):
        self.use_default_setup()
        self.assertGradeChangeMachineReadableStateEqual(6)

        self.update_gc(self.gc_session2, points=10)
        self.assertGradeChangeMachineReadableStateEqual(10)
        self.assertGradeChangeStateEqual("10.0% (/3)")

    def test_update_ealiest_gc_of_ealier_finished_session(self):
        self.use_default_setup()
        self.assertGradeChangeMachineReadableStateEqual(6)

        self.update_gc(self.gc_main_2, update_time=False, points=15)
        self.assertGradeChangeMachineReadableStateEqual(6)
        self.assertGradeChangeStateEqual("6.0% (/3)")

    def test_gc_without_attempt_id(self):
        # TODO: Is it a bug? percentage of GradeChanges without attempt_id are
        # put at the begining of the valid_percentages list.

        # Uncomment the following to see the failure
        # self.use_default_setup()
        # self.assertGradeChangeMachineReadableStateEqual(6)
        # print(self.gc_main_1.grade_time)
        #
        # self.time_increment()

        # create a gc without attempt_id
        gc = factories.GradeChangeFactory.create(  # noqa
            **(self.gc(points=8.5, null_attempt_id=True)))  # noqa
        # print(gc.grade_time)

        machine = self.get_gc_machine()
        self.assertGradeChangeMachineReadableStateEqual(8.5)
        self.assertEqual(machine.valid_percentages, [8.5])

    def test_gc_unavailable(self):
        factories.GradeChangeFactory.create(**(self.gc(points=9.1)))
        factories.GradeChangeFactory.create(
            **(self.gc(points=0, state=g_state.unavailable)))
        machine = self.get_gc_machine()
        self.assertGradeChangeMachineReadableStateEqual("OTHER_STATE")
        self.assertEqual(machine.valid_percentages, [])
        self.assertGradeChangeStateEqual("(other state)")

        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_single_grade(self.student_participation, self.gopp)
        self.assertResponseContextEqual(resp, "avg_grade_percentage", None)
        self.assertResponseContextEqual(resp, "avg_grade_population", 0)

        # failure when unavailable gc follows another grade change
        factories.GradeChangeFactory.create(**(self.gc(points=5)))

        with self.assertRaises(ValueError) as e:
            self.get_gc_stringify_machine_readable_state()
            self.assertIn("cannot accept grade once opportunity has been "
                            "marked 'unavailable'", e.exception)

    def test_gc_exempt(self):
        factories.GradeChangeFactory.create(**(self.gc(points=6)))
        factories.GradeChangeFactory.create(
            **(self.gc(points=0, state=g_state.exempt)))
        machine = self.get_gc_machine()
        self.assertGradeChangeMachineReadableStateEqual("EXEMPT")
        self.assertEqual(machine.valid_percentages, [])
        self.assertGradeChangeStateEqual("(exempt)")

        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_single_grade(self.student_participation, self.gopp)
        self.assertResponseContextEqual(resp, "avg_grade_percentage", None)
        self.assertResponseContextEqual(resp, "avg_grade_population", 0)

        # failure when exempt gc follows another grade change
        factories.GradeChangeFactory.create(**(self.gc(points=5)))

        with self.assertRaises(ValueError) as e:
            self.get_gc_stringify_machine_readable_state()
            self.assertIn("cannot accept grade once opportunity has been "
                            "marked 'exempt'", e.exception)

    def test_gc_do_over(self):
        factories.GradeChangeFactory.create(**(self.gc(points=6)))

        # This creates a GradeChange object with no attempt_id
        factories.GradeChangeFactory.create(
            **(self.gc(points=0, state=g_state.do_over,
                       null_attempt_id=True)))
        machine = self.get_gc_machine()
        self.assertGradeChangeMachineReadableStateEqual("NONE")
        self.assertEqual(machine.valid_percentages, [])
        self.assertGradeChangeStateEqual("- ∅ -")

        # This make sure new grade change objects following do_over gc is
        # consumed without problem
        factories.GradeChangeFactory.create(**(self.gc(points=5)))
        self.assertGradeChangeMachineReadableStateEqual("5")
        machine = self.get_gc_machine()
        self.assertEqual(machine.valid_percentages, [5])
        self.assertGradeChangeStateEqual("5.0%")

    def test_gc_do_over_average_grade_value(self):
        self.use_default_setup()
        factories.GradeChangeFactory.create(
            **(self.gc(points=None, state=g_state.do_over,
                       flow_session=self.session2)))

        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_single_grade(self.student_participation, self.gopp)
        self.assertResponseContextEqual(resp, "avg_grade_percentage", None)
        self.assertResponseContextEqual(resp, "avg_grade_population", 0)

    def test_gc_report_sent(self):
        factories.GradeChangeFactory.create(**(self.gc(points=6)))
        gc2 = factories.GradeChangeFactory.create(
            **(self.gc(points=0, state=g_state.report_sent)))
        machine = self.get_gc_machine()
        self.assertGradeChangeMachineReadableStateEqual("6")
        self.assertGradeChangeStateEqual("6.0%")
        self.assertEqual(machine.last_report_time, gc2.grade_time)

    def test_gc_extension(self):
        factories.GradeChangeFactory.create(**(self.gc(points=6)))
        gc2 = factories.GradeChangeFactory.create(
            **(self.gc(points=0, state=g_state.extension,
                       due_time=self.time + timedelta(days=1))))
        machine = self.get_gc_machine()
        self.assertGradeChangeMachineReadableStateEqual("6")
        self.assertGradeChangeStateEqual("6.0%")
        self.assertEqual(machine.due_time, gc2.due_time)

    def test_gc_grading_started(self):
        factories.GradeChangeFactory.create(**(self.gc(points=6)))
        factories.GradeChangeFactory.create(
            **(self.gc(points=0, state=g_state.grading_started)))
        self.assertGradeChangeMachineReadableStateEqual("6")
        self.assertGradeChangeStateEqual("6.0%")

    def test_gc_retrieved(self):
        factories.GradeChangeFactory.create(**(self.gc(points=6)))
        factories.GradeChangeFactory.create(
            **(self.gc(points=0, state=g_state.retrieved)))
        self.assertGradeChangeMachineReadableStateEqual("6")
        self.assertGradeChangeStateEqual("6.0%")

    def test_gc_non_exist_state(self):
        factories.GradeChangeFactory.create(**(self.gc(points=6)))
        factories.GradeChangeFactory.create(
            **(self.gc(points=0, state="some_state")))

        with self.assertRaises(RuntimeError):
            self.get_gc_stringify_machine_readable_state()

    def test_gc_non_point(self):
        factories.GradeChangeFactory.create(**(self.gc(points=None)))
        self.assertGradeChangeMachineReadableStateEqual("NONE")
        self.assertGradeChangeStateEqual("- ∅ -")

    # }}}


class ViewParticipantGradesTest2(GradesTestMixin, TestCase):
    def setUp(self):
        super(ViewParticipantGradesTest2, self).setUp()
        self.use_default_setup()
        self.gopp_hidden_in_gradebook = factories.GradingOpportunityFactory(
            course=self.course, aggregation_strategy=g_stragety.use_latest,
            flow_id=None, shown_in_grade_book=False,
            identifier="hidden_in_instructor_grade_book")

        self.gopp_hidden_in_gradebook = factories.GradingOpportunityFactory(
            course=self.course, aggregation_strategy=g_stragety.use_latest,
            flow_id=None, shown_in_grade_book=False,
            identifier="only_hidden_in_grade_book")

        self.gopp_hidden_in_participation_gradebook = (
            factories.GradingOpportunityFactory(
                course=self.course,
                shown_in_participant_grade_book=False,
                aggregation_strategy=g_stragety.use_latest,
                flow_id=None, identifier="all_hidden_in_ptcp_gradebook"))

        self.gopp_result_hidden_in_participation_gradebook = (
            factories.GradingOpportunityFactory(
                course=self.course, result_shown_in_participant_grade_book=False,
                aggregation_strategy=g_stragety.use_latest,
                flow_id=None, identifier="result_hidden_in_ptcp_gradebook"))

        self.gc_gopp_result_hidden = factories.GradeChangeFactory(
            **self.gc(points=66.67,
                      opportunity=self.gopp_result_hidden_in_participation_gradebook,
                      state=g_state.graded))

    def test_view_my_grade(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_my_grades()
            self.assertEqual(resp.status_code, 200)
            grade_table = self.get_response_context_value_by_name(
                resp, "grade_table")
            self.assertEqual((len(grade_table)), 2)
            self.assertEqual([g_info.opportunity.identifier
                              for g_info in grade_table],
                             [factories.DEFAULT_GRADE_IDENTIFIER,
                              "result_hidden_in_ptcp_gradebook"])

            # the grade is hidden
            self.assertNotContains(resp, 66.67)

            grade_participation = self.get_response_context_value_by_name(
                resp, "grade_participation")
            self.assertEqual(grade_participation.pk, self.student_participation.pk)

            # shown
            self.assertContains(resp, factories.DEFAULT_GRADE_IDENTIFIER)
            self.assertContains(resp, "result_hidden_in_ptcp_gradebook")

            # hidden
            self.assertNotContains(resp, "hidden_in_instructor_grade_book")
            self.assertNotContains(resp, "all_hidden_in_ptcp_gradebook")

    def test_view_participant_grades(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.get_view_participant_grades(self.student_participation.id)
            self.assertEqual(resp.status_code, 200)
            grade_table = self.get_response_context_value_by_name(
                resp, "grade_table")
            self.assertEqual((len(grade_table)), 3)
            self.assertEqual([g_info.opportunity.identifier
                              for g_info in grade_table],
                             ['all_hidden_in_ptcp_gradebook',
                              factories.DEFAULT_GRADE_IDENTIFIER,
                              "result_hidden_in_ptcp_gradebook"])

            # the grade hidden to participation is show to instructor
            # self.assertContains(resp, "66.67%(not released)")

            grade_participation = self.get_response_context_value_by_name(
                resp, "grade_participation")
            self.assertEqual(grade_participation.pk, self.student_participation.pk)

            # shown
            self.assertContains(resp, factories.DEFAULT_GRADE_IDENTIFIER)
            self.assertContains(resp, "result_hidden_in_ptcp_gradebook")
            self.assertContains(resp, "all_hidden_in_ptcp_gradebook")

            # hidden
            self.assertNotContains(resp, "hidden_in_instructor_grade_book")

        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_participant_grades(
                participation_id=self.instructor_participation.id)
            self.assertEqual(resp.status_code, 403)


class ViewReopenSessionTest(GradesTestMixin, TestCase):
    # grades.view_reopen_session (currently for cases not covered by other tests)

    gopp_id = "la_quiz"

    def setUp(self):
        super(ViewReopenSessionTest, self).setUp()
        self.fs1 = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=False)

        self.fs2 = factories.FlowSessionFactory(
            participation=self.student_participation, in_progress=True)

    def test_flow_desc_not_exist(self):
        with mock.patch("course.content.get_flow_desc") as mock_get_flow_desc:
            from django.core.exceptions import ObjectDoesNotExist
            mock_get_flow_desc.side_effect = ObjectDoesNotExist
            resp = self.get_reopen_session_view(
                self.gopp_id, flow_session_id=self.fs1.pk)
            self.assertEqual(resp.status_code, 404)

    def test_already_in_progress(self):
        # not unsubmit, because we don't have previoius grade visit (which will
        # result in error)
        data = {'set_access_rules_tag': ['<<<NONE>>>'],
                'comment': ['test reopen'],
                'reopen': ''}

        resp = self.post_reopen_session_view(
            self.gopp_id, flow_session_id=self.fs2.pk, data=data)
        self.assertEqual(resp.status_code, 200)

        self.assertEqual(self.mock_add_message.call_count, 1)
        self.assertIn(
            "Cannot reopen a session that's already in progress.",
            self.mock_add_message.call_args[0])
        self.assertTrue(self.fs2.in_progress)

    def test_reopen_success(self):
        resp = self.get_reopen_session_view(
            self.gopp_id, flow_session_id=self.fs1.pk)
        self.assertEqual(resp.status_code, 200)

        # not unsubmit, because we don't have previoius grade visit (which will
        # result in error)
        data = {'set_access_rules_tag': ['<<<NONE>>>'],
                'comment': ['test reopen'],
                'reopen': ''}

        resp = self.post_reopen_session_view(
            self.gopp_id, flow_session_id=self.fs1.pk, data=data)
        self.assertEqual(resp.status_code, 302)

        self.fs1.refresh_from_db()
        self.assertTrue(self.fs1.in_progress)

    def test_set_access_rule_tag(self):
        from relate.utils import dict_to_struct, struct_to_dict

        hacked_flow_desc_dict = self.get_hacked_flow_desc(as_dict=True)
        rules = hacked_flow_desc_dict["rules"]
        rules_dict = struct_to_dict(rules)
        rules_dict["tags"] = ["blahblah"]
        rules = dict_to_struct(rules_dict)
        hacked_flow_desc_dict["rules"] = rules
        hacked_flow_desc = dict_to_struct(hacked_flow_desc_dict)
        assert hacked_flow_desc.rules.tags == ["blahblah"]

        with mock.patch("course.content.get_flow_desc") as mock_get_flow_desc:
            mock_get_flow_desc.return_value = hacked_flow_desc

            # not unsubmit, because we don't have previoius grade visit (which will
            # result in error)
            data = {'set_access_rules_tag': ['blahblah'],
                    'comment': ['test reopen'],
                    'reopen': ''}

            resp = self.post_reopen_session_view(
                self.gopp_id, flow_session_id=self.fs1.pk, data=data)
            self.assertEqual(resp.status_code, 302)

        self.fs1.refresh_from_db()
        self.assertTrue(self.fs1.in_progress)
        self.assertEqual(self.fs1.access_rules_tag, 'blahblah')


class ViewSingleGradeTest(GradesTestMixin, TestCase):
    # grades.view_single_grade (currently for cases not covered by other tests)

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

        fake_regrade_session = mock.patch("course.flow.regrade_session")
        self.mock_regrade_session = fake_regrade_session.start()
        self.addCleanup(fake_regrade_session.stop)

        fake_recalculate_session_grade = mock.patch(
            "course.flow.recalculate_session_grade")
        self.mock_recalculate_session_grade = fake_recalculate_session_grade.start()
        self.addCleanup(fake_recalculate_session_grade.stop)

        fake_expire_flow_session_standalone = mock.patch(
            "course.flow.expire_flow_session_standalone")
        self.mock_expire_flow_session_standalone = (
            fake_expire_flow_session_standalone.start())
        self.addCleanup(fake_expire_flow_session_standalone.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)

    def test_participation_course_not_match(self):
        another_course_participation = factories.ParticipationFactory(
            course=factories.CourseFactory(identifier="another-course"))
        resp = self.get_view_single_grade(another_course_participation, self.gopp)
        self.assertEqual(resp.status_code, 400)

    def test_gopp_course_not_match(self):
        another_course_gopp = factories.GradingOpportunityFactory(
            course=factories.CourseFactory(identifier="another-course"),
            identifier=QUIZ_FLOW_ID)
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.c.get(self.get_single_grade_url(
                self.student_participation.pk, another_course_gopp.pk))
            self.assertEqual(resp.status_code, 400)

    def test_view_other_single_grade_no_pperm(self):
        another_participation = factories.ParticipationFactory(
            course=self.course)
        with self.temporarily_switch_to_user(another_participation.user):
            resp = self.get_view_single_grade(
                self.student_participation, self.gopp, force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)

            resp = self.post_view_single_grade(
                self.student_participation, self.gopp, data={},
                force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)

    def test_view_success(self):
        resp = self.get_view_single_grade(
            self.student_participation, self.gopp)
        self.assertEqual(resp.status_code, 200)

    def test_view_not_shown_in_grade_book(self):
        hidden_gopp = factories.GradingOpportunityFactory(
            course=self.course, identifier="hidden",
            shown_in_grade_book=False)

        resp = self.get_view_single_grade(
            self.student_participation, hidden_gopp)
        self.assertEqual(resp.status_code, 200)
        self.assertIn(
            "This grade is not shown in the grade book.",
            self.mock_add_message.call_args[0])

        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_single_grade(
                self.student_participation, hidden_gopp,
                force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)

    def test_view_not_shown_in_participant_grade_book(self):
        hidden_gopp = factories.GradingOpportunityFactory(
            course=self.course, identifier="hidden",
            shown_in_participant_grade_book=False)

        resp = self.get_view_single_grade(
            self.student_participation, hidden_gopp)
        self.assertEqual(resp.status_code, 200)
        self.assertIn(
            "This grade is not shown in the student grade book.",
            self.mock_add_message.call_args[0])

        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_single_grade(
                self.student_participation, hidden_gopp,
                force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)

    @unittest.skipIf(six.PY2, "PY2 doesn't support subTest")
    def test_post_no_pperm(self):
        another_participation = factories.ParticipationFactory(
            course=self.course)

        # only view_gradebook pperm
        pp = models.ParticipationPermission(
            participation=another_participation,
            permission=pperm.view_gradebook)
        pp.save()

        fs = factories.FlowSessionFactory(
            participation=self.student_participation, flow_id=self.flow_id)

        for op in ["imposedl", "end", "regrade", "recalculate"]:
            with self.subTest(op=op):
                resp = self.post_view_single_grade(
                    self.student_participation, self.gopp,
                    data={"%s_%d" % (op, fs.pk): ''},
                    force_login_instructor=False)
                self.assertEqual(resp.status_code, 403)

    def test_post_no_action_match(self):
        resp = self.post_view_single_grade(
            self.student_participation, self.gopp,
            data={"blablabal": ''})
        self.assertEqual(resp.status_code, 400)

    @unittest.skipIf(six.PY2, "PY2 doesn't support subTest")
    def test_post(self):
        fs = factories.FlowSessionFactory(
            participation=self.student_participation, flow_id=self.flow_id)

        tup = (
            ("imposedl", self.mock_expire_flow_session_standalone,
             "Session deadline imposed."),
            ("end", self.mock_finish_flow_session_standalone, "Session ended."),
            ("regrade", self.mock_regrade_session, "Session regraded."),
            ("recalculate", self.mock_recalculate_session_grade,
             "Session grade recalculated."))

        for op, mock_func, msg in tup:
            with self.subTest(op=op):
                resp = self.post_view_single_grade(
                    self.student_participation, self.gopp,
                    data={"%s_%d" % (op, fs.pk): ''})
                self.assertEqual(resp.status_code, 200)
                self.assertEqual(mock_func.call_count, 1)
                self.assertIn(msg, self.mock_add_message.call_args[0])

                mock_func.reset_mock()
                self.mock_add_message.reset_mock()

    def test_post_invalid_session_op(self):
        fs = factories.FlowSessionFactory(
            participation=self.student_participation, flow_id=self.flow_id)

        resp = self.post_view_single_grade(
            self.student_participation, self.gopp,
            data={"blablabal_%d" % fs.pk: ''})
        self.assertEqual(resp.status_code, 400)

    @unittest.skipIf(six.PY2, "PY2 doesn't support subTest")
    def test_post_keyboard_interrupt(self):
        fs = factories.FlowSessionFactory(
            participation=self.student_participation, flow_id=self.flow_id)

        tup = (
            ("imposedl", self.mock_expire_flow_session_standalone,
             "Session deadline imposed."),
            ("end", self.mock_finish_flow_session_standalone, "Session ended."),
            ("regrade", self.mock_regrade_session, "Session regraded."),
            ("recalculate", self.mock_recalculate_session_grade,
             "Session grade recalculated."))

        err = "foo"
        self.mock_regrade_session.side_effect = KeyboardInterrupt(err)
        self.mock_recalculate_session_grade.side_effect = KeyboardInterrupt(err)
        self.mock_expire_flow_session_standalone.side_effect = KeyboardInterrupt(err)
        self.mock_finish_flow_session_standalone.side_effect = KeyboardInterrupt(err)

        for op, mock_func, msg in tup:
            with self.subTest(op=op):
                resp = self.post_view_single_grade(
                    self.student_participation, self.gopp,
                    data={"%s_%d" % (op, fs.pk): ''})
                self.assertEqual(resp.status_code, 200)
                self.assertNotIn(msg, self.mock_add_message.call_args[0])
                self.assertIn(
                    "Error: KeyboardInterrupt %s" % err,
                    self.mock_add_message.call_args[0])

                mock_func.reset_mock()
                self.mock_add_message.reset_mock()

    def test_view_gopp_flow_desc_not_exist(self):
        with mock.patch("course.content.get_flow_desc") as mock_get_flow_desc:
            from django.core.exceptions import ObjectDoesNotExist
            mock_get_flow_desc.side_effect = ObjectDoesNotExist()
            resp = self.get_view_single_grade(
                self.student_participation, self.gopp)
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextIsNone(
                resp, "flow_sessions_and_session_properties")

    def test_view_gopp_no_flow_id(self):
        gopp = factories.GradingOpportunityFactory(
            course=self.course,
            identifier="no_flow_id",
            flow_id=None)
        factories.GradeChangeFactory(
            **self.gc(
                opportunity=gopp))
        resp = self.get_view_single_grade(
            self.student_participation, gopp)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextIsNone(
            resp, "flow_sessions_and_session_properties")

    def test_filter_out_pre_public_grade_changes(self):
        gopp = factories.GradingOpportunityFactory(
            course=self.course,
            identifier="no_flow_id",
            flow_id=None)
        # 5 gchanges

        factories.GradeChangeFactory(**self.gc(
            opportunity=gopp))
        factories.GradeChangeFactory(**self.gc(
            opportunity=gopp))
        factories.GradeChangeFactory(**self.gc(
            opportunity=gopp))
        fourth_gc = factories.GradeChangeFactory(**self.gc(
            opportunity=gopp))
        factories.GradeChangeFactory(**self.gc(
            opportunity=gopp))

        resp = self.get_view_single_grade(
            self.student_participation, gopp)
        self.assertEqual(resp.status_code, 200)
        resp_gchanges = resp.context["grade_changes"]
        self.assertEqual(len(resp_gchanges), 5)

        # update_gopp
        gopp.hide_superseded_grade_history_before = (
            fourth_gc.grade_time - timedelta(minutes=1))
        gopp.save()

        # view by instructor
        resp = self.get_view_single_grade(
            self.student_participation, gopp)
        self.assertEqual(resp.status_code, 200)
        resp_gchanges = resp.context["grade_changes"]
        self.assertEqual(len(resp_gchanges), 5)

        # view by student
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_single_grade(
                self.student_participation, gopp, force_login_instructor=False)
            self.assertEqual(resp.status_code, 200)
            resp_gchanges = resp.context["grade_changes"]
            self.assertEqual(len(resp_gchanges), 2)


@unittest.SkipTest
class FixingTest(GradesTestMixin, TestCase):
    # currently skipped

    def reopen_session1(self):
        existing_gc_count = models.GradeChange.objects.count()
        reopen_session(now_datetime=local_now(), session=self.session1,
                       generate_grade_change=True,
                       suppress_log=True)
        self.assertEqual(models.GradeChange.objects.count(), existing_gc_count+1)
        self.session1.refresh_from_db()

    def reopen_session2(self):
        existing_gc_count = models.GradeChange.objects.count()
        reopen_session(now_datetime=local_now(), session=self.session2,
                       generate_grade_change=True,
                       suppress_log=True)
        self.assertEqual(models.GradeChange.objects.count(), existing_gc_count+1)
        self.session2.refresh_from_db()

    def test_append_gc_with_session_after_reopen_session2(self):
        self.use_default_setup()
        self.reopen_session2()

        # append a grade change for session2
        # grade_time need to be specified, because the faked gc
        # is using fake time, while reopen a session will create
        # an actual gc using the actual time.

        latest_gc = models.GradeChange.objects.all().order_by("-grade_time")[0]

        self.append_gc(self.gc(points=12, flow_session=self.session2,
                               grade_time=now(),
                               effective_time=latest_gc.effective_time))
        self.assertGradeChangeMachineReadableStateEqual(12)
        self.assertGradeChangeStateEqual("12.00% (/3)")

    def test_append_nonsession_gc_after_reopen_session2(self):
        self.use_default_setup()
        self.reopen_session2()

        # Append a grade change without session
        # grade_time need to be specified, because the faked gc
        # is using fake time, while reopen a session will create
        # an actual gc using the actual time.
        self.append_gc(self.gc(points=11, grade_time=now()))
        self.assertGradeChangeMachineReadableStateEqual(11)
        self.assertGradeChangeStateEqual("11.00% (/3)")

    def test_new_gchange_created_when_finish_flow_use_last_has_activity(self):
        # With use_last_activity_as_completion_time = True, if a flow session HAS
        # last_activity, the expected effective_time of the new gchange should be
        # the last_activity() of the related flow_session.
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            self.start_flow(QUIZ_FLOW_ID)

            # create a flow page visit, then there should be last_activity() for
            # the session.
            self.post_answer_by_ordinal(1, {"answer": ['0.5']})
            self.assertEqual(
                models.FlowPageVisit.objects.filter(answer__isnull=False).count(),
                1)
            last_answered_visit = (
                models.FlowPageVisit.objects.filter(answer__isnull=False).first())
            last_answered_visit.visit_time = now() - timedelta(hours=1)
            last_answered_visit.save()
            self.assertEqual(models.GradeChange.objects.count(), 0)

            with mock.patch("course.flow.get_session_grading_rule") as \
                    mock_get_grading_rule:
                mock_get_grading_rule.side_effect = (
                    get_session_grading_rule_use_last_activity_as_cmplt_time_side_effect)  # noqa
                resp = self.end_flow()
                self.assertEqual(resp.status_code, 200)

            self.assertEqual(models.GradeChange.objects.count(), 1)
            latest_gchange = models.GradeChange.objects.last()
            latest_flow_session = models.FlowSession.objects.last()
            self.assertIsNotNone(latest_flow_session.last_activity())
            self.assertEqual(latest_flow_session.completion_time,
                             latest_flow_session.last_activity())
            self.assertEqual(latest_gchange.effective_time,
                             latest_flow_session.last_activity())

    # {{{ Fixed issue #263 and #417

    def test_update_latest_gc_of_ealier_finished_session(self):
        self.use_default_setup()
        self.assertGradeChangeMachineReadableStateEqual(6)

        # Issue #263 and #417
        # gc_session1 is the GradeChange object of session 1, update it's
        # value won't change the consumed state.
        self.update_gc(self.gc_session1, points=10)
        self.assertGradeChangeMachineReadableStateEqual(6)
        self.assertGradeChangeStateEqual("6.00% (/3)")

    def test_special_case(self):
        # https://github.com/inducer/relate/pull/423#discussion_r162121467
        gc2015 = factories.GradeChangeFactory.create(**(self.gc(points=5)))

        session1 = factories.FlowSessionFactory.create(
            participation=self.student_participation,
            start_time=self.time-timedelta(days=17),
            completion_time=self.time-timedelta(days=14))

        self.time_increment()

        gc2016 = factories.GradeChangeFactory.create(
            **(self.gc(points=0, flow_session=session1, grade_time=self.time)))

        gc2017 = factories.GradeChangeFactory.create(**(self.gc(points=7)))

        session2 = factories.FlowSessionFactory.create(
            participation=self.student_participation,
            start_time=self.time-timedelta(days=17),
            completion_time=self.time-timedelta(days=15))

        self.time_increment()

        gc2018 = factories.GradeChangeFactory.create(
            **(self.gc(points=6, flow_session=session2)))

        assert models.GradingOpportunity.objects.count() == 1
        assert models.GradeChange.objects.count() == 4
        assert models.FlowSession.objects.count() == 2

        self.assertTrue(session2.completion_time < session1.completion_time)
        self.assertTrue(
            gc2015.grade_time < gc2016.grade_time < gc2017.grade_time
            < gc2018.grade_time)

        self.assertGradeChangeMachineReadableStateEqual(gc2017.percentage())

    # {{{ When two grade changes have the same grade_time
    # The expected behavior is GradeChange object with the larger pk
    # dominate. Fixed with #263 and #417

    def test_gcs_have_same_grade_time1(self):
        gc1 = factories.GradeChangeFactory.create(**(self.gc(points=0)))
        session = factories.FlowSessionFactory.create(
            participation=self.student_participation,
            completion_time=gc1.grade_time-timedelta(days=1))
        factories.GradeChangeFactory.create(
            **(self.gc(points=5, flow_session=session,
                       grade_time=gc1.grade_time)))
        self.assertGradeChangeMachineReadableStateEqual(5)
        self.assertGradeChangeStateEqual("5.0% (/2)")

    def test_gc_have_same_grade_time2(self):
        session = factories.FlowSessionFactory.create(
            participation=self.student_participation,
            start_time=self.time-timedelta(days=1),
            completion_time=self.time)
        self.time_increment()
        gc1 = factories.GradeChangeFactory.create(
            **(self.gc(points=5, flow_session=session)))
        factories.GradeChangeFactory.create(
            **(self.gc(points=0, grade_time=gc1.grade_time)))
        self.assertGradeChangeMachineReadableStateEqual(0)
        self.assertGradeChangeStateEqual("0.00% (/2)")
    # }}}

    # {{{ Fix #430

    def test_reopen_session2(self):
        self.use_default_setup()

        # original state
        self.assertGradeChangeMachineReadableStateEqual("6")

        n_gc = models.GradeChange.objects.count()
        self.reopen_session2()

        # A new GradeChange object is created, with state "do_over"
        expected_n_gc = models.GradeChange.objects.count()
        self.assertEqual(expected_n_gc, n_gc + 1)
        self.assertEqual(
            models.GradeChange.objects.order_by("grade_time").last().state,
            g_state.do_over)

        self.assertGradeChangeMachineReadableStateEqual("NONE")
        self.assertGradeChangeStateEqual("- ∅ - (/3)")

    def test_reopen_session_without_existing_gc(self):
        # This is rare, because a completed_session should had created
        # a GradeChange object.
        session_temp = factories.FlowSessionFactory.create(
            participation=self.student_participation, completion_time=self.time)

        existing_gc_count = models.GradeChange.objects.count()
        reopen_session(now_datetime=local_now(), session=session_temp,
                       generate_grade_change=True,
                       suppress_log=True)
        self.assertEqual(models.GradeChange.objects.count(), existing_gc_count)

    def test_reopen_session1(self):
        self.use_default_setup()
        self.assertGradeChangeMachineReadableStateEqual("6")

        n_gc = models.GradeChange.objects.count()
        self.reopen_session1()

        # A new GradeChange object is created, with state "do_over"
        expected_n_gc = models.GradeChange.objects.count()
        self.assertEqual(expected_n_gc, n_gc + 1)
        self.assertEqual(
            models.GradeChange.objects.order_by("grade_time").last().state,
            g_state.do_over)

        # session 1 is not the latest session
        self.assertGradeChangeMachineReadableStateEqual("6")
        self.assertGradeChangeStateEqual("6.00% (/3)")

    def _get_admin_flow_session_delete_url(self, args):
        return reverse("admin:course_flowsession_delete", args=args)

    def _delete_flow_session_admin(self, flow_session):
        exist_flow_session_count = models.FlowSession.objects.count()
        flow_session_delete_url = self._get_admin_flow_session_delete_url(
            args=(flow_session.id,))
        delete_dict = {'post': 'yes'}
        with self.temporarily_switch_to_user(self.superuser):
            resp = self.c.get(flow_session_delete_url)
            self.assertEqual(resp.status_code, 200)
            resp = self.c.post(flow_session_delete_url, data=delete_dict)
        self.assertEqual(resp.status_code, 302)
        self.assertEqual(exist_flow_session_count,
                         models.FlowSession.objects.count() + 1)

    def test_delete_flow_session_admin_new_exempt_gradechange_created(self):
        self.use_default_setup()
        exist_grade_change_count = models.GradeChange.objects.count()

        # session1 has related grade changes, so a new grade change with 'exempt' is
        # created
        self._delete_flow_session_admin(self.session1)

        self.assertEqual(exist_grade_change_count + 1,
                         models.GradeChange.objects.count())

        last_gchange = (
            models.GradeChange.objects
            .order_by("-grade_time").first())
        self.assertIsNone(last_gchange.flow_session)
        self.assertEqual(last_gchange.state, g_state.exempt)

    def test_delete_flow_session_admin_no_new_gradechange_created(self):
        session_temp = factories.FlowSessionFactory.create(
            participation=self.student_participation, completion_time=self.time)

        exist_grade_change_count = models.GradeChange.objects.count()
        last_gchange_of_session_temp = (
            models.GradeChange.objects
            .filter(flow_session=session_temp)
            .order_by("-grade_time")[:1])
        self.assertEqual(last_gchange_of_session_temp.count(), 0)

        # session_temp has no related grade changes, so no new grade change
        # is created after deleted
        self._delete_flow_session_admin(session_temp)

        self.assertEqual(exist_grade_change_count,
                         models.GradeChange.objects.count())

    # }}}

    def test_backward_compatibility_merging_466(self):
        # this make sure after merging https://github.com/inducer/relate/pull/466
        # gchanges are consumed without issue
        self.use_default_setup()
        self.gc_session2.effective_time = None
        self.gc_session2.save()
        self.gc_session2.refresh_from_db()

        # We are not using reopen_session(), because that will create new
        # gchange, which only happen after #466 was merged.
        self.session2.in_progress = True
        self.session2.save()
        self.session2.refresh_from_db()

        machine = self.get_gc_machine()

        # session2's gchange is excluded
        self.assertGradeChangeMachineReadableStateEqual(7)
        self.assertEqual(machine.valid_percentages, [0, 7])
        self.assertGradeChangeStateEqual("7.00% (/2)")

    # {{{ test new gchange created when finishing flow

    def test_new_gchange_created_when_finish_flow_use_last_no_activity(self):
        # With use_last_activity_as_completion_time = True, if a flow session has
        # no last_activity, the expected effective_time of the new gchange should
        # be the completion time of the related flow_session.
        with self.temporarily_switch_to_user(self.student_participation.user):
            self.start_flow(QUIZ_FLOW_ID)
            self.assertEqual(models.GradeChange.objects.count(), 0)

            with mock.patch("course.flow.get_session_grading_rule") as \
                    mock_get_grading_rule:
                mock_get_grading_rule.side_effect = (
                    get_session_grading_rule_use_last_activity_as_cmplt_time_side_effect)  # noqa
                resp = self.end_flow()
                self.assertEqual(resp.status_code, 200)

            self.assertEqual(models.GradeChange.objects.count(), 1)
            latest_gchange = models.GradeChange.objects.last()
            latest_flow_session = models.FlowSession.objects.last()
            self.assertIsNone(latest_flow_session.last_activity())
            self.assertEqual(latest_gchange.effective_time,
                             latest_flow_session.completion_time)

    def test_new_gchange_created_when_finish_flow_not_use_last_no_activity(self):
        # With use_last_activity_as_completion_time = False, if a flow session has
        # no last_activity, the expected effective_time of the new gchange should
        # be the completion time of the related flow_session.
        with self.temporarily_switch_to_user(self.student_participation.user):
            self.start_flow(QUIZ_FLOW_ID)
            self.assertEqual(models.GradeChange.objects.count(), 0)

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

            self.assertEqual(models.GradeChange.objects.count(), 1)
            latest_gchange = models.GradeChange.objects.last()
            latest_flow_session = models.FlowSession.objects.last()
            self.assertIsNone(latest_flow_session.last_activity())
            self.assertEqual(latest_gchange.effective_time,
                             latest_flow_session.completion_time)

    def test_new_gchange_created_when_finish_flow_not_use_last_has_activity(self):
        # With use_last_activity_as_completion_time = False, even if a flow session
        # HAS last_activity, the expected effective_time of the new gchange should
        # be the completion_time of the related flow_session.
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            self.start_flow(QUIZ_FLOW_ID)

            # create a flow page visit, then there should be last_activity() for
            # the session.
            self.post_answer_by_ordinal(1, {"answer": ['0.5']})
            self.assertEqual(
                models.FlowPageVisit.objects.filter(answer__isnull=False).count(),
                1)
            last_answered_visit = (
                models.FlowPageVisit.objects.filter(answer__isnull=False).first())
            last_answered_visit.visit_time = now() - timedelta(hours=1)
            last_answered_visit.save()
            self.assertEqual(models.GradeChange.objects.count(), 0)

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

            self.assertEqual(models.GradeChange.objects.count(), 1)
            latest_gchange = models.GradeChange.objects.last()
            latest_flow_session = models.FlowSession.objects.last()
            self.assertIsNotNone(latest_flow_session.last_activity())
            self.assertNotEqual(latest_flow_session.completion_time,
                             latest_flow_session.last_activity())
            self.assertEqual(latest_gchange.effective_time,
                             latest_flow_session.completion_time)

    # }}}

# vim: fdm=marker