test_grades.py

# -*- coding: utf-8 -*-

from __future__ import division

__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 six
import datetime
from django.test import TestCase
from django.urls import reverse
from django.utils.timezone import now, timedelta
import unittest

from relate.utils import (
    local_now,
    dict_to_struct, struct_to_dict)

from course import models, grades, constants
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 tests.utils import mock  # noqa
from tests.base_test_mixins import (
    SingleCoursePageTestMixin, SingleCourseQuizPageTestMixin)
from tests import factories
from tests.test_flow.test_flow import HackRepoMixin
from tests.constants import QUIZ_FLOW_ID


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):
    time = now() - timedelta(days=10)

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

    def setUp(self):
        super(GradesTestMixin, self).setUp()
        self.gopp.refresh_from_db()
        fake_add_message = mock.patch("course.grades.messages.add_message")
        self.mock_add_message = fake_add_message.start()
        self.addCleanup(fake_add_message.stop)

    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_text
        self.assertEqual(force_text(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) != 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)

    @unittest.skipIf(six.PY2, "PY2 doesn't support subTest")
    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


class GetGradeTableTest(GradesTestMixin, TestCase):
    # test grades.get_grade_table

    @classmethod
    def setUpTestData(cls):  # noqa
        super(GetGradeTableTest, cls).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(GetGradeTableTest, self).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 = [l for l in gchange_kwargs_lists if len(l)]
            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(object):
    def __init__(self):
        self.id = fake_task_id


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

    def setUp(self):
        super(ViewGradesByOpportunityTest, self).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.c.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.c.get(self.get_gradebook_url_by_opp_id(
                another_course_gopp.id))
            self.assertEqual(resp.status_code, 400)

    @unittest.skipIf(six.PY2, "PY2 doesn't support subTest")
    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)

    @unittest.skipIf(six.PY2, "PY2 doesn't support subTest")
    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)

    @unittest.skipIf(six.PY2, "PY2 doesn't support subTest")
    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(self):
        resp = self.get_gradebook_by_opp_view(self.gopp_id)
        self.assertEqual(resp.status_code, 200)

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