__copyright__ = "Copyright (C) 2018 Dong Zhuang, Zesheng Wang, Andreas Kloeckner"

__license__ = """
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""

import datetime
import io
import unittest

import pytest
from django.test import Client, TestCase
from django.urls import reverse
from django.utils.timezone import now, timedelta

from course import constants, grades, models
from course.constants import (
    grade_aggregation_strategy as g_stragety, grade_state_change_types as g_state,
    participation_permission as pperm,
)
from course.flow import reopen_session
from course.grades import (
    get_single_grade_changes_and_state_machine as get_gc_and_machine,
)
from relate.utils import local_now
from tests import factories
from tests.base_test_mixins import (
    HackRepoMixin, MockAddMessageMixing, SingleCoursePageTestMixin,
    SingleCourseQuizPageTestMixin,
)
from tests.constants import QUIZ_FLOW_ID
from tests.utils import mock


def get_session_grading_rule_use_last_activity_as_cmplt_time_side_effect(
        session, flow_desc, now_datetime):
    # The testing flow "quiz-test" didn't set the attribute
    from course.utils import get_session_grading_rule
    actual_grading_rule = get_session_grading_rule(session, flow_desc, now_datetime)
    actual_grading_rule.use_last_activity_as_completion_time = True
    return actual_grading_rule


class GradesTestMixin(SingleCoursePageTestMixin, MockAddMessageMixing):
    time = now() - timedelta(days=10)

    @classmethod
    def setUpTestData(cls):  # noqa
        super().setUpTestData()
        cls.gopp = factories.GradingOpportunityFactory(
            course=cls.course, aggregation_strategy=g_stragety.use_latest)

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

    def use_default_setup(self):  # noqa
        self.session1 = factories.FlowSessionFactory.create(
            participation=self.student_participation, completion_time=self.time)
        self.time_increment()
        self.gc_main_1 = factories.GradeChangeFactory.create(**(self.gc(points=5)))
        self.gc_session1 = factories.GradeChangeFactory.create(**(self.gc(points=0,
                                                       flow_session=self.session1)))

        self.session2 = factories.FlowSessionFactory.create(
            participation=self.student_participation, completion_time=self.time)
        self.gc_main_2 = factories.GradeChangeFactory.create(**(self.gc(points=7)))
        self.gc_session2 = factories.GradeChangeFactory.create(**(self.gc(points=6,
                                                       flow_session=self.session2)))
        assert models.GradingOpportunity.objects.count() == 1
        assert models.GradeChange.objects.count() == 4
        assert models.FlowSession.objects.count() == 2

    def time_increment(self, minute_delta=10):
        self.time += timedelta(minutes=minute_delta)

    @classmethod
    def gc(cls, opportunity=None, state=None, attempt_id=None, points=None,
           max_points=None, comment=None, due_time=None,
           grade_time=None, flow_session=None, null_attempt_id=False, **kwargs):

        if attempt_id is None:
            if flow_session is None:
                if not null_attempt_id:
                    attempt_id = "main"
            else:
                from course.flow import get_flow_session_attempt_id
                attempt_id = get_flow_session_attempt_id(flow_session)
        gc_kwargs = {
            "opportunity": opportunity or cls.gopp,
            "participation": cls.student_participation,
            "state": state or g_state.graded,
            "attempt_id": attempt_id,
            "points": points,
            "max_points": max_points or 100,
            "comment": comment,
            "due_time": due_time,
            "grade_time": grade_time or cls.time,
            "flow_session": flow_session,
        }
        cls.time += timedelta(minutes=10)
        gc_kwargs.update(kwargs)
        return gc_kwargs

    def get_gc_machine(self, gopp=None, participation=None):
        if not gopp:
            gopp = self.gopp
        if not participation:
            participation = self.student_participation
        _, machine = get_gc_and_machine(gopp, participation)
        return machine

    def get_gc_stringify_machine_readable_state(self):
        machine = self.get_gc_machine()
        return machine.stringify_machine_readable_state()

    def get_gc_stringify_state(self):
        machine = self.get_gc_machine()
        return machine.stringify_state()

    def update_gopp_strategy(self, strategy=None):
        if not strategy:
            return
        else:
            self.gopp.aggregation_strategy = strategy
            self.gopp.save()
            self.gopp.refresh_from_db()

    def assertGradeChangeStateEqual(self, expected_state_string=None):  # noqa
        # targeting stringify_state
        state_string = self.get_gc_stringify_state()

        from django.utils.encoding import force_str
        self.assertEqual(force_str(state_string), expected_state_string)

    def assertGradeChangeMachineReadableStateEqual(self, expected_state_string=None):  # noqa
        # targeting stringify_machine_readable_state
        state_string = self.get_gc_stringify_machine_readable_state()
        from decimal import Decimal, InvalidOperation
        try:
            percentage = Decimal(state_string, )
        except InvalidOperation:
            percentage = None

        try:
            expected_percentage = Decimal(expected_state_string)
        except InvalidOperation:
            expected_percentage = None

        not_equal_msg = (
                "%s does not have equal value with '%s'"
                % (state_string, str(expected_percentage))
        )

        if percentage is not None and expected_percentage is not None:
            self.assertTrue(
                abs(percentage - expected_percentage) < 1e-4, msg=not_equal_msg)
        else:
            if type(percentage) is not type(expected_percentage):
                self.fail(not_equal_msg)

        if percentage is None and expected_percentage is None:
            self.assertEqual(state_string, expected_state_string)

    def append_gc(self, gc):
        return factories.GradeChangeFactory.create(**gc)

    def update_gc(self, gc_object, update_time=True, **kwargs):
        # This is alter GradeChange objects via db.

        gc_dict = gc_object.__dict__
        gc_dict.update(**kwargs)
        if update_time:
            gc_dict["grade_time"] = now()
        gc_object.save()
        gc_object.refresh_from_db()


class ViewParticipantGradesTest(GradesTestMixin, TestCase):
    # test grades.view_participant_grades
    def test_pctx_no_participation(self):
        with self.temporarily_switch_to_user(None):
            resp = self.get_my_grades_view()
            self.assertEqual(resp.status_code, 403)

    def test_view_others_grades_no_perm(self):
        other_participation = factories.ParticipationFactory(course=self.course)
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_participant_grades(other_participation.pk)
            self.assertEqual(resp.status_code, 403)

    def test_view_others_grades_no_pctx(self):
        other_participation = factories.ParticipationFactory(course=self.course)
        with self.temporarily_switch_to_user(None):
            resp = self.get_view_participant_grades(other_participation.pk)
            self.assertEqual(resp.status_code, 403)

    def test_view_my_participation_grades(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_view_participant_grades(self.student_participation.pk)
            self.assertEqual(resp.status_code, 200)

    def test_view_my_grades(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_my_grades_view()
            self.assertEqual(resp.status_code, 200)

    def test_ta_view_student_grades(self):
        with self.temporarily_switch_to_user(self.ta_participation.user):
            resp = self.get_view_participant_grades(self.student_participation.pk)
            self.assertEqual(resp.status_code, 200)

    def test_view(self):

        # {{{ gopps. Notice: there is another gopp created in setUp
        # shown for all (gopp with no gchanges)

        # shown for all
        shown_gopp = factories.GradingOpportunityFactory(
            course=self.course, identifier="1shown", name="SHOWN")

        # hidden for all view
        hidden_gopp_all = factories.GradingOpportunityFactory(
            course=self.course, identifier="2hidden_all", name="HIDDEN_ALL",
            shown_in_grade_book=False)

        # hidden if is_privileged_view is True
        hidden_gopp = factories.GradingOpportunityFactory(
            course=self.course, identifier="3hidden", name="HIDDEN",
            shown_in_participant_grade_book=False)

        # result hidden
        shown_gopp_result_hidden = factories.GradingOpportunityFactory(
            course=self.course, identifier="4shown_result_hidden",
            name="SHOWN_RESULT_HIDDEN",
            result_shown_in_participant_grade_book=False)

        # }}}

        # {{{ gchanges
        # this will be consumed
        gchange_shown1 = factories.GradeChangeFactory(
            **self.gc(
                opportunity=shown_gopp,
                attempt_id="main", points=0, max_points=10))

        # this won't be consumed
        gchange_hidden_all = factories.GradeChangeFactory(
            **self.gc(
                opportunity=hidden_gopp_all, attempt_id="hidden_all",
                points=1, max_points=5))

        # this will be consumed
        gchange_result_hidden = factories.GradeChangeFactory(
            **self.gc(
                opportunity=shown_gopp_result_hidden,
                attempt_id="shown_result_hidden", points=15, max_points=15))

        # this will be consumed only when is_privileged_view is True
        gchange_hidden = factories.GradeChangeFactory(
            **self.gc(
                opportunity=hidden_gopp, attempt_id="hidden", points=2,
                max_points=5))

        # this will be consumed
        gchange_shown2 = factories.GradeChangeFactory(
            **self.gc(
                opportunity=shown_gopp,
                attempt_id="main", points=10, max_points=10))

        # this will be consumed
        gchange_shown3 = factories.GradeChangeFactory(
            **self.gc(
                opportunity=shown_gopp,
                attempt_id="main", points=6, max_points=10))
        # }}}

        user = self.student_participation.user
        with self.temporarily_switch_to_user(user):
            with self.subTest(user=user):
                with mock.patch(
                        "course.models.GradeStateMachine.consume") as mock_consume:
                    resp = self.get_my_grades_view()
                    self.assertEqual(resp.status_code, 200)

                    # testing call of GradeStateMachine consume
                    expected_called = [
                        [gchange_shown1, gchange_shown2, gchange_shown3],
                        [gchange_result_hidden]]

                    # no expected to be consumed
                    not_expected_called = [
                        [gchange_hidden_all], [gchange_hidden]]

                    actually_called = []
                    for call in mock_consume.call_args_list:
                        arg, _ = call
                        for not_expected in not_expected_called:
                            self.assertNotIn(not_expected, arg)
                        if len(arg[0]):
                            actually_called.append(arg[0])
                    self.assertListEqual(actually_called, expected_called)

                # non mock call
                resp = self.get_my_grades_view()
                self.assertEqual(resp.status_code, 200)
                self.assertEqual(
                    len(resp.context["grading_opportunities"]), 3)
                self.assertEqual(len(resp.context["grade_table"]), 3)
                self.assertFalse(resp.context["is_privileged_view"])
                self.assertContains(resp, "60.0%", count=1)  # for shown_gopp
                self.assertNotContains(resp, "40.0%")  # for hidden_gopp
                self.assertContains(resp, "(not released)", count=2)  # for hidden_gopp  # noqa

        user = self.ta_participation.user
        with self.temporarily_switch_to_user(user):
            with self.subTest(user=user):
                with mock.patch(
                        "course.models.GradeStateMachine.consume") as mock_consume:
                    resp = self.get_view_participant_grades(
                        self.student_participation.pk)
                    self.assertEqual(resp.status_code, 200)

                    # testing call of GradeStateMachine consume
                    expected_called = [
                        [gchange_shown1, gchange_shown2, gchange_shown3],
                        [gchange_hidden],
                        [gchange_result_hidden]]

                    # no expected to be consumed
                    not_expected_called = [
                        [gchange_hidden_all]]

                    actually_called = []
                    for call in mock_consume.call_args_list:
                        arg, _ = call
                        for not_expected in not_expected_called:
                            self.assertNotIn(not_expected, arg)
                        if len(arg[0]):
                            actually_called.append(arg[0])

                    self.assertListEqual(actually_called, expected_called)

                # non mock call
                resp = self.get_view_participant_grades(
                    self.student_participation.pk)
                self.assertEqual(
                    len(resp.context["grading_opportunities"]), 4)
                self.assertEqual(len(resp.context["grade_table"]), 4)
                self.assertTrue(resp.context["is_privileged_view"])

                self.assertContains(resp, "60.0%", count=1)  # for shown_gopp
                self.assertContains(resp, "40.0%", count=1)  # for hidden_gopp


@pytest.mark.slow
class GetGradeTableTest(GradesTestMixin, TestCase):
    # test grades.get_grade_table

    @classmethod
    def setUpTestData(cls):  # noqa
        super().setUpTestData()
        # 2 more participations
        (cls.ptpt1, cls.ptpt2) = factories.ParticipationFactory.create_batch(
            size=2, course=cls.course)

        # this make sure it filtered by participation status active
        factories.ParticipationFactory(
            course=cls.course, status=constants.participation_status.dropped)

        # another course and a gopp, this make sure it filtered by course
        another_course = factories.CourseFactory(identifier="another-course")
        factories.GradingOpportunityFactory(course=another_course)

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

        # {{{ gopps. Notice: there is another gopp created in setUp
        # shown for all (gopp with no gchanges)

        # shown for all
        shown_gopp = factories.GradingOpportunityFactory(
            course=self.course, identifier="1shown", name="SHOWN")

        # hidden for all view
        hidden_gopp_all = factories.GradingOpportunityFactory(
            course=self.course, identifier="2hidden_all", name="HIDDEN_ALL",
            shown_in_grade_book=False)

        # hidden if is_privileged_view is True
        hidden_gopp = factories.GradingOpportunityFactory(
            course=self.course, identifier="3hidden", name="HIDDEN",
            shown_in_participant_grade_book=False)

        # }}}

        # {{{ gchanges for stu
        stu_shown_gopp_gchanges = [
            self.gc(
                participation=self.student_participation,
                opportunity=shown_gopp,
                attempt_id="main", points=0, max_points=10),
            self.gc(
                participation=self.student_participation,
                opportunity=shown_gopp,
                attempt_id="main", points=2, max_points=10),
            self.gc(
                participation=self.student_participation,
                opportunity=shown_gopp,
                attempt_id="main", points=5, max_points=10),
            self.gc(
                participation=self.student_participation,
                opportunity=shown_gopp,
                attempt_id="main", points=4, max_points=10),
        ]  # expecting 40%

        stu_hidden_gopp_gchanges = [
            self.gc(
                participation=self.student_participation,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=5, max_points=10),
            self.gc(
                participation=self.student_participation,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=4, max_points=10),
            self.gc(
                participation=self.student_participation,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=3, max_points=10),
            self.gc(
                participation=self.student_participation,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=2, max_points=10),
        ]  # expecting 20%

        stu_hidden_all_gopp_gchanges = [
            self.gc(
                participation=self.student_participation,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=5, max_points=10),
            self.gc(
                participation=self.student_participation,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=4, max_points=10),
            self.gc(
                participation=self.student_participation,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=3, max_points=10),
            self.gc(
                participation=self.student_participation,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=2, max_points=10),
        ]  # no result expected

        # }}}

        # {{{ gchanges for ptpt1
        ptpt1shown_gopp_gchanges = [
            self.gc(
                participation=self.ptpt1,
                opportunity=shown_gopp,
                attempt_id="main", points=1, max_points=10),
            self.gc(
                participation=self.ptpt1,
                opportunity=shown_gopp,
                attempt_id="main", points=3, max_points=10),
            self.gc(
                participation=self.ptpt1,
                opportunity=shown_gopp,
                attempt_id="main", points=6, max_points=10),
            self.gc(
                participation=self.ptpt1,
                opportunity=shown_gopp,
                attempt_id="main", points=9, max_points=10),
        ]  # expecting 90%

        ptpt1_hidden_gopp_gchanges = [
            self.gc(
                participation=self.ptpt1,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=10, max_points=10),
            self.gc(
                participation=self.ptpt1,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=9, max_points=10),
            self.gc(
                participation=self.ptpt1,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=8, max_points=10),
            self.gc(
                participation=self.ptpt1,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=7, max_points=10),
        ]  # expecting 70%

        ptpt1_hidden_all_gopp_gchanges = [
            self.gc(
                participation=self.ptpt1,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=3, max_points=10),
            self.gc(
                participation=self.ptpt1,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=2, max_points=10),
            self.gc(
                participation=self.ptpt1,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=1, max_points=10),
            self.gc(
                participation=self.ptpt1,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=1.5, max_points=10),
        ]  # no result expected

        # {{{ gchanges for ptpt2
        ptpt2_shown_gopp_gchanges = [
            self.gc(
                participation=self.ptpt2,
                opportunity=shown_gopp,
                attempt_id="main", points=10, max_points=10),
            self.gc(
                participation=self.ptpt2,
                opportunity=shown_gopp,
                attempt_id="main", points=1, max_points=10),
            self.gc(
                participation=self.ptpt2,
                opportunity=shown_gopp,
                attempt_id="main", points=2, max_points=10),
            self.gc(
                participation=self.ptpt2,
                opportunity=shown_gopp,
                attempt_id="main", points=3.5, max_points=10),
        ]  # expecting 35%

        ptpt2_hidden_gopp_gchanges = [
            self.gc(
                participation=self.ptpt2,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=2, max_points=10),
            self.gc(
                participation=self.ptpt2,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=4, max_points=10),
            self.gc(
                participation=self.ptpt2,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=3, max_points=10),
            self.gc(
                participation=self.ptpt2,
                opportunity=hidden_gopp,
                attempt_id="hidden", points=6.5, max_points=10),
        ]  # expecting 65%

        ptpt2_hidden_all_gopp_gchanges = [
            self.gc(
                participation=self.ptpt2,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=5, max_points=10),
            self.gc(
                participation=self.ptpt2,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=4, max_points=10),
            self.gc(
                participation=self.ptpt2,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=3, max_points=10),
            self.gc(
                participation=self.ptpt2,
                opportunity=hidden_gopp_all,
                attempt_id="hidden_all", points=6, max_points=10),
        ]  # no result expected
        # }}}

        gchange_kwargs_lists = [
            stu_hidden_all_gopp_gchanges,
            stu_hidden_gopp_gchanges,
            stu_shown_gopp_gchanges,

            ptpt1_hidden_all_gopp_gchanges,
            ptpt1_hidden_gopp_gchanges,
            ptpt1shown_gopp_gchanges,

            ptpt2_hidden_all_gopp_gchanges,
            ptpt2_hidden_gopp_gchanges,
            ptpt2_shown_gopp_gchanges]

        from random import shuffle

        while True:
            gchange_kwargs_lists = [lst for lst in gchange_kwargs_lists if len(lst)]
            if not gchange_kwargs_lists:
                break

            shuffle(gchange_kwargs_lists)
            kwarg_list = gchange_kwargs_lists[0]
            gchange_kwargs = kwarg_list.pop(0)
            factories.GradeChangeFactory(**gchange_kwargs)

        participations, grading_opps, grade_table = (
            grades.get_grade_table(self.course))

        self.assertEqual(
            participations, [
                self.instructor_participation,
                self.ta_participation,
                self.student_participation,
                self.ptpt1,
                self.ptpt2])

        # hidden_gopp_all not shown
        # ordered by identifier
        self.assertListEqual(grading_opps, [shown_gopp, hidden_gopp, self.gopp])

        self.assertEqual(len(grade_table), 5)
        for i in range(5):
            self.assertEqual(len(grade_table[i]), 3)

        self.assertEqual(grade_table[2][0].grade_state_machine.percentage(), 40)
        self.assertEqual(grade_table[2][1].grade_state_machine.percentage(), 20)
        self.assertEqual(grade_table[2][2].grade_state_machine.percentage(), None)

        self.assertEqual(grade_table[3][0].grade_state_machine.percentage(), 90)
        self.assertEqual(grade_table[3][1].grade_state_machine.percentage(), 70)
        self.assertEqual(grade_table[3][2].grade_state_machine.percentage(), None)

        self.assertEqual(grade_table[4][0].grade_state_machine.percentage(), 35)
        self.assertEqual(grade_table[4][1].grade_state_machine.percentage(), 65)
        self.assertEqual(grade_table[4][2].grade_state_machine.percentage(), None)

    def test(self):
        for _i in range(10):
            self.run_test()
            factories.UserFactory.reset_sequence(0)
            self.setUp()


fake_access_rules_tag = "fake_tag"
fake_task_id = "abcdef123"


class MockAsyncRes:
    def __init__(self):
        self.id = fake_task_id


class ViewGradesByOpportunityTest(GradesTestMixin, TestCase):
    # test grades.view_grades_by_opportunity

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

        # create 2 flow sessions, one with access_rules_tag
        factories.FlowSessionFactory(
            participation=self.student_participation,
            flow_id=QUIZ_FLOW_ID, in_progress=False, page_count=15)

        factories.FlowSessionFactory(
            participation=self.student_participation,
            access_rules_tag=fake_access_rules_tag, page_count=15)

        fake_expire_in_progress_sessions = mock.patch(
            "course.tasks.expire_in_progress_sessions.delay",
            return_value=MockAsyncRes())
        self.mock_expire_in_progress_sessions = (
            fake_expire_in_progress_sessions.start())
        self.addCleanup(fake_expire_in_progress_sessions.stop)

        fake_finish_in_progress_sessions = mock.patch(
            "course.tasks.finish_in_progress_sessions.delay",
            return_value=MockAsyncRes())
        self.mock_finish_in_progress_sessions = (
            fake_finish_in_progress_sessions.start())
        self.addCleanup(fake_finish_in_progress_sessions.stop)

        fake_regrade_flow_sessions = mock.patch(
            "course.tasks.regrade_flow_sessions.delay",
            return_value=MockAsyncRes())
        self.mock_regrade_flow_sessions = fake_regrade_flow_sessions.start()
        self.addCleanup(fake_regrade_flow_sessions.stop)

        fake_recalculate_ended_sessions = mock.patch(
            "course.tasks.recalculate_ended_sessions.delay",
            return_value=MockAsyncRes())
        self.mock_recalculate_ended_sessions = (
            fake_recalculate_ended_sessions.start())
        self.addCleanup(fake_recalculate_ended_sessions.stop)

    gopp_id = "la_quiz"

    def test_no_permission(self):
        with self.temporarily_switch_to_user(None):
            resp = self.get_gradebook_by_opp_view(
                self.gopp_id, force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)
            resp = self.post_gradebook_by_opp_view(
                self.gopp_id, {}, force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)

        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_gradebook_by_opp_view(
                self.gopp_id, force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)
            resp = self.post_gradebook_by_opp_view(
                self.gopp_id, {}, force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)

    def test_gopp_does_not_exist(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.client.get(self.get_gradebook_url_by_opp_id("2"))
            self.assertEqual(resp.status_code, 404)

    def test_gopp_course_not_match(self):
        another_course = factories.CourseFactory(identifier="another-course")
        another_course_gopp = factories.GradingOpportunityFactory(
            course=another_course, identifier=self.gopp_id)

        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.client.get(self.get_gradebook_url_by_opp_id(
                another_course_gopp.id))
            self.assertEqual(resp.status_code, 400)

    def test_batch_op_no_permission(self):
        with self.temporarily_switch_to_user(self.ta_participation.user):
            for op in ["expire", "end", "regrade", "recalculate"]:
                with self.subTest(user=self.ta_participation.user, op=op):
                    resp = self.post_gradebook_by_opp_view(
                        self.gopp_id,
                        post_data={"rule_tag": grades.RULE_TAG_NONE_STRING,
                                   "past_due_only": True,
                                   op: ""},
                        force_login_instructor=False)

                    # because post is neglected for user without those pperms
                    self.assertEqual(resp.status_code, 200)
                    self.assertEqual(
                        self.mock_expire_in_progress_sessions.call_count, 0)
                    self.assertEqual(
                        self.mock_finish_in_progress_sessions.call_count, 0)
                    self.assertEqual(
                        self.mock_regrade_flow_sessions.call_count, 0)
                    self.assertEqual(
                        self.mock_recalculate_ended_sessions.call_count, 0)

    def test_batch_op_no_permission2(self):
        # with partitial permission
        permission_ops = [
            (pperm.batch_end_flow_session, "end"),
            (pperm.batch_impose_flow_session_deadline, "expire"),
            (pperm.batch_regrade_flow_session, "regrade"),
            (pperm.batch_recalculate_flow_session_grade, "recalculate")]

        from itertools import combinations
        comb = list(combinations(permission_ops, 2))
        comb += [reversed(c) for c in comb]

        with self.temporarily_switch_to_user(self.ta_participation.user):
            for po in comb:
                allowed, not_allowed = po
                pp = models.ParticipationPermission(
                    participation=self.ta_participation,
                    permission=allowed[0])
                pp.save()
                op = not_allowed[1]

                with self.subTest(user=self.ta_participation.user, op=op):
                    resp = self.post_gradebook_by_opp_view(
                        self.gopp_id,
                        post_data={"rule_tag": grades.RULE_TAG_NONE_STRING,
                                   "past_due_only": True,
                                   op: ""},
                        force_login_instructor=False)

                self.assertEqual(resp.status_code, 403)

                # revoke permission
                pp.delete()

        self.assertEqual(
            self.mock_expire_in_progress_sessions.call_count, 0)
        self.assertEqual(
            self.mock_finish_in_progress_sessions.call_count, 0)
        self.assertEqual(
            self.mock_regrade_flow_sessions.call_count, 0)
        self.assertEqual(
            self.mock_recalculate_ended_sessions.call_count, 0)

    def test_batch_op(self):
        for op in ["expire", "end", "regrade", "recalculate"]:
            for rule_tag in [fake_access_rules_tag, grades.RULE_TAG_NONE_STRING]:
                with self.subTest(user=self.instructor_participation.user, op=op):
                    resp = self.post_gradebook_by_opp_view(
                        self.gopp_id,
                        post_data={"rule_tag": rule_tag,
                                   "past_due_only": True,
                                   op: ""})

                    self.assertRedirects(
                        resp, reverse(
                            "relate-monitor_task",
                            kwargs={"task_id": fake_task_id}),
                        fetch_redirect_response=False)

        self.assertEqual(
            self.mock_expire_in_progress_sessions.call_count, 2)
        self.assertEqual(
            self.mock_finish_in_progress_sessions.call_count, 2)
        self.assertEqual(
            self.mock_regrade_flow_sessions.call_count, 2)
        self.assertEqual(
            self.mock_recalculate_ended_sessions.call_count, 2)

    def test_invalid_batch_op(self):
        resp = self.post_gradebook_by_opp_view(
            self.gopp_id,
            post_data={"rule_tag": grades.RULE_TAG_NONE_STRING,
                       "past_due_only": True,
                       "invalid_op": ""})

        # because post is neglected for user without those pperms
        self.assertEqual(resp.status_code, 400)
        self.assertEqual(
            self.mock_expire_in_progress_sessions.call_count, 0)
        self.assertEqual(
            self.mock_finish_in_progress_sessions.call_count, 0)
        self.assertEqual(
            self.mock_regrade_flow_sessions.call_count, 0)
        self.assertEqual(
            self.mock_recalculate_ended_sessions.call_count, 0)

    def test_post_form_invalid(self):
        with mock.patch(
                "course.grades.ModifySessionsForm.is_valid") as mock_form_valid:
            mock_form_valid.return_value = False
            resp = self.post_gradebook_by_opp_view(
                self.gopp_id,
                post_data={"rule_tag": grades.RULE_TAG_NONE_STRING,
                           "past_due_only": True,
                           "end": ""})

        # just ignore
        self.assertEqual(resp.status_code, 200)
        self.assertEqual(
            self.mock_expire_in_progress_sessions.call_count, 0)
        self.assertEqual(
            self.mock_finish_in_progress_sessions.call_count, 0)
        self.assertEqual(
            self.mock_regrade_flow_sessions.call_count, 0)
        self.assertEqual(
            self.mock_recalculate_ended_sessions.call_count, 0)

    def test_get_flow_status(self):
        factories.FlowSessionFactory(participation=self.student_participation,
                                     in_progress=True, page_count=13)

        # There're 3 participations, student has 2 finished session,
        # 1 in-progress session

        not_started = '<span class="badge bg-danger">not started</span>'
        finished = '<span class="badge bg-success">finished</span>'
        unfinished = '<span class="badge bg-warning">unfinished</span>'

        resp = self.get_gradebook_by_opp_view(self.gopp_id)
        self.assertEqual(resp.status_code, 200)
        # The instructor and ta didn't start the session
        self.assertContains(resp, not_started, count=2, html=True)

        self.assertContains(resp, finished, count=2, html=True)
        self.assertContains(resp, unfinished, count=1, html=True)

        resp = self.get_gradebook_by_opp_view(self.gopp_id, view_page_grades=True)
        self.assertEqual(resp.status_code, 200)

        # The student_participation has 2 session in setUp
        self.assertContains(resp, finished, count=2, html=True)
        self.assertContains(resp, unfinished, count=1, html=True)

        # no "not started" when view_page_grades
        self.assertContains(resp, not_started, count=0, html=True)

        # remove all flow sessions
        for fs in models.FlowSession.objects.all():
            fs.delete()

        resp = self.get_gradebook_by_opp_view(self.gopp_id)
        self.assertEqual(resp.status_code, 200)
        self.assertContains(resp, not_started, count=3, html=True)
        self.assertContains(resp, finished, count=0, html=True)
        self.assertContains(resp, unfinished, count=0, html=True)

        resp = self.get_gradebook_by_opp_view(self.gopp_id, view_page_grades=True)
        self.assertEqual(resp.status_code, 200)

        # no "not started" when view_page_grades
        self.assertContains(resp, not_started, count=0, html=True)
        self.assertContains(resp, finished, count=0, html=True)
        self.assertContains(resp, unfinished, count=0, html=True)

    def test_get_with_multiple_flow_sessions(self):
        factories.FlowSessionFactory(
            participation=self.student_participation,
            flow_id=QUIZ_FLOW_ID,
            in_progress=True)
        resp = self.get_gradebook_by_opp_view(self.gopp_id)
        self.assertEqual(resp.status_code, 200)

    def test_get_with_multiple_flow_sessions_view_page_grade(self):
        factories.FlowSessionFactory(
            participation=self.student_participation,
            flow_id=QUIZ_FLOW_ID,
            in_progress=True,
            page_count=12
        )

        resp = self.get_gradebook_by_opp_view(self.gopp_id, view_page_grades=True)
        self.assertEqual(resp.status_code, 200)

    def test_non_session_gopp(self):
        gopp = factories.GradingOpportunityFactory(
            course=self.course, identifier="another_gopp", flow_id=None)

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

        resp = self.get_gradebook_by_opp_view(gopp.identifier, view_page_grades=True)
        self.assertEqual(resp.status_code, 200)

        resp = self.get_gradebook_by_opp_view(gopp.identifier)
        self.assertEqual(resp.status_code, 200)


class GradesChangeStateMachineTest(GradesTestMixin, TestCase):

    def test_no_gradechange(self):
        # when no grade change object exists
        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_default_setup(self):
        self.use_default_setup()
        self.assertGradeChangeMachineReadableStateEqual(6)
        self.assertGradeChangeStateEqual("6.0% (/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)

    def test_change_aggregate_strategy_average(self):
        self.use_default_setup()
        self.update_gopp_strategy(g_stragety.avg_grade)
        self.assertGradeChangeMachineReadableStateEqual(4.333)
        self.assertGradeChangeStateEqual("4.3% (/3)")

    def test_change_aggregate_strategy_earliest(self):
        self.use_default_setup()
        self.update_gopp_strategy(g_stragety.use_earliest)
        self.assertGradeChangeMachineReadableStateEqual(0)
        self.assertGradeChangeStateEqual("0.0% (/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", 0)
        self.assertResponseContextEqual(resp, "avg_grade_population", 1)

    def test_change_aggregate_strategy_max(self):
        self.use_default_setup()
        self.update_gopp_strategy(g_stragety.max_grade)
        self.assertGradeChangeMachineReadableStateEqual(7)
        self.assertGradeChangeStateEqual("7.0% (/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", 7)
        self.assertResponseContextEqual(resp, "avg_grade_population", 1)

    def test_change_aggregate_strategy_max_none(self):
        # when no grade change has percentage
        self.update_gopp_strategy(g_stragety.max_grade)
        self.assertGradeChangeMachineReadableStateEqual("NONE")
        self.assertGradeChangeStateEqual("- ∅ -")

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

        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_change_aggregate_strategy_min(self):
        self.use_default_setup()
        self.update_gopp_strategy(g_stragety.min_grade)
        self.assertGradeChangeMachineReadableStateEqual(0)
        self.assertGradeChangeStateEqual("0.0% (/3)")

    def test_change_aggregate_strategy_min_none(self):
        # when no grade change has percentage
        self.update_gopp_strategy(g_stragety.min_grade)
        self.assertGradeChangeMachineReadableStateEqual("NONE")
        self.assertGradeChangeStateEqual("- ∅ -")

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

    def test_change_aggregate_strategy_invalid(self):
        self.use_default_setup()
        self.update_gopp_strategy("invalid_strategy")
        with self.assertRaises(ValueError):
            self.get_gc_stringify_machine_readable_state()

    def test_average_grade_value(self):
        # Other tests for course.grades.average_grade
        self.use_default_setup()
        self.assertGradeChangeMachineReadableStateEqual(6)
        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().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().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.assertAddMessageCallCount(1)
        self.assertAddMessageCalledWith(
            "Cannot reopen a session that's already in progress.")
        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 = (
            self.get_hacked_flow_desc_with_access_rule_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().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.client.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.assertAddMessageCalledWith(
            "This grade is not shown in the grade book.")

        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.assertAddMessageCalledWith(
            "This grade is not shown in the student grade book.")

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

    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.assertAddMessageCalledWith(msg, reset=True)
                mock_func.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)

    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.assertAddMessageNotCalledWith(msg, reset=False)
                self.assertAddMessageCalledWith(
                    "Error: KeyboardInterrupt %s" % err, reset=True)

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

        client = Client()
        client.force_login(cls.student_participation.user)

        cls.start_flow(client, cls.flow_id)
        cls.submit_page_answer_by_page_id_and_test(
            client, cls.page_id, answer_data={"answer": 0.25})
        cls.end_flow(client)

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

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

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

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

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

    def get_zip_file_buf_from_response(self, resp):
        return io.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 = io.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])

    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 = (
            self.get_hacked_flow_desc_with_access_rule_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()))
        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.client.get(flow_session_delete_url)
            self.assertEqual(resp.status_code, 200)
            resp = self.client.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
