base_test_mixins.py

from __future__ import division

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

__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 tempfile
import os
import shutil
import hashlib
import datetime
import memcache
from copy import deepcopy
from django.test import Client, override_settings, mock
from django.urls import reverse, resolve
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.exceptions import ImproperlyConfigured

from course.models import (
    Course, Participation, ParticipationRole, FlowSession, FlowPageData,
    FlowPageVisit)
from course.constants import participation_status, user_status
from course.content import get_course_repo_path


CREATE_SUPERUSER_KWARGS = {
    "username": "test_admin",
    "password": "test_admin",
    "email": "test_admin@example.com",
    "first_name": "Test",
    "last_name": "Admin"}

SINGLE_COURSE_SETUP_LIST = [
    {
        "course": {
            "identifier": "test-course",
            "name": "Test Course",
            "number": "CS123",
            "time_period": "Fall 2016",
            "hidden": False,
            "listed": True,
            "accepts_enrollment": True,
            "git_source": "git://github.com/inducer/relate-sample",
            "course_file": "course.yml",
            "events_file": "events.yml",
            "enrollment_approval_required": False,
            "enrollment_required_email_suffix": "",
            "preapproval_require_verified_inst_id": True,
            "from_email": "inform@tiker.net",
            "notify_email": "inform@tiker.net"},
        "participations": [
            {
                "role_identifier": "instructor",
                "user": {
                    "username": "test_instructor",
                    "password": "test_instructor",
                    "email": "test_instructor@example.com",
                    "first_name": "Test",
                    "last_name": "Instructor"},
                "status": participation_status.active
            },
            {
                "role_identifier": "ta",
                "user": {
                    "username": "test_ta",
                    "password": "test",
                    "email": "test_ta@example.com",
                    "first_name": "Test",
                    "last_name": "TA"},
                "status": participation_status.active
            },
            {
                "role_identifier": "student",
                "user": {
                    "username": "test_student",
                    "password": "test",
                    "email": "test_student@example.com",
                    "first_name": "Test",
                    "last_name": "Student"},
                "status": participation_status.active
            }
        ],
    }
]

TWO_COURSE_SETUP_LIST = deepcopy(SINGLE_COURSE_SETUP_LIST)
TWO_COURSE_SETUP_LIST[0]["course"]["identifier"] = "test-course1"
TWO_COURSE_SETUP_LIST += deepcopy(SINGLE_COURSE_SETUP_LIST)
TWO_COURSE_SETUP_LIST[1]["course"]["identifier"] = "test-course2"

NONE_PARTICIPATION_USER_CREATE_KWARG_LIST = [
    {
        "username": "test_user1",
        "password": "test_user1",
        "email": "test_user1@suffix.com",
        "first_name": "Test",
        "last_name": "User1",
        "institutional_id": "test_user1_institutional_id",
        "institutional_id_verified": True,
        "status": user_status.active
    },
    {
        "username": "test_user2",
        "password": "test_user2",
        "email": "test_user2@nosuffix.com",
        "first_name": "Test",
        "last_name": "User2",
        "institutional_id": "test_user2_institutional_id",
        "institutional_id_verified": False,
        "status": user_status.active
    },
    {
        "username": "test_user3",
        "password": "test_user3",
        "email": "test_user3@suffix.com",
        "first_name": "Test",
        "last_name": "User3",
        "institutional_id": "test_user3_institutional_id",
        "institutional_id_verified": True,
        "status": user_status.unconfirmed
    },
    {
        "username": "test_user4",
        "password": "test_user4",
        "email": "test_user4@no_suffix.com",
        "first_name": "Test",
        "last_name": "User4",
        "institutional_id": "test_user4_institutional_id",
        "institutional_id_verified": False,
        "status": user_status.unconfirmed
    }
]

try:
    mc = memcache.Client(['127.0.0.1:11211'])
except Exception:
    pass


def git_source_url_to_cache_keys(url):
    url_hash = hashlib.md5(url.encode("utf-8")).hexdigest()
    return (
        "test_course:%s" % url_hash,
        "test_sha:%s" % url_hash
    )


class CourseCreateFailure(Exception):
    pass


