test_sysadmin.py

"""
Provide tests for sysadmin dashboard feature in sysadmin.py
"""
import glob
import os
import re
import shutil
import unittest
from datetime import datetime
from uuid import uuid4

import mongoengine
from django.conf import settings
from django.core.urlresolvers import reverse
from django.test.client import Client
from django.test.utils import override_settings
from pytz import UTC
from nose.plugins.attrib import attr
from opaque_keys.edx.keys import CourseKey

from dashboard.git_import import GitImportErrorNoDir
from dashboard.models import CourseImportLog
from student.roles import CourseStaffRole, GlobalStaff
from student.tests.factories import UserFactory
from util.date_utils import DEFAULT_DATE_TIME_FORMAT, get_time_display
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
from xmodule.modulestore.tests.mongo_connection import MONGO_HOST, MONGO_PORT_NUM

TEST_MONGODB_LOG = {
    'host': MONGO_HOST,
    'port': MONGO_PORT_NUM,
    'user': '',
    'password': '',
    'db': 'test_xlog',
}

FEATURES_WITH_SSL_AUTH = settings.FEATURES.copy()
FEATURES_WITH_SSL_AUTH['AUTH_USE_CERTIFICATES'] = True


class SysadminBaseTestCase(SharedModuleStoreTestCase):
    """
    Base class with common methods used in XML and Mongo tests
    """

    TEST_REPO = 'https://github.com/mitocw/edx4edx_lite.git'
    TEST_BRANCH = 'testing_do_not_delete'
    TEST_BRANCH_COURSE = CourseKey.from_string('MITx/edx4edx_branch/edx4edx')

    def setUp(self):
        """Setup test case by adding primary user."""
        super(SysadminBaseTestCase, self).setUp()
        self.user = UserFactory.create(username='test_user',
                                       email='test_user+sysadmin@edx.org',
                                       password='foo')
        self.client = Client()

    def _setstaff_login(self):
        """Makes the test user staff and logs them in"""
        GlobalStaff().add_users(self.user)
        self.client.login(username=self.user.username, password='foo')

    def _add_edx4edx(self, branch=None):
        """Adds the edx4edx sample course"""
        post_dict = {'repo_location': self.TEST_REPO, 'action': 'add_course', }
        if branch:
            post_dict['repo_branch'] = branch
        return self.client.post(reverse('sysadmin_courses'), post_dict)

    def _rm_edx4edx(self):
        """Deletes the sample course from the XML store"""
        def_ms = modulestore()
        course_path = '{0}/edx4edx_lite'.format(
            os.path.abspath(settings.DATA_DIR))
        try:
            # using XML store
            course = def_ms.courses.get(course_path, None)
        except AttributeError:
            # Using mongo store
            course = def_ms.get_course(CourseKey.from_string('MITx/edx4edx/edx4edx'))

        # Delete git loaded course
        response = self.client.post(
            reverse('sysadmin_courses'),
            {
                'course_id': course.id.to_deprecated_string(),
                'action': 'del_course',
            }
        )
        self.addCleanup(self._rm_glob, '{0}_deleted_*'.format(course_path))

        return response

    def _rm_glob(self, path):
        """
        Create a shell expansion of passed in parameter and iteratively
        remove them.  Must only expand to directories.
        """
        for path in glob.glob(path):
            shutil.rmtree(path)

    def _mkdir(self, path):
        """
        Create directory and add the cleanup for it.
        """
        os.mkdir(path)
        self.addCleanup(shutil.rmtree, path)


@attr(shard=1)
@override_settings(
    MONGODB_LOG=TEST_MONGODB_LOG,
    GIT_REPO_DIR=settings.TEST_ROOT / "course_repos_{}".format(uuid4().hex)
)
@unittest.skipUnless(settings.FEATURES.get('ENABLE_SYSADMIN_DASHBOARD'),
                     "ENABLE_SYSADMIN_DASHBOARD not set")
