test_grades.py

        self.assertGradeChangeStateEqual("6.0% (/3)")

        # make sure participations with pperm.included_in_grade_statistics
        # are not included
        factories.GradeChangeFactory.create(**(self.gc(
            participation=self.instructor_participation, points=2)))
        factories.GradeChangeFactory.create(**(self.gc(
            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):
        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)


class EditGradingOpportunityTest(GradesTestMixin, TestCase):
    # test grades.edit_grading_opportunity

    def get_edit_grading_opportunity_url(self, opp_id, course_identifier=None):
        course_identifier = course_identifier or self.get_default_course_identifier()
        kwargs = {"course_identifier": course_identifier,
                  "opportunity_id": opp_id}
        return reverse("relate-edit_grading_opportunity", kwargs=kwargs)

    def get_edit_grading_opportunity_view(self, opp_id, course_identifier=None,
                                          force_login_instructor=True):
        course_identifier = course_identifier or self.get_default_course_identifier()
        if not force_login_instructor:
            user = self.get_logged_in_user()
        else:
            user = self.instructor_participation.user

        with self.temporarily_switch_to_user(user):
            return self.c.get(
                self.get_edit_grading_opportunity_url(opp_id, course_identifier))

    def post_edit_grading_opportunity_view(self, opp_id, data,
                                           course_identifier=None,
                                           force_login_instructor=True):
        course_identifier = course_identifier or self.get_default_course_identifier()
        if not force_login_instructor:
            user = self.get_logged_in_user()
        else:
            user = self.instructor_participation.user

        with self.temporarily_switch_to_user(user):
            return self.c.post(
                self.get_edit_grading_opportunity_url(opp_id, course_identifier),
                data)

    def edit_grading_opportunity_post_data(
            self, name, identifier, page_scores_in_participant_gradebook=False,
            hide_superseded_grade_history_before=None,
            op="sumbit", shown_in_participant_grade_book=True,
            aggregation_strategy=constants.grade_aggregation_strategy.use_latest,
            shown_in_grade_book=True, result_shown_in_participant_grade_book=True,
            **kwargs):

        data = {"name": name,
                "identifier": identifier,
                op: '',
                "aggregation_strategy": aggregation_strategy}

        if page_scores_in_participant_gradebook:
            data["page_scores_in_participant_gradebook"] = ''

        if hide_superseded_grade_history_before:
            if isinstance(hide_superseded_grade_history_before, datetime.datetime):
                date_time_picker_time_format = "%Y-%m-%d %H:%M"
                hide_superseded_grade_history_before = (
                    hide_superseded_grade_history_before.strftime(
                        date_time_picker_time_format))
            data["hide_superseded_grade_history_before"] = (
                hide_superseded_grade_history_before)
        if shown_in_participant_grade_book:
            data["shown_in_participant_grade_book"] = ''
        if shown_in_grade_book:
            data["shown_in_grade_book"] = ''
        if result_shown_in_participant_grade_book:
            data["result_shown_in_participant_grade_book"] = ''

        data.update(kwargs)
        return data

    def test_get_add_new(self):
        resp = self.get_edit_grading_opportunity_view(-1)
        self.assertEqual(resp.status_code, 200)

    def test_post_get_add_new(self):
        name = "my Gopp"
        identifier = "my_gopp"
        data = self.edit_grading_opportunity_post_data(
            name=name, identifier=identifier)
        resp = self.post_edit_grading_opportunity_view(-1, data=data)
        gopps = models.GradingOpportunity.objects.all()
        self.assertEqual(gopps.count(), 2)
        my_gopp = gopps.last()
        self.assertEqual(my_gopp.name, name)
        self.assertEqual(my_gopp.identifier, identifier)
        self.assertRedirects(
            resp, self.get_edit_grading_opportunity_url(my_gopp.pk),
            fetch_redirect_response=False)

    def test_course_not_match(self):
        another_course = factories.CourseFactory(identifier="another-course")
        another_course_gopp = factories.GradingOpportunityFactory(
            course=another_course)
        gopps = models.GradingOpportunity.objects.all()
        self.assertEqual(gopps.count(), 2)

        resp = self.get_edit_grading_opportunity_view(
            another_course_gopp.id, course_identifier=self.course.identifier)
        self.assertEqual(resp.status_code, 400)

    def test_view_edit_grading_opportunity(self):
        my_gopp = factories.GradingOpportunityFactory(
            course=self.course, identifier="another_gopp")

        data = self.edit_grading_opportunity_post_data(
            name=my_gopp.name, identifier=my_gopp.identifier,
            shown_in_grade_book=False)

        resp = self.post_edit_grading_opportunity_view(my_gopp.id, data=data)

        self.assertRedirects(
            resp, self.get_edit_grading_opportunity_url(my_gopp.pk),
            fetch_redirect_response=False)

        my_gopp.refresh_from_db()
        self.assertEqual(my_gopp.shown_in_grade_book, False)

    def test_view_edit_grading_opportunity_form_invalid(self):
        my_gopp = factories.GradingOpportunityFactory(
            course=self.course, identifier="another_gopp")

        data = self.edit_grading_opportunity_post_data(
            name=my_gopp.name, identifier=my_gopp.identifier,
            shown_in_grade_book=False)
        with mock.patch(
                "course.grades.EditGradingOpportunityForm.is_valid"
        ) as mock_form_is_valid:
            mock_form_is_valid.return_value = False

            resp = self.post_edit_grading_opportunity_view(my_gopp.id, data=data)
            self.assertEqual(resp.status_code, 200)

        my_gopp.refresh_from_db()
        self.assertEqual(my_gopp.shown_in_grade_book, True)


class DownloadAllSubmissionsTest(SingleCourseQuizPageTestMixin,
                                 HackRepoMixin, TestCase):
    # test grades.download_all_submissions (for cases not covered by other tests)

    page_id = "half"
    my_access_rule_tag = "my_access_rule_tag"

    @classmethod
    def setUpTestData(cls):  # noqa
        super(DownloadAllSubmissionsTest, cls).setUpTestData()

        # with this faked commit_sha, we may do multiple submissions
        cls.course.active_git_commit_sha = (
            "my_fake_commit_sha_for_download_submissions")
        cls.course.save()
        with cls.temporarily_switch_to_user(cls.student_participation.user):
            cls.start_flow(cls.flow_id)
            cls.submit_page_answer_by_page_id_and_test(
                cls.page_id, answer_data={"answer": 0.25})
            cls.end_flow()

            fs = models.FlowSession.objects.first()
            fs.access_rules_tag = cls.my_access_rule_tag
            fs.save()

            cls.start_flow(cls.flow_id)
            cls.submit_page_answer_by_page_id_and_test("proof")
            cls.submit_page_answer_by_page_id_and_test(cls.page_id)
            cls.end_flow()

        # create an in_progress flow, with the same page submitted
        another_particpation = factories.ParticipationFactory(
            course=cls.course)
        with cls.temporarily_switch_to_user(another_particpation.user):
            cls.start_flow(cls.flow_id)
            cls.submit_page_answer_by_page_id_and_test(cls.page_id)

            # create a flow with no answers
            cls.start_flow(cls.flow_id)
            cls.end_flow()

    @property
    def group_page_id(self):
        _, group_id = self.get_page_ordinal_via_page_id(
            self.page_id, with_group_id=True)
        return "%s/%s" % (group_id, self.page_id)

    def get_zip_file_buf_from_response(self, resp):
        return six.BytesIO(resp.content)

    def assertDownloadedFileZippedExtensionCount(self, resp, extensions, counts):  # noqa

        assert isinstance(extensions, list)
        assert isinstance(counts, list)
        assert len(extensions) == len(counts)
        prefix, zip_file = resp["Content-Disposition"].split('=')
        self.assertEqual(prefix, "attachment; filename")
        self.assertEqual(resp.get('Content-Type'), "application/zip")
        buf = six.BytesIO(resp.content)
        import zipfile
        with zipfile.ZipFile(buf, 'r') as zf:
            self.assertIsNone(zf.testzip())

            for f in zf.filelist:
                self.assertTrue(f.file_size > 0)

            for i, ext in enumerate(extensions):
                self.assertEqual(
                    len([f for f in zf.filelist if
                         f.filename.endswith(ext)]), counts[i])

    def test_no_rules_tag(self):
        hacked_flow_desc = self.get_hacked_flow_desc(del_rules=True)
        with mock.patch("course.content.get_flow_desc") as mock_get_flow_desc:
            mock_get_flow_desc.return_value = hacked_flow_desc

            with self.temporarily_switch_to_user(self.instructor_participation.user):
                resp = self.post_download_all_submissions_by_group_page_id(
                    group_page_id=self.group_page_id, flow_id=self.flow_id)
                self.assertEqual(resp.status_code, 200)
                self.assertDownloadedFileZippedExtensionCount(
                    resp, [".txt"], [1])

    def test_download_first_attempt(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.post_download_all_submissions_by_group_page_id(
                group_page_id=self.group_page_id, flow_id=self.flow_id,
                which_attempt="first")

            self.assertEqual(resp.status_code, 200)
            self.assertDownloadedFileZippedExtensionCount(
                resp, [".txt"], [1])

    def test_download_all_attempts(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.post_download_all_submissions_by_group_page_id(
                group_page_id=self.group_page_id, flow_id=self.flow_id,
                which_attempt="all")

            self.assertEqual(resp.status_code, 200)
            self.assertDownloadedFileZippedExtensionCount(
                resp, [".txt"], [2])

    # Fixme
    @unittest.skipIf(six.PY2, "'utf8' codec can't decode byte 0x99 in "
                              "position 10: invalid start byte")
    def test_download_include_feedback(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.post_download_all_submissions_by_group_page_id(
                group_page_id=self.group_page_id, flow_id=self.flow_id,
                include_feedback=True)

            self.assertEqual(resp.status_code, 200)
            self.assertDownloadedFileZippedExtensionCount(
                resp, [".txt"], [2])

    def test_download_include_feedback_no_feedback(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            another_group_page_id = (
                self.group_page_id.replace(self.page_id, "proof"))
            resp = self.post_download_all_submissions_by_group_page_id(
                group_page_id=another_group_page_id, flow_id=self.flow_id,
                include_feedback=True)

            self.assertEqual(resp.status_code, 200)
            self.assertDownloadedFileZippedExtensionCount(
                resp, [".pdf"], [1])

    def test_download_include_extra_file(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            import os
            with open(
                    os.path.join(os.path.dirname(__file__),
                                 '../resource',
                                 'test_file.pdf'), 'rb') as extra_file:
                resp = self.post_download_all_submissions_by_group_page_id(
                    group_page_id=self.group_page_id, flow_id=self.flow_id,
                    extra_file=extra_file)

            self.assertEqual(resp.status_code, 200)
            self.assertDownloadedFileZippedExtensionCount(
                resp, [".txt", ".pdf"], [1, 1])

    def test_download_in_progress(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.post_download_all_submissions_by_group_page_id(
                group_page_id=self.group_page_id, flow_id=self.flow_id,
                non_in_progress_only=False)

            self.assertEqual(resp.status_code, 200)
            self.assertDownloadedFileZippedExtensionCount(
                resp, [".txt"], [2])

    def test_download_other_access_rule_tags(self):
        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"] = [self.my_access_rule_tag, "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 == [self.my_access_rule_tag, "blahblah"]

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

            with self.temporarily_switch_to_user(self.instructor_participation.user):
                resp = self.post_download_all_submissions_by_group_page_id(
                    group_page_id=self.group_page_id, flow_id=self.flow_id,
                    restrict_to_rules_tag=self.my_access_rule_tag)

                self.assertEqual(resp.status_code, 200)

                self.assertDownloadedFileZippedExtensionCount(
                    resp, [".txt"], [1])


class PointsEqualTest(unittest.TestCase):
    # grades.points_equal
    def test(self):
        from decimal import Decimal
        self.assertTrue(grades.points_equal(None, None))
        self.assertFalse(grades.points_equal(Decimal(1.11), None))
        self.assertFalse(grades.points_equal(None, Decimal(1.11)))
        self.assertTrue(grades.points_equal(Decimal(1.11), Decimal(1.11)))
        self.assertFalse(grades.points_equal(Decimal(1.11), Decimal(1.12)))


@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()))