class ResponseContextMixin(object):
    """
    Response context refers to "the template Context instance that was used
    to render the template that produced the response content".
    Ref: https://docs.djangoproject.com/en/dev/topics/testing/tools/#django.test.Response.context  # noqa
    """
    def get_response_context_value_by_name(self, response, context_name):
        value = response.context.__getitem__(context_name)
        self.assertIsNotNone(
            value,
            msg="%s does not exist in given response" % context_name)
        return value

    def assertResponseContextIsNone(self, resp, context_name):  # noqa
        try:
            value = self.get_response_context_value_by_name(resp, context_name)
        except AssertionError:
            # the context item doesn't exist
            pass
        else:
            self.assertIsNone(value)

    def assertResponseContextIsNotNone(self, resp, context_name):  # noqa
        value = self.get_response_context_value_by_name(resp, context_name)
        self.assertIsNotNone(value)

    def assertResponseContextEqual(self, resp, context_name, expected_value):  # noqa
        value = self.get_response_context_value_by_name(resp, context_name)
        self.assertEqual(value, expected_value)

    def assertResponseContextContains(self, resp,  # noqa
                                      context_name, expected_value, html=False):
        value = self.get_response_context_value_by_name(resp, context_name)
        if not html:
            self.assertIn(expected_value, value)
        else:
            self.assertInHTML(expected_value, value)

    def assertResponseContextRegex(  # noqa
            self, resp,  # noqa
            context_name, expected_value_regex):
        value = self.get_response_context_value_by_name(resp, context_name)
        six.assertRegex(self, value, expected_value_regex)

    def get_response_context_answer_feedback(self, response):
        return self.get_response_context_value_by_name(response, "feedback")

    def assertResponseContextAnswerFeedbackContainsFeedback(  # noqa
                                        self, response, expected_feedback):
        answer_feedback = self.get_response_context_answer_feedback(response)
        self.assertTrue(hasattr(answer_feedback, "feedback"))
        self.assertIn(expected_feedback, answer_feedback.feedback)

    def assertResponseContextAnswerFeedbackCorrectnessEquals(  # noqa
                                        self, response, expected_correctness):
        answer_feedback = self.get_response_context_answer_feedback(response)
        if expected_correctness is None:
            try:
                self.assertTrue(hasattr(answer_feedback, "correctness"))
            except AssertionError:
                pass
            else:
                self.assertIsNone(answer_feedback.correctness)
        else:
            from decimal import Decimal
            self.assertEqual(answer_feedback.correctness,
                                    Decimal(str(expected_correctness)))

    def get_response_body(self, response):
        return self.get_response_context_value_by_name(response, "body")

    def get_page_response_correct_answer(self, response):
        return self.get_response_context_value_by_name(response, "correct_answer")

    def get_page_response_feedback(self, response):
        return self.get_response_context_value_by_name(response, "feedback")

    def debug_print_response_context_value(self, resp, context_name):
        try:
            value = self.get_response_context_value_by_name(resp, context_name)
            print("\n-----------context %s-------------"
                  % context_name)
            if isinstance(value, (list, tuple)):
                from course.validation import ValidationWarning
                for v in value:
                    if isinstance(v, ValidationWarning):
                        print(v.text)
                    else:
                        print(repr(v))
            else:
                print(value)
            print("-----------context end-------------\n")
        except AssertionError:
            print("\n-------no value for context %s----------" % context_name)


