test_flow.py

from __future__ import division

__copyright__ = "Copyright (C) 2018 Dong Zhuang"

__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 itertools

import unittest
from django import http
from django.urls import reverse
from django.contrib.sessions.middleware import SessionMiddleware
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import PermissionDenied
from django.test import TestCase, RequestFactory
from django.utils.timezone import now, timedelta
from django.core import mail

from relate.utils import dict_to_struct, StyledForm

from course import models, flow
from course import constants
from course.constants import grade_aggregation_strategy as g_strategy
from course.constants import flow_permission as fperm
from course.utils import FlowSessionStartRule, FlowSessionGradingRule

from tests.base_test_mixins import (
    CoursesTestMixinBase, SingleCourseQuizPageTestMixin, SingleCourseTestMixin,
    HackRepoMixin)
from tests.constants import QUIZ_FLOW_ID
from tests.utils import mock, may_run_expensive_tests, SKIP_EXPENSIVE_TESTS_REASON
from tests import factories


def get_flow_permissions_list(excluded=None):
    if not isinstance(excluded, list):
        excluded = [excluded]
    all_flow_permissions = dict(constants.FLOW_PERMISSION_CHOICES).keys()
    return [fp for fp in all_flow_permissions if fp not in excluded]


#{{{ test flow.adjust_flow_session_page_data

def flow_page_data_save_side_effect(self, *args, **kwargs):
    if self.page_id == "half1":
        raise RuntimeError("this error should not have been raised!")


class AdjustFlowSessionPageDataTest(
        SingleCourseQuizPageTestMixin, HackRepoMixin, TestCase):
    # test flow.adjust_flow_session_page_data

    initial_commit_sha = "my_fake_commit_sha_1"

    @classmethod
    def setUpTestData(cls):  # noqa
        super(AdjustFlowSessionPageDataTest, cls).setUpTestData()
        cls.start_flow(flow_id=cls.flow_id)

    def test_remove_rename_and_revive(self):
        # {{{ 1st round: do a visit
        resp = self.c.get(self.get_page_url_by_ordinal(0))
        self.assertEqual(resp.status_code, 200)

        fpds_1st = models.FlowPageData.objects.all()
        fpd_ids_1st = list(fpds_1st.values_list("page_id", flat=True))
        welcome_page_title_1st = fpds_1st.get(page_id="welcome").title
        # }}}

        # {{{ 2nd round: change sha
        self.course.active_git_commit_sha = "my_fake_commit_sha_2"
        self.course.save()

        resp = self.c.get(self.get_page_url_by_ordinal(0))
        self.assertEqual(resp.status_code, 200)

        fpds_2nd = models.FlowPageData.objects.all()
        welcome_page_title_2nd = fpds_2nd.get(page_id="welcome").title
        fpd_ids_2nd = list(fpds_2nd.values_list("page_id", flat=True))

        # the page (with page_id "welcome") has changed title
        self.assertNotEqual(welcome_page_title_1st, welcome_page_title_2nd)

        # these two pages have removed page_ordinal
        # (those in group2 are not considered)
        page_ids_removed_in_2nd = {"half1", "lsq2"}
        self.assertTrue(
            page_ids_removed_in_2nd
            < set(list(
                fpds_2nd.filter(
                    page_ordinal=None).values_list("page_id", flat=True)))
        )

        page_ids_introduced_in_2nd = {"half1_id_renamed", "half_again2"}
        self.assertNotIn(page_ids_introduced_in_2nd, fpd_ids_1st)
        self.assertTrue(page_ids_introduced_in_2nd < set(fpd_ids_2nd))

        self.assertTrue(set(fpd_ids_2nd) > set(fpd_ids_1st))
        # }}}

        # {{{ 3rd round: revive back
        self.course.active_git_commit_sha = "my_fake_commit_sha_1"
        self.course.save()

        resp = self.c.get(self.get_page_url_by_ordinal(0))
        self.assertEqual(resp.status_code, 200)

        fpds_3rd = models.FlowPageData.objects.all()
        fpd_ids_3rd = list(fpds_3rd.values_list("page_id", flat=True))
        welcome_page_title_3rd = fpds_2nd.get(page_id="welcome").title
        self.assertEqual(welcome_page_title_1st, welcome_page_title_3rd)

        # no page_data instances are removed
        self.assertSetEqual(set(fpd_ids_2nd), set(fpd_ids_3rd))
        self.assertSetEqual(
            page_ids_introduced_in_2nd,
            set(list(
                fpds_3rd.filter(
                    page_ordinal=None).values_list("page_id", flat=True))))
        for page_id in page_ids_removed_in_2nd:
            self.assertIsNotNone(
                models.FlowPageData.objects.get(page_id=page_id).page_ordinal)
            # }}}

    # disabled by AK 2020-05-03: why is it valid to set a flow page's ordinal
    # to None while it is in use?
    def no_test_remove_page_with_non_ordinal(self):
        resp = self.c.get(self.get_page_url_by_ordinal(0))
        self.assertEqual(resp.status_code, 200)

        # change this page's ordinal to None before change the commit_sha,
        # so that no save is needed when update course, for this page
        fpd = models.FlowPageData.objects.get(page_id="half1")
        fpd.page_ordinal = None
        fpd.save()

        with mock.patch(
                "course.models.FlowPageData.save",
                autospec=True) as mock_fpd_save:
            mock_fpd_save.side_effect = flow_page_data_save_side_effect

            self.course.active_git_commit_sha = "my_fake_commit_sha_2"
            self.course.save()

            resp = self.c.get(self.get_page_url_by_ordinal(0))
            self.assertEqual(resp.status_code, 200)

# }}}


class GradePageVisitTest(SingleCourseQuizPageTestMixin, TestCase):
    # patching tests for flow.grade_page_visits
    def test_not_is_submitted_answer(self):
        visit = mock.MagicMock()
        visit_grade_model = mock.MagicMock()
        visit.is_submitted_answer = False

        expected_error_msg = "cannot grade ungraded answer"
        with self.assertRaises(RuntimeError) as cm:
            flow.grade_page_visit(visit, visit_grade_model)
        self.assertIn(expected_error_msg, str(cm.exception))

        with self.assertRaises(RuntimeError) as cm:
            flow.grade_page_visit(visit, visit_grade_model, {"key": "value"})
        self.assertIn(expected_error_msg, str(cm.exception))

        with self.assertRaises(RuntimeError) as cm:
            flow.grade_page_visit(visit, visit_grade_model, {"key": "value"}, False)
        self.assertIn(expected_error_msg, str(cm.exception))

    def test_page_answer_not_gradable(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            self.start_flow(self.flow_id)
            fpvgs = models.FlowPageVisitGrade.objects.all()
            self.assertEqual(fpvgs.count(), 0)

            page_id = "age_group"

            self.submit_page_answer_by_page_id_and_test(
                page_id, do_grading=True, expected_grades=0)

            fpvgs = models.FlowPageVisitGrade.objects.filter(