test_grades.py

__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):
        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):
        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):