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 os
import six
from collections import OrderedDict

import unittest
from django.test import TestCase
from django.utils.timezone import now, timedelta

from relate.utils import dict_to_struct

from course.content import get_repo_blob
from course import models, flow
from course import constants
from course.constants import grade_aggregation_strategy as g_strategy
from course.utils import FlowSessionStartRule, FlowSessionGradingRule

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

YAML_PATH = os.path.join(os.path.dirname(__file__), 'resource')


class Blob(object):
    def __init__(self, yaml_file_name):
        with open(os.path.join(YAML_PATH, yaml_file_name), "rb") as f:
            data = f.read()
        self.data = data


# This is need to for correctly getting other blob
current_commit_sha = b"4124e0c23e369d6709a670398167cb9c2fe52d35"

COMMIT_SHA_MAP = {
    "flows/%s.yml" % QUIZ_FLOW_ID: [
        {"my_fake_commit_sha_1": {"path": "fake-quiz-test1.yml"}},
        {"my_fake_commit_sha_2": {"path": "fake-quiz-test2.yml"}},

        {"my_fake_commit_sha_for_grades1": {
            "path": "fake-quiz-test-for-grade1.yml",
            "page_ids": ["half", "krylov", "quarter"]}},
        {"my_fake_commit_sha_for_grades2": {
            "path": "fake-quiz-test-for-grade2.yml",
            "page_ids": ["krylov", "quarter"]}},

        {"my_fake_commit_sha_for_gradesinfo": {
            "path": "fake-quiz-test-for-gradeinfo.yml",
            "page_ids": ["half", "krylov", "matrix_props", "age_group",
                         "anyup", "proof", "neumann"]
        }},

        {"my_fake_commit_sha_for_grade_flow_session": {
            "path": "fake-quiz-test-for-grade_flow_session.yml",
            "page_ids": ["anyup"]}},
        {"my_fake_commit_sha_for_grade_flow_session2": {
            "path": "fake-quiz-test-for-grade_flow_session2.yml",
            "page_ids": ["anyup"]}}
    ],

    "flow/%s.yml" % "001-linalg-recap":
        [{"my_fake_commit_sha_3": {"path": "fake-001-linalg-recap.yml"}}]
}


def get_repo_side_effect(repo, full_name, commit_sha, allow_tree=True):
    commit_sha_path_maps = COMMIT_SHA_MAP.get(full_name)
    if commit_sha_path_maps:
        assert isinstance(commit_sha_path_maps, list)
        for cs_map in commit_sha_path_maps:
            if commit_sha.decode() in cs_map:
                path = cs_map[commit_sha.decode()]["path"]
                return Blob(path)

    return get_repo_blob(repo, full_name, current_commit_sha, allow_tree=allow_tree)


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 BatchFakeGetRepoBlobMixin(object):
    def setUp(self):
        super(BatchFakeGetRepoBlobMixin, self).setUp()
        batch_fake_get_repo_blob = mock.patch(
            "course.content.get_repo_blob")
        self.batch_mock_get_repo_blob = batch_fake_get_repo_blob.start()
        self.batch_mock_get_repo_blob.side_effect = get_repo_side_effect
        self.addCleanup(batch_fake_get_repo_blob.stop)

    def get_current_page_ids(self):
        current_sha = self.course.active_git_commit_sha
        for commit_sha_path_maps in COMMIT_SHA_MAP.values():
            for cs_map in commit_sha_path_maps:
                if current_sha in cs_map:
                    return cs_map[current_sha]["page_ids"]

        raise ValueError("Page_ids for that commit_sha doesn't exist")

    def assertGradeInfoEqual(self, resp, expected_grade_info_dict=None):  # noqa
        grade_info = resp.context["grade_info"]

        assert isinstance(grade_info, flow.GradeInfo)
        if not expected_grade_info_dict:
            import json
            error_msg = ("\n%s" % json.dumps(OrderedDict(
                sorted(
                    [(k, v) for (k, v) in six.iteritems(grade_info.__dict__)])),
                indent=4))
            error_msg = error_msg.replace("null", "None")
            self.fail(error_msg)

        assert isinstance(expected_grade_info_dict, dict)

        grade_info_dict = grade_info.__dict__
        not_match_infos = []
        for k in grade_info_dict.keys():
            if grade_info_dict[k] != expected_grade_info_dict[k]:
                not_match_infos.append(
                    "'%s' is expected to be %s, while got %s"
                    % (k, str(expected_grade_info_dict[k]),
                       str(grade_info_dict[k])))

        if not_match_infos:
            self.fail("\n".join(not_match_infos))


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

    def setUp(self):
        super(AdjustFlowSessionPageDataTest, self).setUp()
        self.c.force_login(self.student_participation.user)

    def test_remove_rename_and_revive(self):
        self.course.active_git_commit_sha = "my_fake_commit_sha_1"
        self.course.save()

        self.start_flow(flow_id=self.flow_id)

        # {{{ 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)))
        )