class SuperuserCreateMixin(ResponseContextMixin):
    create_superuser_kwargs = CREATE_SUPERUSER_KWARGS

    @classmethod
    def setUpTestData(cls):  # noqa
        # Create superuser, without this, we cannot
        # create user, course and participation.
        cls.superuser = cls.create_superuser()
        cls.c = Client()
        cls.settings_git_root_override = (
            override_settings(GIT_ROOT=tempfile.mkdtemp()))
        cls.settings_git_root_override.enable()
        super(SuperuserCreateMixin, cls).setUpTestData()

    @classmethod
    def tearDownClass(cls):  # noqa
        super(SuperuserCreateMixin, cls).tearDownClass()

    @classmethod
    def create_superuser(cls):
        return get_user_model().objects.create_superuser(
                                                **cls.create_superuser_kwargs)

    def get_impersonate_view_url(self):
        return reverse("relate-impersonate")

    def get_stop_impersonate_view_url(self):
        return reverse("relate-stop_impersonating")

    def get_impersonate(self):
        return self.c.get(self.get_impersonate_view_url())

    def post_impersonate(self, impersonatee, follow=True):
        data = {"add_impersonation_header": ["on"],
                "submit": [''],
                }
        data["user"] = [str(impersonatee.pk)]
        return self.c.post(self.get_impersonate_view_url(), data, follow=follow)

    def get_stop_impersonate(self, follow=True):
        return self.c.get(self.get_stop_impersonate_view_url(), follow=follow)

    def post_stop_impersonate(self, follow=True):
        data = {"submit": ['']}
        return self.c.post(
            self.get_stop_impersonate_view_url(), data, follow=follow)

    def get_fake_time_url(self):
        return reverse("relate-set_fake_time")

    def get_set_fake_time(self):
        return self.c.get(self.get_fake_time_url())

    def post_set_fake_time(self, data, follow=True):
        return self.c.post(self.get_fake_time_url(), data, follow=follow)

    def assertSessionFakeTimeEqual(self, session, expected_date_time):  # noqa
        fake_time_timestamp = session.get("relate_fake_time", None)
        if fake_time_timestamp is None:
            faked_time = None
            if expected_date_time is not None:
                raise AssertionError(
                    "the session doesn't have 'relate_fake_time' attribute")
        else:
            faked_time = datetime.datetime.fromtimestamp(fake_time_timestamp)
        self.assertEqual(faked_time, expected_date_time)

    def assertSessionFakeTimeIsNone(self, session):  # noqa
        self.assertSessionFakeTimeEqual(session, None)

    def get_set_pretend_facilities_url(self):
        return reverse("relate-set_pretend_facilities")

    def get_set_pretend_facilities(self):
        return self.c.get(self.get_set_pretend_facilities_url())

    def post_set_pretend_facilities(self, data, follow=True):
        return self.c.post(self.get_set_pretend_facilities_url(), data,
                           follow=follow)

    def assertSessionPretendFacilitiesContains(self, session, expected_facilities):  # noqa
        pretended = session.get("relate_pretend_facilities", None)
        if expected_facilities is None:
            return self.assertIsNone(pretended)
        if pretended is None:
            raise AssertionError(
                "the session doesn't have "
                "'relate_pretend_facilities' attribute")

        if isinstance(expected_facilities, (list, tuple)):
            self.assertTrue(set(expected_facilities).issubset(set(pretended)))
        else:
            self.assertTrue(expected_facilities in pretended)

    def assertSessionPretendFacilitiesIsNone(self, session):  # noqa
        pretended = session.get("relate_pretend_facilities", None)
        self.assertIsNone(pretended)


# {{{ defined here so that they can be used by in classmethod and instance method

def get_flow_page_ordinal_from_page_id(flow_session_id, page_id):
    flow_page_data = FlowPageData.objects.get(
        flow_session__id=flow_session_id,
        page_id=page_id
    )
    return flow_page_data.page_ordinal


def get_flow_page_id_from_page_ordinal(flow_session_id, page_ordinal):
    flow_page_data = FlowPageData.objects.get(
        flow_session__id=flow_session_id,
        page_ordinal=page_ordinal
    )
    return flow_page_data.page_id

# }}}