class TestSysAdminMongoCourseImport(SysadminBaseTestCase):
    """
    Check that importing into the mongo module store works
    """

    @classmethod
    def tearDownClass(cls):
        """Delete mongo log entries after test."""
        super(TestSysAdminMongoCourseImport, cls).tearDownClass()
        try:
            mongoengine.connect(TEST_MONGODB_LOG['db'])
            CourseImportLog.objects.all().delete()
        except mongoengine.connection.ConnectionError:
            pass

    def _setstaff_login(self):
        """
        Makes the test user staff and logs them in
        """

        self.user.is_staff = True
        self.user.save()

        self.client.login(username=self.user.username, password='foo')

    def test_missing_repo_dir(self):
        """
        Ensure that we handle a missing repo dir
        """

        self._setstaff_login()

        if os.path.isdir(settings.GIT_REPO_DIR):
            shutil.rmtree(settings.GIT_REPO_DIR)

        # Create git loaded course
        response = self._add_edx4edx()
        self.assertIn(GitImportErrorNoDir(settings.GIT_REPO_DIR).message,
                      response.content.decode('UTF-8'))

    def test_mongo_course_add_delete(self):
        """
        This is the same as TestSysadmin.test_xml_course_add_delete,
        but it uses a mongo store
        """

        self._setstaff_login()
        self._mkdir(settings.GIT_REPO_DIR)

        def_ms = modulestore()
        self.assertNotEqual('xml', def_ms.get_modulestore_type(None))

        self._add_edx4edx()
        course = def_ms.get_course(CourseKey.from_string('MITx/edx4edx/edx4edx'))
        self.assertIsNotNone(course)

        self._rm_edx4edx()
        course = def_ms.get_course(CourseKey.from_string('MITx/edx4edx/edx4edx'))
        self.assertIsNone(course)

    def test_course_info(self):
        """
        Check to make sure we are getting git info for courses
        """
        # Regex of first 3 columns of course information table row for
        # test course loaded from git. Would not have sha1 if
        # git_info_for_course failed.
        table_re = re.compile(r"""
            <tr>\s+
            <td>edX\sAuthor\sCourse</td>\s+  # expected test git course name
            <td>MITx/edx4edx/edx4edx</td>\s+  # expected test git course_id
            <td>[a-fA-F\d]{40}</td>  # git sha1 hash
        """, re.VERBOSE)

        self._setstaff_login()
        self._mkdir(settings.GIT_REPO_DIR)

        # Make sure we don't have any git hashes on the page
        response = self.client.get(reverse('sysadmin_courses'))
        self.assertNotRegexpMatches(response.content, table_re)

        # Now add the course and make sure it does match
        response = self._add_edx4edx()
        self.assertRegexpMatches(response.content, table_re)

    def test_gitlogs(self):
        """
        Create a log entry and make sure it exists
        """

        self._setstaff_login()
        self._mkdir(settings.GIT_REPO_DIR)

        self._add_edx4edx()
        response = self.client.get(reverse('gitlogs'))

        # Check that our earlier import has a log with a link to details
        self.assertIn('/gitlogs/MITx/edx4edx/edx4edx', response.content)

        response = self.client.get(
            reverse('gitlogs_detail', kwargs={
                'course_id': 'MITx/edx4edx/edx4edx'}))

        self.assertIn('======&gt; IMPORTING course',
                      response.content)

        self._rm_edx4edx()

    def test_gitlog_date(self):
        """
        Make sure the date is timezone-aware and being converted/formatted
        properly.
        """

        tz_names = [
            'America/New_York',  # UTC - 5
            'Asia/Pyongyang',    # UTC + 9
            'Europe/London',     # UTC
            'Canada/Yukon',      # UTC - 8
            'Europe/Moscow',     # UTC + 4
        ]
        tz_format = DEFAULT_DATE_TIME_FORMAT

        self._setstaff_login()
        self._mkdir(settings.GIT_REPO_DIR)

        self._add_edx4edx()
        date = CourseImportLog.objects.first().created.replace(tzinfo=UTC)

        for timezone in tz_names:
            with (override_settings(TIME_ZONE=timezone)):
                date_text = get_time_display(date, tz_format, settings.TIME_ZONE)
                response = self.client.get(reverse('gitlogs'))
                self.assertIn(date_text, response.content.decode('UTF-8'))

        self._rm_edx4edx()

    def test_gitlog_bad_course(self):
        """
        Make sure we gracefully handle courses that don't exist.
        """
        self._setstaff_login()
        response = self.client.get(
            reverse('gitlogs_detail', kwargs={
                'course_id': 'Not/Real/Testing'}))
        self.assertEqual(404, response.status_code)

    def test_gitlog_no_logs(self):
        """
        Make sure the template behaves well when rendered despite there not being any logs.
        (This is for courses imported using methods other than the git_add_course command)
        """

        self._setstaff_login()
        self._mkdir(settings.GIT_REPO_DIR)

        self._add_edx4edx()

        # Simulate a lack of git import logs
        import_logs = CourseImportLog.objects.all()
        import_logs.delete()

        response = self.client.get(
            reverse('gitlogs_detail', kwargs={
                'course_id': 'MITx/edx4edx/edx4edx'
            })
        )
        self.assertIn(
            'No git import logs have been recorded for this course.',
            response.content
        )

        self._rm_edx4edx()

    def test_gitlog_pagination_out_of_range_invalid(self):
        """
        Make sure the pagination behaves properly when the requested page is out
        of range.
        """

        self._setstaff_login()

        mongoengine.connect(TEST_MONGODB_LOG['db'])

        for _ in xrange(15):
            CourseImportLog(
                course_id=CourseKey.from_string("test/test/test"),
                location="location",
                import_log="import_log",
                git_log="git_log",
                repo_dir="repo_dir",
                created=datetime.now()
            ).save()

        for page, expected in [(-1, 1), (1, 1), (2, 2), (30, 2), ('abc', 1)]:
            response = self.client.get(
                '{}?page={}'.format(
                    reverse('gitlogs'),
                    page
                )
            )
            self.assertIn(
                'Page {} of 2'.format(expected),
                response.content
            )

        CourseImportLog.objects.delete()

    def test_gitlog_courseteam_access(self):
        """
        Ensure course team users are allowed to access only their own course.
        """

        self._mkdir(settings.GIT_REPO_DIR)

        self._setstaff_login()
        self._add_edx4edx()
        self.user.is_staff = False
        self.user.save()
        logged_in = self.client.login(username=self.user.username,
                                      password='foo')
        response = self.client.get(reverse('gitlogs'))
        # Make sure our non privileged user doesn't have access to all logs
        self.assertEqual(response.status_code, 404)
        # Or specific logs
        response = self.client.get(reverse('gitlogs_detail', kwargs={
            'course_id': 'MITx/edx4edx/edx4edx'
        }))
        self.assertEqual(response.status_code, 404)

        # Add user as staff in course team
        def_ms = modulestore()
        course = def_ms.get_course(CourseKey.from_string('MITx/edx4edx/edx4edx'))
        CourseStaffRole(course.id).add_users(self.user)

        self.assertTrue(CourseStaffRole(course.id).has_user(self.user))
        logged_in = self.client.login(username=self.user.username,
                                      password='foo')
        self.assertTrue(logged_in)

        response = self.client.get(
            reverse('gitlogs_detail', kwargs={
                'course_id': 'MITx/edx4edx/edx4edx'
            }))
        self.assertIn('======&gt; IMPORTING course',
                      response.content)

        self._rm_edx4edx()