test_views.py

from __future__ import annotations


__copyright__ = "Copyright (C) 2017 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 datetime
import unittest

from celery import states, uuid
from django import http
from django.test import RequestFactory, TestCase
from django.test.utils import override_settings
from django.urls import reverse
from django.utils.timezone import now, timedelta

from course import constants, models, views
from relate.celery import app
from relate.utils import as_local_time
from tests import factories
from tests.base_test_mixins import (
    CoursesTestMixinBase,
    HackRepoMixin,
    MockAddMessageMixing,
    SingleCoursePageTestMixin,
    SingleCourseTestMixin,
)
from tests.constants import DATE_TIME_PICKER_TIME_FORMAT
from tests.test_auth import AuthTestMixin
from tests.utils import mock


RELATE_FACILITIES = {
    # intentionally to be different from local_settings_example.py
    "test_center1": {
        "ip_ranges": [
            "192.168.100.0/24",
            ],
        "exams_only": False,
    },
}


class SetFakeTimeTest(SingleCourseTestMixin, TestCase):
    # test views.set_fake_time
    fake_time = datetime.datetime(2038, 12, 31, 0, 0, 0, 0)
    set_fake_time_data = {"time": fake_time.strftime(DATE_TIME_PICKER_TIME_FORMAT),
                          "set": [""]}
    unset_fake_time_data = {"time": set_fake_time_data["time"], "unset": [""]}

    def test_set_fake_time_by_anonymous(self):
        with self.temporarily_switch_to_user(None):
            # the faking url is not rendered in template
            resp = self.client.get(self.course_page_url)
            self.assertNotContains(resp, self.get_fake_time_url())

            resp = self.get_set_fake_time()
            self.assertEqual(resp.status_code, 302)

            resp = self.post_set_fake_time(self.set_fake_time_data, follow=False)
            self.assertEqual(resp.status_code, 302)
            self.assertSessionFakeTimeIsNone(self.client.session)

    def test_set_fake_time_no_pperm(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            # the faking url is not rendered in template
            resp = self.client.get(self.course_page_url)
            self.assertNotContains(resp, self.get_fake_time_url())

            resp = self.get_set_fake_time()
            self.assertEqual(resp.status_code, 403)

            resp = self.post_set_fake_time(self.set_fake_time_data)
            self.assertEqual(resp.status_code, 403)
            self.assertSessionFakeTimeIsNone(self.client.session)

    def test_set_fake_time_by_instructor(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            # the faking url is rendered in template
            resp = self.client.get(self.course_page_url)
            self.assertContains(resp, self.get_fake_time_url())

            resp = self.get_set_fake_time()
            self.assertEqual(resp.status_code, 200)

            # set fake time
            resp = self.post_set_fake_time(self.set_fake_time_data)
            self.assertEqual(resp.status_code, 200)
            self.assertSessionFakeTimeEqual(self.client.session, self.fake_time)

            # revisit the page, just to make sure it works
            resp = self.get_set_fake_time()
            self.assertEqual(resp.status_code, 200)

            # unset fake time
            resp = self.post_set_fake_time(self.unset_fake_time_data)
            self.assertEqual(resp.status_code, 200)
            self.assertSessionFakeTimeIsNone(self.client.session)

    def test_set_fake_time_by_instructor_when_impersonating(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.get_set_fake_time()
            self.assertEqual(resp.status_code, 200)

            self.post_impersonate_view(impersonatee=self.student_participation.user)
            # the faking url is rendered in template
            resp = self.client.get(self.course_page_url)
            self.assertContains(resp, self.get_fake_time_url())

            # set fake time
            resp = self.post_set_fake_time(self.set_fake_time_data)
            self.assertEqual(resp.status_code, 200)
            self.assertSessionFakeTimeEqual(self.client.session, self.fake_time)

            # unset fake time
            resp = self.post_set_fake_time(self.unset_fake_time_data)
            self.assertEqual(resp.status_code, 200)
            self.assertSessionFakeTimeIsNone(self.client.session)

    def test_form_invalid(self):
        with mock.patch("course.views.FakeTimeForm.is_valid") as mock_is_valid:
            mock_is_valid.return_value = False
            with self.temporarily_switch_to_user(self.instructor_participation.user):
                resp = self.post_set_fake_time(self.set_fake_time_data)
                self.assertEqual(resp.status_code, 200)

                # fake failed
                self.assertSessionFakeTimeIsNone(self.client.session)


class GetNowOrFakeTimeTest(unittest.TestCase):
    # test views.get_now_or_fake_time
    mock_now_value = mock.MagicMock()

    def setUp(self):
        fake_get_fake_time = mock.patch("course.views.get_fake_time")
        self.mock_get_fake_time = fake_get_fake_time.start()
        self.addCleanup(fake_get_fake_time.stop)
        fake_now = mock.patch("django.utils.timezone.now")
        self.mock_now = fake_now.start()
        self.mock_now.return_value = self.mock_now_value
        self.addCleanup(fake_now.stop)
        rf = RequestFactory()
        self.request = rf.get("/")

    def test_fake_time_is_none(self):
        self.mock_get_fake_time.return_value = None
        self.assertEqual(
            views.get_now_or_fake_time(self.request), self.mock_now_value)

    def test_fake_time_is_not_none(self):
        mock_fake_time = mock.MagicMock()
        self.mock_get_fake_time.return_value = mock_fake_time
        self.assertEqual(
            views.get_now_or_fake_time(self.request), mock_fake_time)


@override_settings(RELATE_FACILITIES=RELATE_FACILITIES)
class TestSetPretendFacilities(SingleCourseTestMixin, TestCase):
    set_pretend_facilities_data = {
        "facilities": ["test_center1"],
        "custom_facilities": [],
        "add_pretend_facilities_header": ["on"],
        "set": [""]}
    unset_pretend_facilities_data = set_pretend_facilities_data.copy()
    unset_pretend_facilities_data.pop("set")
    unset_pretend_facilities_data["unset"] = [""]

    def test_pretend_facilities_by_anonymous(self):
        with self.temporarily_switch_to_user(None):
            # the pretending url is not rendered in template
            resp = self.client.get(self.course_page_url)
            self.assertNotContains(resp, self.get_set_pretend_facilities_url())

            resp = self.get_set_pretend_facilities()
            self.assertEqual(resp.status_code, 302)

            resp = self.post_set_pretend_facilities(
                self.set_pretend_facilities_data, follow=False)
            self.assertEqual(resp.status_code, 302)
            self.assertSessionPretendFacilitiesIsNone(self.client.session)

    def test_pretend_facilities_no_pperm(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            # the pretending url is not rendered in template
            resp = self.client.get(self.course_page_url)
            self.assertNotContains(resp, self.get_set_pretend_facilities_url())

            resp = self.get_set_pretend_facilities()
            self.assertEqual(resp.status_code, 403)

            resp = self.post_set_pretend_facilities(
                self.set_pretend_facilities_data)
            self.assertEqual(resp.status_code, 403)
            self.assertSessionPretendFacilitiesIsNone(self.client.session)

    def test_pretend_facilities_by_instructor(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            # the pretending url is rendered in template
            resp = self.client.get(self.course_page_url)
            self.assertContains(resp, self.get_set_pretend_facilities_url())

            resp = self.get_set_pretend_facilities()
            self.assertEqual(resp.status_code, 200)

            resp = self.post_set_pretend_facilities(
                self.set_pretend_facilities_data)
            self.assertEqual(resp.status_code, 200)
            self.assertSessionPretendFacilitiesContains(self.client.session,
                                                        "test_center1")

            # revisit the page, just to make sure it works
            resp = self.get_set_pretend_facilities()
            self.assertEqual(resp.status_code, 200)

            resp = self.post_set_pretend_facilities(
                self.unset_pretend_facilities_data)
            self.assertEqual(resp.status_code, 200)
            self.assertSessionPretendFacilitiesIsNone(self.client.session)

    def test_pretend_facilities_by_instructor_when_impersonating(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):

            self.post_impersonate_view(impersonatee=self.student_participation.user)

            # the pretending url is rendered in template
            resp = self.client.get(self.course_page_url)
            self.assertContains(resp, self.get_set_pretend_facilities_url())

            resp = self.get_set_pretend_facilities()
            self.assertEqual(resp.status_code, 200)

            resp = self.post_set_pretend_facilities(
                self.set_pretend_facilities_data)
            self.assertEqual(resp.status_code, 200)
            self.assertSessionPretendFacilitiesContains(self.client.session,
                                                        "test_center1")

            resp = self.post_set_pretend_facilities(
                self.unset_pretend_facilities_data)
            self.assertEqual(resp.status_code, 200)
            self.assertSessionPretendFacilitiesIsNone(self.client.session)

    def test_form_invalid(self):
        with mock.patch("course.views.FakeFacilityForm.is_valid") as mock_is_valid:
            mock_is_valid.return_value = False
            with self.temporarily_switch_to_user(self.instructor_participation.user):
                resp = self.post_set_pretend_facilities(
                    self.set_pretend_facilities_data)
                self.assertEqual(resp.status_code, 200)

                # pretending failed
                self.assertSessionPretendFacilitiesIsNone(self.client.session)


class TestEditCourse(SingleCourseTestMixin, MockAddMessageMixing, TestCase):

    def setUp(self):
        super().setUp()
        self.rf = RequestFactory()

    def test_non_auth_edit_get(self):
        with self.temporarily_switch_to_user(None):
            resp = self.get_edit_course()
        self.assertTrue(resp.status_code, 404)

    def test_non_auth_edit_post(self):
        with self.temporarily_switch_to_user(None):
            resp = self.post_edit_course(data={})
        self.assertTrue(resp.status_code, 404)

    def test_student_edit_get(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_edit_course()
        self.assertTrue(resp.status_code, 404)

    def test_student_edit_post(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.post_edit_course(data={})
        self.assertTrue(resp.status_code, 404)

    def test_instructor_edit_get(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.get_edit_course()
        self.assertTrue(resp.status_code, 200)

    def test_set_up_new_course_form_invalid(self):
        for field_name in ["name", "number", "time_period",
                           "git_source", "from_email", "notify_email"]:
            form_data = self.copy_course_dict_and_set_attrs_for_post()
            del form_data[field_name]
            request = self.rf.post(self.get_set_up_new_course_url(), data=form_data)
            request.user = self.instructor_participation.user
            form = views.EditCourseForm(request.POST)
            self.assertFalse(form.is_valid())

    def test_instructor_edit_post_unchanged(self):
        # test when form data is the same with current instance,
        # the message shows "no change"
        with mock.patch("course.views.EditCourseForm.is_valid") as mock_is_valid, \
            mock.patch("course.views.EditCourseForm.has_changed") as mock_changed, \
            mock.patch("course.views.render_course_page"), \
                mock.patch("course.views._") as mock_gettext:

            mock_is_valid.return_value = True
            mock_changed.return_value = False
            mock_gettext.side_effect = lambda x: x
            request = self.rf.post(self.get_edit_course_url(),
                                   data=mock.MagicMock())
            request.user = self.instructor_participation.user
            course_identifier = self.get_default_course_identifier()
            resp = views.edit_course(request, course_identifier)
            self.assertTrue(resp.status_code, 200)
            self.assertAddMessageCallCount(1)
            self.assertAddMessageCalledWith(
                "No change was made on the settings.")

    def test_instructor_edit_post_saved(self):
        # test when form data is_valid and different with the current instance,
        # the message shows "success"
        with mock.patch("course.views.EditCourseForm.is_valid") as mock_is_valid, \
            mock.patch("course.views.EditCourseForm.has_changed") as mock_changed, \
            mock.patch("course.views.EditCourseForm.save")as mock_save, \
            mock.patch("course.views.render_course_page"), \
                mock.patch("course.views._") as mock_gettext:

            mock_save.return_value = self.course
            mock_is_valid.return_value = True
            mock_changed.return_value = True
            mock_gettext.side_effect = lambda x: x
            request = self.rf.post(self.get_edit_course_url(),
                                   data=mock.MagicMock())
            request.user = self.instructor_participation.user
            course_identifier = self.get_default_course_identifier()
            resp = views.edit_course(request, course_identifier)
            self.assertTrue(resp.status_code, 200)
            self.assertAddMessageCallCount(1)
            self.assertAddMessageCalledWith(
                "Successfully updated course settings.")

    @override_settings(LANGUAGES=(("en-us", "English"),))
    def test_instructor_edit_post_saved_default(self):
        # test when form is valid, the message show success
        self.course.force_lang = "en-us"
        self.course.save()
        data = self.copy_course_dict_and_set_attrs_for_post({"force_lang": ""})
        with mock.patch("course.views.EditCourseForm.save") as mock_save, \
            mock.patch("course.views.render_course_page"), \
                mock.patch("course.views._") as mock_gettext:

            mock_gettext.side_effect = lambda x: x
            mock_save.return_value = self.course
            request = self.rf.post(self.get_edit_course_url(),
                                   data=data)
            request.user = self.instructor_participation.user
            course_identifier = self.get_default_course_identifier()
            resp = views.edit_course(request, course_identifier)
            self.assertTrue(resp.status_code, 200)
            self.assertTrue(mock_save.call_count, 1)
            self.assertAddMessageCallCount(1)
            self.assertAddMessageCalledWith(
                "Successfully updated course settings.")

    @override_settings(LANGUAGES=(("en-us", "English"),))
    def test_instructor_edit_db_saved_default(self):
        # test force_lang can be an empty string (default value)
        self.course.force_lang = "en-us"
        self.course.save()

        self.course.force_lang = ""
        self.course.save()
        self.assertEqual(self.course.force_lang, "")

    def test_instructor_post_save_spaces_as_force_lang(self):
        # current force_lang is "", testing that the save won't occur
        data = self.copy_course_dict_and_set_attrs_for_post({"force_lang": "   "})
        with mock.patch("course.views.EditCourseForm.save") as mock_form_save, \
            mock.patch("course.views.render_course_page"), \
                mock.patch("course.views._") as mock_gettext:

            mock_gettext.side_effect = lambda x: x
            request = self.rf.post(self.get_edit_course_url(),
                                   data=data)
            request.user = self.instructor_participation.user
            course_identifier = self.get_default_course_identifier()
            resp = views.edit_course(request, course_identifier)
            self.assertEqual(mock_form_save.call_count, 0)
            self.assertTrue(resp.status_code, 200)
            self.assertAddMessageCallCount(1)
            self.assertAddMessageCalledWith(
                "No change was made on the settings.")

    def test_instructor_db_save_spaces_as_force_lang(self):
        # current force_lang is "", testing that the force_lang is still ""
        self.course.force_lang = "   "
        self.course.save()
        self.course.refresh_from_db()
        self.assertEqual(len(self.course.force_lang), 0)

    def test_instructor_edit_post_form_invalid(self):
        with mock.patch("course.views.EditCourseForm.is_valid") as mock_is_valid, \
            mock.patch("course.views.render_course_page"), \
                mock.patch("course.views._") as mock_gettext:

            mock_is_valid.return_value = False
            mock_gettext.side_effect = lambda x: x
            request = self.rf.post(self.get_edit_course_url(),
                                   data=mock.MagicMock())
            request.user = self.instructor_participation.user
            course_identifier = self.get_default_course_identifier()
            resp = views.edit_course(request, course_identifier)
            self.assertTrue(resp.status_code, 200)
            self.assertAddMessageCallCount(1)
            self.assertAddMessageCalledWith(
                "Failed to update course settings.")

    def test_instructor_db_save_invalid_force_lang(self):
        # test db save failure
        self.course.force_lang = "invalid_lang"
        from django.core.exceptions import ValidationError
        with self.assertRaises(ValidationError):
            self.course.save()


class GenerateSshKeypairTest(CoursesTestMixinBase, AuthTestMixin, TestCase):
    def get_generate_ssh_keypair_url(self):
        return reverse("relate-generate_ssh_keypair")

    def test_anonymous(self):
        with self.temporarily_switch_to_user(None):
            resp = self.client.get(self.get_generate_ssh_keypair_url())
            self.assertEqual(resp.status_code, 302)

            expected_redirect_url = self.get_sign_in_choice_url(
                redirect_to=self.get_generate_ssh_keypair_url())

            self.assertRedirects(resp, expected_redirect_url,
                                 fetch_redirect_response=False)

    def test_not_staff(self):
        user = factories.UserFactory()
        assert not user.is_staff
        with self.temporarily_switch_to_user(user):
            resp = self.client.get(self.get_generate_ssh_keypair_url())
            self.assertEqual(resp.status_code, 403)

    def test_success(self):
        user = factories.UserFactory()
        user.is_staff = True
        user.save()
        with self.temporarily_switch_to_user(user):
            resp = self.client.get(self.get_generate_ssh_keypair_url())
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextContains(
                resp, "public_key",
                ["-----BEGIN RSA PRIVATE KEY-----",
                 "-----END RSA PRIVATE KEY-----"], in_bulk=True)
            self.assertResponseContextContains(
                resp, "private_key",
                ["ssh-rsa", "relate-course-key"], in_bulk=True)


class HomeTest(CoursesTestMixinBase, TestCase):
    # test views.home

    def test(self):
        course1 = factories.CourseFactory(hidden=False)
        course2 = factories.CourseFactory(
            identifier="course2", hidden=True)
        course3 = factories.CourseFactory(listed=False,
            identifier="course3", hidden=False)
        course4 = factories.CourseFactory(
            identifier="course4", hidden=False, end_date=now() - timedelta(days=1))

        user = factories.UserFactory()
        factories.ParticipationFactory(
            course=course1, user=user, roles=["instructor"])
        factories.ParticipationFactory(
            course=course2, user=user, roles=["instructor"])
        factories.ParticipationFactory(
            course=course3, user=user, roles=["instructor"])

        with self.temporarily_switch_to_user(None):
            resp = self.client.get("/")
        self.assertResponseContextEqual(resp, "current_courses", [course1])
        self.assertResponseContextEqual(resp, "past_courses", [course4])

        with self.temporarily_switch_to_user(user):
            resp = self.client.get("/")
            self.assertResponseContextEqual(
                resp, "current_courses", [course1, course2])
            self.assertResponseContextEqual(resp, "past_courses", [course4])


class CheckCourseStateTest(SingleCourseTestMixin, TestCase):
    # test views.check_course_state
    def test_course_not_hidden(self):
        views.check_course_state(self.course, None)
        views.check_course_state(self.course, self.student_participation)
        views.check_course_state(self.course, self.ta_participation)
        views.check_course_state(self.course, self.instructor_participation)

    def test_course_hidden(self):
        self.course.hidden = True
        self.course.save()
        with self.assertRaises(views.PermissionDenied):
            views.check_course_state(self.course, None)

        with self.assertRaises(views.PermissionDenied):
            views.check_course_state(self.course, self.student_participation)

        views.check_course_state(self.course, self.ta_participation)
        views.check_course_state(self.course, self.instructor_participation)


class StaticPageTest(SingleCourseTestMixin, TestCase):
    # test views.static_page
    def get_static_page_url(self, page_path, course_identifier=None):
        course_identifier = course_identifier or self.get_default_course_identifier()
        return reverse("relate-content_page",
                       kwargs={"course_identifier": course_identifier,
                               "page_path": page_path})

    def get_static_page(self, page_path, course_identifier=None):
        return self.client.get(
                self.get_static_page_url(page_path, course_identifier))

    def test_success(self):
        resp = self.get_static_page("test")
        self.assertEqual(resp.status_code, 200)
        self.assertContains(
            resp, "<h1>Demo page</h1>", html=True)
        self.assertContains(
            resp, "I am just a simple demo page. Go back to the "
                  '<a href="/course/test-course/">course page</a>?')

    def test_404(self):
        resp = self.get_static_page("hello")
        self.assertEqual(resp.status_code, 404)


class CoursePageTest(SingleCourseTestMixin, MockAddMessageMixing, TestCase):
    # test views.course_page

    # {{{ test show enroll button
    def test_student_no_enroll_button(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.client.get(self.get_course_page_url())
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextEqual(resp, "show_enroll_button", False)

    def test_anonymous_show_enroll_button(self):
        with self.temporarily_switch_to_user(None):
            resp = self.client.get(self.get_course_page_url())
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextEqual(resp, "show_enroll_button", True)

    def test_non_participation_show_enroll_button(self):
        user = factories.UserFactory()
        with self.temporarily_switch_to_user(user):
            resp = self.client.get(self.get_course_page_url())
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextEqual(resp, "show_enroll_button", True)

    def test_requested_not_show_enroll_button(self):
        requested = factories.ParticipationFactory(
            course=self.course, status=constants.participation_status.requested)
        with self.temporarily_switch_to_user(requested.user):
            resp = self.client.get(self.get_course_page_url())
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextEqual(resp, "show_enroll_button", False)

        self.assertAddMessageCallCount(1)
        self.assertAddMessageCalledWith(
            "Your enrollment request is pending. You will be "
            "notified once it has been acted upon.")

    def test_requested_hint_for_set_instid(self):
        requested = factories.ParticipationFactory(
            course=self.course, status=constants.participation_status.requested)
        factories.ParticipationPreapprovalFactory(
            course=self.course, institutional_id="inst_id1234")
        with self.temporarily_switch_to_user(requested.user):
            resp = self.client.get(self.get_course_page_url())
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextEqual(resp, "show_enroll_button", False)

        self.assertAddMessageCallCount(2)
        self.assertAddMessageCalledWith(
            "Your institutional ID is not verified or "
            "preapproved. Please contact your course "
            "staff.", reset=True)

        # remove course verify inst_id requirements
        self.course.preapproval_require_verified_inst_id = False
        self.course.save()

        with self.temporarily_switch_to_user(requested.user):
            resp = self.client.get(self.get_course_page_url())
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextEqual(resp, "show_enroll_button", False)

        self.assertAddMessageCallCount(1, reset=True)

        # remove user inst_id
        requested.user.institutional_id = ""
        requested.user.save()
        with self.temporarily_switch_to_user(requested.user):
            resp = self.client.get(self.get_course_page_url())
            self.assertEqual(resp.status_code, 200)
            self.assertResponseContextEqual(resp, "show_enroll_button", False)

        self.assertAddMessageCallCount(2)
        self.assertAddMessageCalledWith(
            "This course uses institutional ID for enrollment preapproval, "
            "please <a href='/profile/?referer=/course/test-course/"
            "&set_inst_id=1' role='button' class='btn btn-md btn-primary'>"
            "fill in your institutional ID &nbsp;&raquo;</a> in your profile.")

    # }}}


class GetMediaTest(SingleCourseTestMixin, TestCase):
    # test views.get_media
    # currently only mock test, because there's no media in the sample repo

    def get_media_url(self, media_path, course_identifier=None, commit_sha=None):
        course_identifier = course_identifier or self.get_default_course_identifier()
        if commit_sha is None:
            commit_sha = self.course.active_git_commit_sha

        return reverse("relate-get_media",
                       kwargs={"course_identifier": course_identifier,
                               "commit_sha": commit_sha,
                               "media_path": media_path})

    def get_media_view(self, media_path, course_identifier=None, commit_sha=None):
        return self.client.get(
            self.get_media_url(media_path, course_identifier, commit_sha))

    def test(self):
        resp = self.get_media_view("foo.jpg")
        self.assertEqual(resp.status_code, 404)

    def test_func_call(self):
        with mock.patch(
                "course.views.get_repo_file_response") as mock_get_repo_file_resp:
            mock_get_repo_file_resp.return_value = http.HttpResponse("hi")
            resp = self.get_media_view("foo.jpg")
            self.assertEqual(resp.status_code, 200)
            self.assertEqual(mock_get_repo_file_resp.call_count, 1)

    def test_course_does_not_exist_404(self):
        with mock.patch(
                "course.views.get_repo_file_response") as mock_get_repo_file_resp:
            mock_get_repo_file_resp.return_value = http.HttpResponse("hi")
            resp = self.get_media_view("foo.jpg", course_identifier="no-course")
            self.assertEqual(resp.status_code, 404)
            self.assertEqual(mock_get_repo_file_resp.call_count, 0)


class GetRepoFileTestMixin(SingleCourseTestMixin):
    force_login_student_for_each_test = True

    # test views.get_repo_file and  views.get_current_repo_file
    def get_repo_file_url(self, path, course_identifier=None, commit_sha=None):
        course_identifier = course_identifier or self.get_default_course_identifier()
        if commit_sha is None:
            commit_sha = self.course.active_git_commit_sha

        return reverse("relate-get_repo_file",
                       kwargs={"course_identifier": course_identifier,
                               "commit_sha": commit_sha,
                               "path": path})

    def get_repo_file_view(self, path, course_identifier=None, commit_sha=None):
        return self.client.get(
            self.get_repo_file_url(path, course_identifier, commit_sha))

    def get_current_repo_file_url(self, path, course_identifier=None):
        course_identifier = course_identifier or self.get_default_course_identifier()
        return reverse("relate-get_current_repo_file",
                       kwargs={"course_identifier": course_identifier,
                               "path": path})

    def get_current_repo_file_view(self, path, course_identifier=None):
        return self.client.get(
            self.get_current_repo_file_url(path, course_identifier))


class GetRepoFileTest(GetRepoFileTestMixin, TestCase):
    # test views.get_repo_file
    def test_file_not_exist(self):

        repo_file = "images/file_not_exist.png"
        tup = ((None, 404),
               (self.student_participation.user, 404),
               (self.ta_participation.user, 404),
               (self.instructor_participation.user, 404))
        for user, status_code in tup:
            with self.subTest(user=user):
                with self.temporarily_switch_to_user(user):
                    resp = self.get_repo_file_view(repo_file)
                    self.assertEqual(resp.status_code, status_code)

                    resp = self.get_current_repo_file_view(repo_file)
                    self.assertEqual(resp.status_code, status_code)

    def test_commit_sha_not_exist(self):
        repo_file = "images/django-logo.png"
        tup = ((None, 403),
               (self.student_participation.user, 403),
               (self.ta_participation.user, 403),
               (self.instructor_participation.user, 403))
        for user, status_code in tup:
            with self.subTest(user=user):
                with self.temporarily_switch_to_user(user):
                    resp = self.get_repo_file_view(repo_file, commit_sha="123abc")
                    self.assertEqual(resp.status_code, status_code)

    def test_content_type(self):
        tup = (
            ("images/django-logo.png", "image/png"),
            ("images/classroom.jpeg", "image/jpeg"),
            ("pdfs/sample.pdf", "application/pdf"),
        )
        for repo_file, content_type in tup:
            with self.subTest(repo_file=repo_file):
                resp = self.get_repo_file_view(repo_file)
                self.assertEqual(resp.status_code, 200)
                self.assertEqual(resp["Content-Type"], content_type)

                resp = self.get_current_repo_file_view(repo_file)
                self.assertEqual(resp.status_code, 200)
                self.assertEqual(resp["Content-Type"], content_type)


class GetRepoFileTestMocked(GetRepoFileTestMixin, HackRepoMixin, TestCase):
    """
    Test views.get_repo_file, with get_repo_blob mocked as class level,
    the purpose is to test role permissions to repo files

        in_exam:
        - "*.jpeg"

        ta:
        - "django-logo.png"

    """

    initial_commit_sha = "abcdef001"

    def test_accessible_by_ta_and_above_fullname(self):
        repo_file = "images/django-logo.png"
        tup = ((None, 403),
               (self.student_participation.user, 403),
               (self.ta_participation.user, 200),
               (self.instructor_participation.user, 200))
        for user, status_code in tup:
            with self.subTest(user=user):
                with self.temporarily_switch_to_user(user):
                    resp = self.get_repo_file_view(repo_file)
                    self.assertEqual(resp.status_code, status_code)

                    resp = self.get_current_repo_file_view(repo_file)
                    self.assertEqual(resp.status_code, status_code)

    def test_accessible_in_exam(self):
        repo_file = "images/classroom.jpeg"
        tup = ((None, 403),
               (self.student_participation.user, 403),
               (self.ta_participation.user, 200),
               (self.instructor_participation.user, 200))
        for user, status_code in tup:
            with self.subTest(user=user):
                with self.temporarily_switch_to_user(user):
                    resp = self.get_repo_file_view(repo_file)
                    self.assertEqual(resp.status_code, status_code)

                    resp = self.get_current_repo_file_view(repo_file)
                    self.assertEqual(resp.status_code, status_code)

    def test_in_exam(self):
        req = RequestFactory()
        repo_file = "images/classroom.jpeg"
        request = req.get(self.get_repo_file_url(repo_file))
        request.relate_exam_lockdown = True

        from django.contrib.auth.models import AnonymousUser
        users = (AnonymousUser(),
                 self.student_participation.user,
                 self.ta_participation.user,
                 self.instructor_participation.user)

        for user in users:
            request.user = user
            with self.subTest(user=user):
                response = views.get_repo_file(
                    request, self.course.identifier,
                    self.course.active_git_commit_sha, repo_file)
                self.assertEqual(response.status_code, 200)


class ManageInstantFlowRequestsTest(SingleCoursePageTestMixin, TestCase):
    # test views.manage_instant_flow_requests

    def get_manage_instant_flow_requests_url(self, course_identifier=None):
        return reverse(
            "relate-manage_instant_flow_requests",
            kwargs={"course_identifier":
                        course_identifier or self.get_default_course_identifier()})

    def get_manage_instant_flow_requests_view(
            self, course_identifier=None, force_login_instructor=True):
        course_identifier or self.get_default_course_identifier()

        if not force_login_instructor:
            u = self.get_logged_in_user()
        else:
            u = self.instructor_participation.user
        with self.temporarily_switch_to_user(u):
            return self.client.get(
                self.get_manage_instant_flow_requests_url(course_identifier))

    def post_manage_instant_flow_requests_view(
            self, data, course_identifier=None, force_login_instructor=True):
        course_identifier or self.get_default_course_identifier()

        if not force_login_instructor:
            u = self.get_logged_in_user()
        else:
            u = self.instructor_participation.user
        with self.temporarily_switch_to_user(u):
            return self.client.post(
                self.get_manage_instant_flow_requests_url(course_identifier),
                data=data)

    def get_default_post_data(self, action="add", **kwargs):
        data = {
            "flow_id": self.flow_id,
            "duration_in_minutes": 20,
            action: ""
        }
        data.update(kwargs)
        return data

    def test_anonymous(self):
        with self.temporarily_switch_to_user(None):
            resp = self.get_manage_instant_flow_requests_view(
                force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)

            resp = self.post_manage_instant_flow_requests_view(
                data=self.get_default_post_data(), force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)

            self.assertEqual(models.InstantFlowRequest.objects.count(), 0)

    def test_student(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.get_manage_instant_flow_requests_view(
                force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)

            resp = self.post_manage_instant_flow_requests_view(
                data=self.get_default_post_data(), force_login_instructor=False)
            self.assertEqual(resp.status_code, 403)

            self.assertEqual(models.InstantFlowRequest.objects.count(), 0)

    def test_add(self):
        resp = self.get_manage_instant_flow_requests_view()
        self.assertEqual(resp.status_code, 200)

        resp = self.post_manage_instant_flow_requests_view(
            data=self.get_default_post_data())
        self.assertEqual(resp.status_code, 200)

        self.assertEqual(models.InstantFlowRequest.objects.count(), 1)
        self.assertEqual(
            models.InstantFlowRequest.objects.filter(
                cancelled=False).count(), 1)

    def test_cancel(self):
        # 2 cancellable
        factories.InstantFlowRequestFactory(
            course=self.course, flow_id=self.flow_id,
            start_time=now() - timedelta(minutes=5),
            end_time=now() + timedelta(minutes=1))

        factories.InstantFlowRequestFactory(
            course=self.course, flow_id=self.flow_id,
            start_time=now() - timedelta(minutes=15),
            end_time=now() + timedelta(minutes=10))

        # not started
        inr3 = factories.InstantFlowRequestFactory(
            course=self.course, flow_id=self.flow_id,
            start_time=now() + timedelta(minutes=15),
            end_time=now() + timedelta(minutes=35))

        # expired
        inr4 = factories.InstantFlowRequestFactory(
            course=self.course, flow_id=self.flow_id,
            start_time=now() - timedelta(minutes=15),
            end_time=now() - timedelta(minutes=3))

        resp = self.post_manage_instant_flow_requests_view(
            data=self.get_default_post_data(action="cancel"))
        self.assertEqual(resp.status_code, 200)

        self.assertEqual(models.InstantFlowRequest.objects.count(), 4)
        self.assertEqual(
            models.InstantFlowRequest.objects.filter(
                cancelled=True).count(), 2)

        inr3.refresh_from_db()
        self.assertFalse(inr3.cancelled)
        inr4.refresh_from_db()
        self.assertFalse(inr4.cancelled)

    def test_form_invalid(self):
        with mock.patch(
                "course.views.InstantFlowRequestForm.is_valid") as mock_is_valid:
            mock_is_valid.return_value = False

            with self.temporarily_switch_to_user(self.instructor_participation.user):
                # add
                resp = self.post_manage_instant_flow_requests_view(
                    data=self.get_default_post_data())
                self.assertEqual(resp.status_code, 200)

                self.assertEqual(models.InstantFlowRequest.objects.count(), 0)

                factories.InstantFlowRequestFactory(
                    course=self.course, flow_id=self.flow_id,
                    start_time=now() - timedelta(minutes=5),
                    end_time=now() + timedelta(minutes=1))

                resp = self.post_manage_instant_flow_requests_view(
                    data=self.get_default_post_data(action="cancel"))
                self.assertEqual(resp.status_code, 200)

                self.assertEqual(models.InstantFlowRequest.objects.count(), 1)
                self.assertEqual(
                    models.InstantFlowRequest.objects.filter(
                        cancelled=True).count(), 0)

    def test_invalid_operation(self):
        resp = self.post_manage_instant_flow_requests_view(
            data=self.get_default_post_data(action="unknown"))
        self.assertEqual(resp.status_code, 400)

        self.assertEqual(models.InstantFlowRequest.objects.count(), 0)

        factories.InstantFlowRequestFactory(
            course=self.course, flow_id=self.flow_id,
            start_time=now() - timedelta(minutes=5),
            end_time=now() + timedelta(minutes=1))

        resp = self.post_manage_instant_flow_requests_view(
            data=self.get_default_post_data(action="unknown"))
        self.assertEqual(resp.status_code, 400)

        self.assertEqual(models.InstantFlowRequest.objects.count(), 1)
        self.assertEqual(
            models.InstantFlowRequest.objects.filter(
                cancelled=True).count(), 0)


class TestFlowTest(SingleCoursePageTestMixin, TestCase):
    # test views.test_flow

    def get_test_flow_url(self, course_identifier=None):
        return reverse(
            "relate-test_flow",
            kwargs={"course_identifier":
                        course_identifier or self.get_default_course_identifier()})

    def get_test_flow_view(
            self, course_identifier=None, force_login_instructor=True):