class CoursesTestMixinBase(SuperuserCreateMixin):

    # A list of Dicts, each of which contain a course dict and a list of
    # participations. See SINGLE_COURSE_SETUP_LIST for the setup for one course.
    courses_setup_list = []
    none_participation_user_create_kwarg_list = []
    courses_attributes_extra_list = None
    override_settings_at_post_create_course = {}

    @classmethod
    def setUpTestData(cls):  # noqa
        super(CoursesTestMixinBase, cls).setUpTestData()
        cls.default_flow_params = None
        cls.n_courses = 0
        if cls.courses_attributes_extra_list is not None:
            if (len(cls.courses_attributes_extra_list)
                    != len(cls.courses_setup_list)):
                raise ValueError(
                    "'courses_attributes_extra_list' must has equal length "
                    "with courses")

        for i, course_setup in enumerate(cls.courses_setup_list):
            if "course" not in course_setup:
                continue

            cls.n_courses += 1
            course_identifier = course_setup["course"]["identifier"]
            course_setup_kwargs = course_setup["course"]
            if cls.courses_attributes_extra_list:
                extra_attrs = cls.courses_attributes_extra_list[i]
                assert isinstance(extra_attrs, dict)
                course_setup_kwargs.update(extra_attrs)
            try:
                cls.create_course(**course_setup_kwargs)
            except Exception:
                raise

            course = Course.objects.get(identifier=course_identifier)
            if "participations" in course_setup:
                for participation in course_setup["participations"]:
                    create_user_kwargs = participation.get("user")
                    if not create_user_kwargs:
                        continue
                    role_identifier = participation.get("role_identifier")
                    if not role_identifier:
                        continue
                    cls.create_participation(
                        course=course,
                        user_or_create_user_kwargs=create_user_kwargs,
                        role_identifier=role_identifier,
                        status=participation.get("status",
                                                 participation_status.active)
                    )

            # Remove superuser from participation for further test
            # such as impersonate in auth module
            try:
                superuser_participations = (
                    Participation.objects.filter(user=cls.superuser))
                for sp in superuser_participations:
                    Participation.delete(sp)
            except Participation.DoesNotExist:
                pass

            cls.non_participation_users = get_user_model().objects.none()
            if cls.none_participation_user_create_kwarg_list:
                pks = []
                for create_user_kwargs in (
                        cls.none_participation_user_create_kwarg_list):
                    user = cls.create_user(create_user_kwargs)
                    pks.append(user.pk)
                cls.non_participation_users = (
                    get_user_model().objects.filter(pk__in=pks))

        cls.course_qset = Course.objects.all()

    @classmethod
    def tearDownClass(cls):
        cls.c.logout()
        super(CoursesTestMixinBase, cls).tearDownClass()

    @classmethod
    def create_user(cls, create_user_kwargs):
        user, created = get_user_model().objects.get_or_create(
            email__iexact=create_user_kwargs["email"], defaults=create_user_kwargs)
        if created:
            try:
                # TODO: why pop failed here?
                password = create_user_kwargs["password"]
            except Exception:
                raise
            user.set_password(password)
            user.save()
        return user

    @classmethod
    def create_participation(
            cls, course, user_or_create_user_kwargs,
            role_identifier=None, status=None):
        if isinstance(user_or_create_user_kwargs, get_user_model()):
            user = user_or_create_user_kwargs
        else:
            assert isinstance(user_or_create_user_kwargs, dict)
            user = cls.create_user(user_or_create_user_kwargs)
        if status is None:
            status = participation_status.active
        participation, p_created = Participation.objects.get_or_create(
            user=user,
            course=course,
            status=status
        )
        if role_identifier is None:
            role_identifier = "student"
        if p_created:
            role = ParticipationRole.objects.filter(
                course=course, identifier=role_identifier)
            participation.roles.set(role)
        return participation

    @classmethod
    def post_create_course(cls, raise_error=True, **create_course_kwargs):
        # To speed up, use create_course instead, this is better used for tests
        cls.c.force_login(cls.superuser)
        existing_course_count = Course.objects.count()
        with override_settings(**cls.override_settings_at_post_create_course):
            resp = cls.c.post(
                reverse("relate-set_up_new_course"), create_course_kwargs)
        if raise_error:
            all_courses = Course.objects.all()
            if not all_courses.count() == existing_course_count + 1:
                error_string = None
                # most probably the reason course creation form error
                form_context = resp.context.__getitem__("form")
                assert form_context
                error_list = []
                if form_context.errors:
                    error_list = [
                        "%s: %s"
                        % (field,
                           "\n".join(["%s:%s" % (type(e).__name__, str(e))
                                      for e in errs]))
                        for field, errs
                        in six.iteritems(form_context.errors.as_data())]
                non_field_errors = form_context.non_field_errors()
                if non_field_errors:
                    error_list.append(repr(non_field_errors))
                if error_list:
                    error_string = "\n".join(error_list)
                if not error_string:
                    error_string = ("course creation failed for unknown errors")
                raise CourseCreateFailure(error_string)
            # the course is created successfully
            last_course = all_courses.last()
            assert last_course
            url_cache_key, commit_sha_cach_key = (
                git_source_url_to_cache_keys(last_course.git_source))
            mc.set_multi({url_cache_key: get_course_repo_path(last_course),
                          commit_sha_cach_key: last_course.active_git_commit_sha},
                         time=120)
        return resp

    @classmethod
    def create_course(cls, raise_error=True, **create_course_kwargs):
        has_cached_repo = False
        repo_cache_key, commit_sha_cach_key = (
            git_source_url_to_cache_keys(create_course_kwargs["git_source"]))
        try:
            exist_course_repo_path = mc.get(repo_cache_key)
            exist_commit_sha = mc.get(commit_sha_cach_key)
            if os.path.isdir(exist_course_repo_path):
                has_cached_repo = bool(exist_course_repo_path and exist_commit_sha)
            else:
                has_cached_repo = False
        except Exception:
            pass

        if not has_cached_repo:
            # fall back to post create
            return cls.post_create_course(
                raise_error=raise_error, **create_course_kwargs)
        existing_course_count = Course.objects.count()
        new_course_repo_path = os.path.join(settings.GIT_ROOT,
                                        create_course_kwargs["identifier"])
        shutil.copytree(exist_course_repo_path, new_course_repo_path)
        create_kwargs = deepcopy(create_course_kwargs)
        create_kwargs["active_git_commit_sha"] = exist_commit_sha
        Course.objects.create(**create_kwargs)
        assert Course.objects.count() == existing_course_count + 1

    @classmethod
    def get_course_page_url(cls, course_identifier=None):
        course_identifier = course_identifier or cls.get_default_course_identifier()
        return reverse("relate-course_page", args=[course_identifier])

    def get_logged_in_user(self):
        try:
            logged_in_user_id = self.c.session['_auth_user_id']
            from django.contrib.auth import get_user_model
            logged_in_user = get_user_model().objects.get(
                pk=int(logged_in_user_id))
        except KeyError:
            logged_in_user = None
        return logged_in_user

    def temporarily_switch_to_user(self, switch_to):
        _self = self

        from functools import wraps

        class ClientUserSwitcher(object):
            def __init__(self, switch_to):
                self.client = _self.c
                self.switch_to = switch_to
                self.logged_in_user = _self.get_logged_in_user()

            def __enter__(self):
                if self.logged_in_user == self.switch_to:
                    return
                if self.switch_to is None:
                    self.client.logout()
                    return
                self.client.force_login(self.switch_to)

            def __exit__(self, exc_type, exc_val, exc_tb):
                if self.logged_in_user == self.switch_to:
                    return
                if self.logged_in_user is None:
                    self.client.logout()
                    return
                self.client.force_login(self.logged_in_user)

            def __call__(self, func):
                @wraps(func)
                def wrapper(*args, **kw):
                    with self:
                        return func(*args, **kw)
                return wrapper

        return ClientUserSwitcher(switch_to)

    @classmethod
    def get_default_course(cls):
        if Course.objects.count() > 1:
            raise AttributeError(
                "'course' arg can not be omitted for "
                "testcases with more than one courses")
        raise NotImplementedError

    @classmethod
    def get_default_course_identifier(cls):
        if Course.objects.count() > 1:
            raise AttributeError(
                "'course_identifier' arg can not be omitted for "
                "testcases with more than one courses")
        raise NotImplementedError

    def get_latest_session_id(self, course_identifier):
        flow_session_qset = FlowSession.objects.filter(
            course__identifier=course_identifier).order_by('-pk')[:1]
        if flow_session_qset:
            return flow_session_qset[0].id
        else:
            return None

    def get_default_flow_session_id(self, course_identifier):
        raise NotImplementedError

    @classmethod
    def update_default_flow_session_id(cls, course_identifier):
        raise NotImplementedError

    def get_default_instructor_user(self, course_identifier):
        return Participation.objects.filter(
            course__identifier=course_identifier,
            roles__identifier="instructor",
            status=participation_status.active
        ).first().user

    @classmethod
    def update_course_attribute(cls, attrs, course=None):
        # course instead of course_identifier because we need to do
        # refresh_from_db
        assert isinstance(attrs, dict)
        course = course or cls.get_default_course()
        if attrs:
            course.__dict__.update(attrs)
            course.save()
            course.refresh_from_db()

    @classmethod
    def start_flow(cls, flow_id, course_identifier=None):
        """
        Notice: this is a classmethod, so this will change the data
        created in setUpTestData, so don't do this in individual tests, or
        testdata will be different between tests.
        """
        course_identifier = course_identifier or cls.get_default_course_identifier()
        existing_session_count = FlowSession.objects.all().count()
        params = {"course_identifier": course_identifier,
                  "flow_id": flow_id}
        resp = cls.c.post(reverse("relate-view_start_flow", kwargs=params))
        assert resp.status_code == 302
        new_session_count = FlowSession.objects.all().count()
        assert new_session_count == existing_session_count + 1
        _, _, params = resolve(resp.url)
        del params["page_ordinal"]
        cls.default_flow_params = params
        cls.update_default_flow_session_id(course_identifier)
        return resp

    @classmethod
    def end_flow(cls, course_identifier=None, flow_session_id=None):
        """
        Be cautious that this is a classmethod
        """
        if cls.default_flow_params is None:
            raise RuntimeError("There's no started flow_sessions.")
        params = deepcopy(cls.default_flow_params)
        if course_identifier:
            params["course_identifier"] = course_identifier
        if flow_session_id:
            params["flow_session_id"] = flow_session_id
        resp = cls.c.post(reverse("relate-finish_flow_session_view",
                                  kwargs=params), {'submit': ['']})
        return resp

    def get_flow_params(self, course_identifier=None, flow_session_id=None):
        course_identifier = (
            course_identifier or self.get_default_course_identifier())
        if flow_session_id is None:
            flow_session_id = self.get_default_flow_session_id(course_identifier)
        return {
            "course_identifier": course_identifier,
            "flow_session_id": flow_session_id
        }

    def get_page_params(self, course_identifier=None, flow_session_id=None,
                        page_ordinal=None):
        page_params = self.get_flow_params(course_identifier, flow_session_id)
        if page_ordinal is None:
            page_ordinal = 0
        page_params.update({"page_ordinal": page_ordinal})
        return page_params

    def get_page_ordinal_via_page_id(
            self, page_id, course_identifier=None, flow_session_id=None):
        flow_params = self.get_flow_params(course_identifier, flow_session_id)
        return (
            get_flow_page_ordinal_from_page_id(
                flow_params["flow_session_id"], page_id))

    def get_page_view_url_by_ordinal(
            self, viewname, page_ordinal, course_identifier=None,
            flow_session_id=None):
        page_params = self.get_page_params(
            course_identifier, flow_session_id, page_ordinal)
        return reverse(viewname, kwargs=page_params)

    def get_page_view_url_by_page_id(
            self, viewname, page_id, course_identifier=None, flow_session_id=None):
        page_ordinal = self.get_page_ordinal_via_page_id(
            page_id, course_identifier, flow_session_id)
        return self.get_page_view_url_by_ordinal(
            viewname, page_ordinal, course_identifier, flow_session_id)

    def get_page_url_by_ordinal(
            self, page_ordinal, course_identifier=None, flow_session_id=None):
        return self.get_page_view_url_by_ordinal(
            "relate-view_flow_page",
            page_ordinal, course_identifier, flow_session_id)

    def get_page_url_by_page_id(
            self, page_id, course_identifier=None, flow_session_id=None):
        page_ordinal = self.get_page_ordinal_via_page_id(
            page_id, course_identifier, flow_session_id)
        return self.get_page_url_by_ordinal(
            page_ordinal, course_identifier, flow_session_id)

    def get_page_grading_url_by_ordinal(
            self, page_ordinal, course_identifier=None, flow_session_id=None):
        return self.get_page_view_url_by_ordinal(
            "relate-grade_flow_page",
            page_ordinal, course_identifier, flow_session_id)

    def get_page_grading_url_by_page_id(
            self, page_id, course_identifier=None, flow_session_id=None):
        page_ordinal = self.get_page_ordinal_via_page_id(
            page_id, course_identifier, flow_session_id)
        return self.get_page_grading_url_by_ordinal(
            page_ordinal, course_identifier, flow_session_id)

    def post_answer_by_ordinal(
            self, page_ordinal, answer_data,
            course_identifier=None, flow_session_id=None):
        submit_data = answer_data
        submit_data.update({"submit": ["Submit final answer"]})
        resp = self.c.post(
            self.get_page_url_by_ordinal(
                page_ordinal, course_identifier, flow_session_id),
            submit_data)
        return resp

    def post_answer_by_page_id(self, page_id, answer_data,
                               course_identifier=None, flow_session_id=None):
        page_ordinal = self.get_page_ordinal_via_page_id(
            page_id, course_identifier, flow_session_id)
        return self.post_answer_by_ordinal(
            page_ordinal, answer_data, course_identifier, flow_session_id)

    @classmethod
    def post_answer_by_ordinal_class(cls, page_ordinal, answer_data,
                                     course_identifier, flow_session_id):
        submit_data = answer_data
        submit_data.update({"submit": ["Submit final answer"]})
        page_params = {
            "course_identifier": course_identifier,
            "flow_session_id": flow_session_id,
            "page_ordinal": page_ordinal
        }
        page_url = reverse("relate-view_flow_page", kwargs=page_params)
        resp = cls.c.post(page_url, submit_data)
        return resp

    @classmethod
    def post_answer_by_page_id_class(cls, page_id, answer_data,
                                     course_identifier, flow_session_id):
        page_ordinal = get_flow_page_ordinal_from_page_id(flow_session_id, page_id)
        return cls.post_answer_by_ordinal_class(page_ordinal, answer_data,
                                                course_identifier, flow_session_id)

    def post_grade_by_ordinal(self, page_ordinal, grade_data,
                              course_identifier=None, flow_session_id=None,
                              force_login_instructor=True):
        post_data = {"submit": [""]}
        post_data.update(grade_data)

        page_params = self.get_page_params(
            course_identifier, flow_session_id, page_ordinal)

        force_login_user = self.get_logged_in_user()
        if force_login_instructor:
            force_login_user = self.get_default_instructor_user(
                page_params["course_identifier"])

        with self.temporarily_switch_to_user(force_login_user):
            response = self.c.post(
                self.get_page_grading_url_by_ordinal(**page_params),
                data=post_data,
                follow=True)
        return response

    def post_grade_by_page_id(self, page_id, grade_data,
                              course_identifier=None, flow_session_id=None,
                              force_login_instructor=True):
        page_ordinal = self.get_page_ordinal_via_page_id(
            page_id, course_identifier, flow_session_id)

        return self.post_grade_by_ordinal(
            page_ordinal, grade_data, course_identifier,
            flow_session_id, force_login_instructor)

    def assertSessionScoreEqual(  # noqa
            self, expect_score, course_identifier=None, flow_session_id=None):
        if flow_session_id is None:
            flow_params = self.get_flow_params(course_identifier, flow_session_id)
            flow_session_id = flow_params["flow_session_id"]
        flow_session = FlowSession.objects.get(id=flow_session_id)
        if expect_score is not None:
            from decimal import Decimal
            self.assertEqual(flow_session.points, Decimal(str(expect_score)))
        else:
            self.assertIsNone(flow_session.points)

    def get_page_submit_history_url_by_ordinal(
            self, page_ordinal, course_identifier=None, flow_session_id=None):
        return self.get_page_view_url_by_ordinal(
            "relate-get_prev_answer_visits_dropdown_content",
            page_ordinal, course_identifier, flow_session_id)

    def get_page_grade_history_url_by_ordinal(
            self, page_ordinal, course_identifier=None, flow_session_id=None):
        return self.get_page_view_url_by_ordinal(
            "relate-get_prev_grades_dropdown_content",
            page_ordinal, course_identifier, flow_session_id)

    def get_page_submit_history_by_ordinal(
            self, page_ordinal, course_identifier=None, flow_session_id=None):
        resp = self.c.get(
            self.get_page_submit_history_url_by_ordinal(
                page_ordinal, course_identifier, flow_session_id),
            HTTP_X_REQUESTED_WITH='XMLHttpRequest')
        return resp

    def get_page_grade_history_by_ordinal(
            self, page_ordinal, course_identifier=None, flow_session_id=None):
        resp = self.c.get(
            self.get_page_grade_history_url_by_ordinal(
                page_ordinal, course_identifier, flow_session_id),
            HTTP_X_REQUESTED_WITH='XMLHttpRequest')
        return resp

    def assertSubmitHistoryItemsCount(  # noqa
            self, page_ordinal, expected_count, course_identifier=None,
            flow_session_id=None):
        resp = self.get_page_submit_history_by_ordinal(
            page_ordinal, course_identifier, flow_session_id)
        import json
        result = json.loads(resp.content.decode())["result"]
        self.assertEqual(len(result), expected_count)

    def assertGradeHistoryItemsCount(  # noqa
            self, page_ordinal, expected_count,
            course_identifier=None,
            flow_session_id=None,
            force_login_instructor=True):

        if course_identifier is None:
            course_identifier = self.get_default_course_identifier()

        if force_login_instructor:
            switch_to = self.get_default_instructor_user(course_identifier)
        else:
            switch_to = self.get_logged_in_user()

        with self.temporarily_switch_to_user(switch_to):
            resp = self.get_page_grade_history_by_ordinal(
                page_ordinal, course_identifier, flow_session_id)

        import json
        result = json.loads(resp.content.decode())["result"]
        self.assertEqual(len(result), expected_count)

    def get_update_course_url(self, course_identifier=None):
        if course_identifier is None:
            course_identifier = self.get_default_course_identifier()
        return reverse("relate-update_course", args=[course_identifier])

    def post_update_course_content(self, commit_sha,
                                   fetch_update=False,
                                   prevent_discarding_revisions=True,
                                   force_login_instructor=True,
                                   course=None,
                                   ):
        # course instead of course_identifier because we need to do
        # refresh_from_db
        course = course or self.get_default_course()

        try:
            commit_sha = commit_sha.decode()
        except Exception:
            pass

        data = {"new_sha": [commit_sha]}

        if not prevent_discarding_revisions:
            data["prevent_discarding_revisions"] = ["on"]

        if not fetch_update:
            data["update"] = ["Update"]
        else:
            data["fetch_update"] = ["Fetch and update"]

        force_login_user = None
        if force_login_instructor:
            force_login_user = self.get_default_instructor_user(course.identifier)

        with self.temporarily_switch_to_user(force_login_user):
            response = self.c.post(
                self.get_update_course_url(course.identifier), data)
            course.refresh_from_db()

        return response

    def get_page_data_by_page_id(
            self, page_id, course_identifier=None, flow_session_id=None):
        flow_params = self.get_flow_params(course_identifier, flow_session_id)
        return FlowPageData.objects.get(
            flow_session_id=flow_params["flow_session_id"], page_id=page_id)

    def get_page_visits(self, course_identifier=None,
                        flow_session_id=None, page_ordinal=None, page_id=None,
                        **kwargs):
        query_kwargs = {}
        if kwargs.get("answer_visit", False):
            query_kwargs.update({"answer__isnull": False})
        flow_params = self.get_flow_params(course_identifier, flow_session_id)
        query_kwargs.update({"flow_session_id": flow_params["flow_session_id"]})
        if page_ordinal is not None:
            query_kwargs.update({"page_data__page_ordinal": page_ordinal})
        elif page_id is not None:
            query_kwargs.update({"page_data__page_id": page_id})
        return FlowPageVisit.objects.filter(**query_kwargs)

    def get_last_answer_visit(self, course_identifier=None,
                              flow_session_id=None, page_ordinal=None,
                              page_id=None, assert_not_none=True):
        result_qset = self.get_page_visits(course_identifier,
                                           flow_session_id, page_ordinal, page_id,
                                           answer_visit=True).order_by('-pk')[:1]
        if result_qset:
            result = result_qset[0]
        else:
            result = None
        if assert_not_none:
            self.assertIsNotNone(result, "The query returns None")
        return result


class SingleCourseTestMixin(CoursesTestMixinBase):
    courses_setup_list = SINGLE_COURSE_SETUP_LIST

    @classmethod
    def setUpTestData(cls):  # noqa
        super(SingleCourseTestMixin, cls).setUpTestData()