add api access request management command
This commit is contained in:
jansenk
@@ -0,0 +1,101 @@
|
||||
""" Management command to create an ApiAccessRequest for given users """
|
||||
import logging
|
||||
from django.core.management.base import BaseCommand, CommandError
|
||||
from django.contrib.auth.models import User
|
||||
from django.contrib.sites.models import Site
|
||||
|
||||
from openedx.core.djangoapps.api_admin.models import (
|
||||
ApiAccessConfig,
|
||||
ApiAccessRequest,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Command(BaseCommand):
|
||||
"""
|
||||
Create an ApiAccessRequest for the given user
|
||||
|
||||
Example usage:
|
||||
$ ./manage.py lms create_api_request <username> --create-config
|
||||
"""
|
||||
|
||||
help = 'Create an ApiAccessRequest for the given user'
|
||||
DEFAULT_WEBSITE = 'www.test-edx-example-website.edu'
|
||||
DEFAULT_REASON = 'Generated by management job create_api_request'
|
||||
|
||||
def add_arguments(self, parser):
|
||||
parser.add_argument('username')
|
||||
parser.add_argument(
|
||||
'--create-config',
|
||||
action='store_true',
|
||||
help='Create ApiAccessConfig if it does not exist'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--status',
|
||||
choices=[choice[0] for choice in ApiAccessRequest.STATUS_CHOICES],
|
||||
default=ApiAccessRequest.APPROVED,
|
||||
help='Status of the created ApiAccessRequest'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--reason',
|
||||
default=self.DEFAULT_REASON,
|
||||
help='Reason that the ApiAccessRequest is being created'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--website',
|
||||
default=self.DEFAULT_WEBSITE,
|
||||
help='Website associated with the user of the created ApiAccessRequest'
|
||||
)
|
||||
|
||||
def handle(self, *args, **options):
|
||||
if options.get('create_config'):
|
||||
self.create_api_access_config()
|
||||
user = self.get_user(options.get('username'))
|
||||
self.create_api_access_request(
|
||||
user,
|
||||
options.get('status'),
|
||||
options.get('reason'),
|
||||
options.get('website'),
|
||||
)
|
||||
|
||||
def get_user(self, username):
|
||||
try:
|
||||
return User.objects.get(username=username)
|
||||
except User.DoesNotExist:
|
||||
raise CommandError(u'User {} not found'.format(username))
|
||||
|
||||
def create_api_access_request(self, user, status, reason, website):
|
||||
"""
|
||||
Creates an ApiAccessRequest with the given values.
|
||||
"""
|
||||
try:
|
||||
ApiAccessRequest.objects.create(
|
||||
user=user,
|
||||
status=status,
|
||||
website=website,
|
||||
reason=reason,
|
||||
site=Site.objects.get_current(),
|
||||
)
|
||||
except Exception as e:
|
||||
msg = u'Unable to create ApiAccessRequest for {}. Exception is {}: {}'.format(
|
||||
user.username,
|
||||
type(e).__name__,
|
||||
e
|
||||
)
|
||||
raise CommandError(msg)
|
||||
logger.info(u'Created ApiAccessRequest for user {}'.format(user.username))
|
||||
|
||||
def create_api_access_config(self):
|
||||
"""
|
||||
Creates an active ApiAccessConfig if one does not currectly exist
|
||||
"""
|
||||
try:
|
||||
_, created = ApiAccessConfig.objects.get_or_create(enabled=True)
|
||||
except Exception as e:
|
||||
msg = u'Unable to create ApiAccessConfig. Exception is {}: {}'.format(type(e).__name__, e)
|
||||
raise CommandError(msg)
|
||||
if created:
|
||||
logger.info(u'Created ApiAccessConfig')
|
||||
else:
|
||||
logger.info(u'ApiAccessConfig already exists')
|
||||
@@ -0,0 +1,97 @@
|
||||
from mock import patch
|
||||
|
||||
import ddt
|
||||
from django.conf import settings
|
||||
from django.contrib.sites.models import Site
|
||||
from django.core.management import call_command
|
||||
from django.core.management.base import CommandError
|
||||
from django.test import TestCase
|
||||
import unittest
|
||||
|
||||
from openedx.core.djangoapps.api_admin.management.commands import create_api_access_request
|
||||
from openedx.core.djangoapps.api_admin.models import (
|
||||
ApiAccessConfig,
|
||||
ApiAccessRequest,
|
||||
)
|
||||
from student.tests.factories import UserFactory
|
||||
|
||||
|
||||
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Tests only valid in lms')
|
||||
@ddt.ddt
|
||||
class TestCreateApiAccessRequest(TestCase):
|
||||
""" Test create_api_access_request command """
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super(TestCreateApiAccessRequest, cls).setUpClass()
|
||||
cls.command = 'create_api_access_request'
|
||||
cls.user = UserFactory()
|
||||
|
||||
def assert_models_exist(self, expect_request_exists, expect_config_exists):
|
||||
self.assertEqual(
|
||||
ApiAccessRequest.objects.filter(user=self.user).exists(),
|
||||
expect_request_exists
|
||||
)
|
||||
self.assertEqual(
|
||||
ApiAccessConfig.objects.filter(enabled=True).exists(),
|
||||
expect_config_exists
|
||||
)
|
||||
|
||||
@ddt.data(False, True)
|
||||
def test_create_api_access_request(self, create_config):
|
||||
self.assert_models_exist(False, False)
|
||||
call_command(self.command, self.user.username, create_config=create_config)
|
||||
self.assert_models_exist(True, create_config)
|
||||
|
||||
def test_config_already_exists(self):
|
||||
ApiAccessConfig.objects.create(enabled=True)
|
||||
self.assert_models_exist(False, True)
|
||||
call_command(self.command, self.user.username, create_config=True)
|
||||
self.assert_models_exist(True, True)
|
||||
|
||||
def test_user_not_found(self):
|
||||
with self.assertRaisesRegex(CommandError, r'User .*? not found'):
|
||||
call_command(self.command, 'not-a-user-notfound-nope')
|
||||
|
||||
@patch('openedx.core.djangoapps.api_admin.models.ApiAccessRequest.objects.create')
|
||||
def test_api_request_error(self, mocked_method):
|
||||
mocked_method.side_effect = Exception()
|
||||
|
||||
self.assert_models_exist(False, False)
|
||||
|
||||
with self.assertRaisesRegex(CommandError, r'Unable to create ApiAccessRequest .*'):
|
||||
call_command(self.command, self.user.username)
|
||||
|
||||
self.assert_models_exist(False, False)
|
||||
|
||||
@patch('openedx.core.djangoapps.api_admin.models.ApiAccessConfig.objects.get_or_create')
|
||||
def test_api_config_error(self, mocked_method):
|
||||
mocked_method.side_effect = Exception()
|
||||
self.assert_models_exist(False, False)
|
||||
|
||||
with self.assertRaisesRegex(CommandError, r'Unable to create ApiAccessConfig\. .*'):
|
||||
call_command(self.command, self.user.username, create_config=True)
|
||||
|
||||
self.assert_models_exist(False, False)
|
||||
|
||||
def test_optional_fields(self):
|
||||
self.assert_models_exist(False, False)
|
||||
|
||||
call_command(
|
||||
self.command,
|
||||
self.user.username,
|
||||
status='approved',
|
||||
reason='whatever',
|
||||
website='test-site.edx.horse'
|
||||
)
|
||||
self.assert_models_exist(True, False)
|
||||
request = ApiAccessRequest.objects.get(user=self.user)
|
||||
self.assertEqual(request.status, 'approved')
|
||||
self.assertEqual(request.reason, 'whatever')
|
||||
self.assertEqual(request.website, 'test-site.edx.horse')
|
||||
|
||||
def test_default_values(self):
|
||||
call_command(self.command, self.user.username)
|
||||
request = ApiAccessRequest.objects.get(user=self.user)
|
||||
self.assertEqual(request.site, Site.objects.get_current())
|
||||
self.assertEqual(request.status, ApiAccessRequest.APPROVED)
|
||||
Reference in New Issue
Block a user