Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved manage_user and manage_group from edx-management-commands

Browse Source 
We need to create a user profile in order for users to be usable. This, coupled with the fact, that edx-platform is the owner of auth information, means these commands belong here.

ECOM-4310
This commit is contained in:
Clinton Blackburn
2016-05-13 02:07:12 -04:00
parent b6b5fdc28e
commit ebf32dc208
7 changed files with 567 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
cms/envs/common.py
View File

@@ -906,9 +906,6 @@ INSTALLED_APPS = (
# Static i18n support
'statici18n',
# Management commands used for configuration automation
'edx_management_commands.management_commands',
# Tagging
'cms.lib.xblock.tagging',
)

126
common/djangoapps/student/management/commands/manage_group.py Normal file
View File

@@ -0,0 +1,126 @@
"""
Management command `manage_group` is used to idempotently create Django groups
and set their permissions by name.
"""
from django.apps import apps
from django.contrib.auth.models import Group, Permission
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
from django.utils.translation import gettext as _
class Command(BaseCommand):
# pylint: disable=missing-docstring
help = 'Creates the specified group, if it does not exist, and sets its permissions.'
def add_arguments(self, parser):
parser.add_argument('group_name')
parser.add_argument('--remove', dest='is_remove', action='store_true')
parser.add_argument('-p', '--permissions', nargs='*', default=[])
def _handle_remove(self, group_name):
try:
Group.objects.get(name=group_name).delete() # pylint: disable=no-member
self.stderr.write(_('Removed group: "{}"').format(group_name))
except Group.DoesNotExist:
self.stderr.write(_('Did not find a group with name "{}" - skipping.').format(group_name))
@transaction.atomic
def handle(self, group_name, is_remove, permissions=None, *args, **options):
if is_remove:
self._handle_remove(group_name)
return
old_permissions = set()
group, created = Group.objects.get_or_create(name=group_name) # pylint: disable=no-member
if created:
try:
# Needed for sqlite backend (i.e. in tests) because
# name.max_length won't be enforced by the db.
# See also http://www.sqlite.org/faq.html#q9
group.full_clean()
except ValidationError as exc:
# give a more helpful error
raise CommandError(
_(
'Invalid group name: "{group_name}". {messages}'
).format(
group_name=group_name,
messages=exc.messages[0]
)
)
self.stderr.write(_('Created new group: "{}"').format(group_name))
else:
self.stderr.write(_('Found existing group: "{}"').format(group_name))
old_permissions = set(group.permissions.all())
new_permissions = self._resolve_permissions(permissions or set())
add_permissions = new_permissions - old_permissions
remove_permissions = old_permissions - new_permissions
self.stderr.write(
_(
'Adding {codenames} permissions to group "{group}"'
).format(
codenames=[ap.name for ap in add_permissions],
group=group.name
)
)
self.stderr.write(
_(
'Removing {codenames} permissions from group "{group}"'
).format(
codenames=[rp.codename for rp in remove_permissions],
group=group.name
)
)
group.permissions = new_permissions
group.save()
def _resolve_permissions(self, permissions):
new_permissions = set()
for permission in permissions:
try:
app_label, model_name, codename = permission.split(':')
except ValueError:
# give a more helpful error
raise CommandError(_(
'Invalid permission option: "{}". Please specify permissions '
'using the format: app_label:model_name:permission_codename.'
).format(permission))
# this will raise a LookupError if it fails.
try:
model_class = apps.get_model(app_label, model_name)
except LookupError as exc:
raise CommandError(str(exc))
content_type = ContentType.objects.get_for_model(model_class)
try:
new_permission = Permission.objects.get( # pylint: disable=no-member
content_type=content_type,
codename=codename,
)
except Permission.DoesNotExist:
# give a more helpful error
raise CommandError(
_(
'Invalid permission codename: "{codename}". No such permission exists '
'for the model {module}.{model_name}.'
).format(
codename=codename,
module=model_class.__module__,
model_name=model_class.__name__,
)
)
new_permissions.add(new_permission)
return new_permissions

132
common/djangoapps/student/management/commands/manage_user.py Normal file
View File

@@ -0,0 +1,132 @@
"""
Management command `manage_user` is used to idempotently create or remove
Django users, set/unset permission bits, and associate groups by name.
"""
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
from django.utils.translation import gettext as _
from student.models import UserProfile
class Command(BaseCommand):
# pylint: disable=missing-docstring
help = 'Creates the specified user, if it does not exist, and sets its groups.'
def add_arguments(self, parser):
parser.add_argument('username')
parser.add_argument('email')
parser.add_argument('--remove', dest='is_remove', action='store_true')
parser.add_argument('--superuser', dest='is_superuser', action='store_true')
parser.add_argument('--staff', dest='is_staff', action='store_true')
parser.add_argument('-g', '--groups', nargs='*', default=[])
def _maybe_update(self, user, attribute, new_value):
"""
DRY helper. If the specified attribute of the user differs from the
specified value, it will be updated.
"""
old_value = getattr(user, attribute)
if new_value != old_value:
self.stderr.write(
_('Setting {attribute} for user "{username}" to "{new_value}"').format(
attribute=attribute, username=user.username, new_value=new_value
)
)
setattr(user, attribute, new_value)
def _check_email_match(self, user, email):
"""
DRY helper.
Requiring the user to specify both username and email will help catch
certain issues, for example if the expected username has already been
taken by someone else.
"""
if user.email != email:
# The passed email address doesn't match this username's email address.
# Assume a problem and fail.
raise CommandError(
_(
'Skipping user "{}" because the specified and existing email '
'addresses do not match.'
).format(user.username)
)
def _handle_remove(self, username, email):
try:
user = get_user_model().objects.get(username=username)
except get_user_model().DoesNotExist:
self.stderr.write(_('Did not find a user with username "{}" - skipping.').format(username))
return
self._check_email_match(user, email)
self.stderr.write(_('Removing user: "{}"').format(user))
user.delete()
@transaction.atomic
def handle(self, username, email, is_remove, is_staff, is_superuser, groups, *args, **options):
if is_remove:
return self._handle_remove(username, email)
old_groups, new_groups = set(), set()
user, created = get_user_model().objects.get_or_create(
username=username,
defaults={'email': email}
)
if created:
user.set_unusable_password()
self.stderr.write(_('Created new user: "{}"').format(user))
else:
# NOTE, we will not update the email address of an existing user.
self.stderr.write(_('Found existing user: "{}"').format(user))
self._check_email_match(user, email)
old_groups = set(user.groups.all())
self._maybe_update(user, 'is_staff', is_staff)
self._maybe_update(user, 'is_superuser', is_superuser)
# Ensure the user has a profile
try:
__ = user.profile
except UserProfile.DoesNotExist:
UserProfile.objects.create(user=user)
self.stderr.write(_('Created new profile for user: "{}"').format(user))
# resolve the specified groups
for group_name in groups or set():
try:
group = Group.objects.get(name=group_name) # pylint: disable=no-member
new_groups.add(group)
except Group.DoesNotExist:
# warn, but move on.
self.stderr.write(_('Could not find a group named "{}" - skipping.').format(group_name))
add_groups = new_groups - old_groups
remove_groups = old_groups - new_groups
self.stderr.write(
_(
'Adding user "{username}" to groups {group_names}'
).format(
username=user.username,
group_names=[g.name for g in add_groups]
)
)
self.stderr.write(
_(
'Removing user "{username}" from groups {group_names}'
).format(
username=user.username,
group_names=[g.name for g in remove_groups]
)
)
user.groups = new_groups
user.save()

187
common/djangoapps/student/management/tests/test_manage_group.py Normal file
View File

@@ -0,0 +1,187 @@
"""
Unit tests for user_management management commands.
"""
import sys
import ddt
from django.contrib.auth.models import Group, Permission
from django.contrib.contenttypes.models import ContentType
from django.core.management import call_command, CommandError
from django.test import TestCase
TEST_EMAIL = 'test@example.com'
TEST_GROUP = 'test-group'
TEST_USERNAME = 'test-user'
TEST_DATA = (
{},
{
TEST_GROUP: ['add_group', 'change_group', 'change_group'],
},
{
'other-group': ['add_group', 'change_group', 'change_group'],
},
)
@ddt.ddt
class TestManageGroupCommand(TestCase):
"""
Tests the `manage_group` command.
"""
def set_group_permissions(self, group_permissions):
"""
Sets up a before-state for groups and permissions in tests, which
can be checked afterward to ensure that a failed atomic
operation has not had any side effects.
"""
content_type = ContentType.objects.get_for_model(Group)
for group_name, permission_codenames in group_permissions.items():
group = Group.objects.create(name=group_name)
for codename in permission_codenames:
group.permissions.add(
Permission.objects.get(content_type=content_type, codename=codename) # pylint: disable=no-member
)
def check_group_permissions(self, group_permissions):
"""
Checks that the current state of the database matches the specified groups and
permissions.
"""
self.check_groups(group_permissions.keys())
for group_name, permission_codenames in group_permissions.items():
self.check_permissions(group_name, permission_codenames)
def check_groups(self, group_names):
"""
DRY helper.
"""
self.assertEqual(set(group_names), {g.name for g in Group.objects.all()}) # pylint: disable=no-member
def check_permissions(self, group_name, permission_codenames):
"""
DRY helper.
"""
self.assertEqual(
set(permission_codenames),
{p.codename for p in Group.objects.get(name=group_name).permissions.all()} # pylint: disable=no-member
)
@ddt.data(
*(
(data, args, exception)
for data in TEST_DATA
for args, exception in (
((), 'too few arguments' if sys.version_info.major == 2 else 'required: group_name'), # no group name
(('x' * 81,), 'invalid group name'), # invalid group name
((TEST_GROUP, 'some-other-group'), 'unrecognized arguments'), # multiple arguments
((TEST_GROUP, '--some-option', 'dummy'), 'unrecognized arguments') # unexpected option name
)
)
)
@ddt.unpack
def test_invalid_input(self, initial_group_permissions, command_args, exception_message):
"""
Ensures that invalid inputs result in errors with relevant output,
and that no persistent state is changed.
"""
self.set_group_permissions(initial_group_permissions)
with self.assertRaises(CommandError) as exc_context:
call_command('manage_group', *command_args)
self.assertIn(exception_message, str(exc_context.exception).lower())
self.check_group_permissions(initial_group_permissions)
@ddt.data(*TEST_DATA)
def test_invalid_permission(self, initial_group_permissions):
"""
Ensures that a permission that cannot be parsed or resolved results in
and error and that no persistent state is changed.
"""
self.set_group_permissions(initial_group_permissions)
# not parseable
with self.assertRaises(CommandError) as exc_context:
call_command('manage_group', TEST_GROUP, '--permissions', 'fail')
self.assertIn('invalid permission option', str(exc_context.exception).lower())
self.check_group_permissions(initial_group_permissions)
# not parseable
with self.assertRaises(CommandError) as exc_context:
call_command('manage_group', TEST_GROUP, '--permissions', 'f:a:i:l')
self.assertIn('invalid permission option', str(exc_context.exception).lower())
self.check_group_permissions(initial_group_permissions)
# invalid app label
with self.assertRaises(CommandError) as exc_context:
call_command('manage_group', TEST_GROUP, '--permissions', 'nonexistent-label:dummy-model:dummy-perm')
self.assertIn('no installed app', str(exc_context.exception).lower())
self.assertIn('nonexistent-label', str(exc_context.exception).lower())
self.check_group_permissions(initial_group_permissions)
# invalid model name
with self.assertRaises(CommandError) as exc_context:
call_command('manage_group', TEST_GROUP, '--permissions', 'auth:nonexistent-model:dummy-perm')
self.assertIn('nonexistent-model', str(exc_context.exception).lower())
self.check_group_permissions(initial_group_permissions)
# invalid model name
with self.assertRaises(CommandError) as exc_context:
call_command('manage_group', TEST_GROUP, '--permissions', 'auth:Group:nonexistent-perm')
self.assertIn('invalid permission codename', str(exc_context.exception).lower())
self.assertIn('nonexistent-perm', str(exc_context.exception).lower())
self.check_group_permissions(initial_group_permissions)
def test_group(self):
"""
Ensures that groups are created if they don't exist and reused if they do.
"""
self.check_groups([])
call_command('manage_group', TEST_GROUP)
self.check_groups([TEST_GROUP])
# check idempotency
call_command('manage_group', TEST_GROUP)
self.check_groups([TEST_GROUP])
def test_group_remove(self):
"""
Ensures that groups are removed if they exist and we exit cleanly otherwise.
"""
self.set_group_permissions({TEST_GROUP: ['add_group']})
self.check_groups([TEST_GROUP])
call_command('manage_group', TEST_GROUP, '--remove')
self.check_groups([])
# check idempotency
call_command('manage_group', TEST_GROUP, '--remove')
self.check_groups([])
def test_permissions(self):
"""
Ensures that permissions are set on the group as specified.
"""
self.check_groups([])
call_command('manage_group', TEST_GROUP, '--permissions', 'auth:Group:add_group')
self.check_groups([TEST_GROUP])
self.check_permissions(TEST_GROUP, ['add_group'])
# check idempotency
call_command('manage_group', TEST_GROUP, '--permissions', 'auth:Group:add_group')
self.check_groups([TEST_GROUP])
self.check_permissions(TEST_GROUP, ['add_group'])
# check adding a permission
call_command('manage_group', TEST_GROUP, '--permissions', 'auth:Group:add_group', 'auth:Group:change_group')
self.check_groups([TEST_GROUP])
self.check_permissions(TEST_GROUP, ['add_group', 'change_group'])
# check removing a permission
call_command('manage_group', TEST_GROUP, '--permissions', 'auth:Group:change_group')
self.check_groups([TEST_GROUP])
self.check_permissions(TEST_GROUP, ['change_group'])
# check removing all permissions
call_command('manage_group', TEST_GROUP)
self.check_groups([TEST_GROUP])
self.check_permissions(TEST_GROUP, [])

122
common/djangoapps/student/management/tests/test_manage_user.py Normal file
View File

@@ -0,0 +1,122 @@
"""
Unit tests for user_management management commands.
"""
import itertools
import ddt
from django.contrib.auth.models import Group, User
from django.core.management import call_command, CommandError
from django.test import TestCase
TEST_EMAIL = 'test@example.com'
TEST_USERNAME = 'test-user'
@ddt.ddt
class TestManageUserCommand(TestCase):
"""
Tests the `manage_user` command.
"""
def test_user(self):
"""
Ensures that users are created if they don't exist and reused if they do.
"""
# pylint: disable=no-member
self.assertEqual([], list(User.objects.all()))
call_command('manage_user', TEST_USERNAME, TEST_EMAIL)
user = User.objects.get(username=TEST_USERNAME)
self.assertEqual(user.username, TEST_USERNAME)
self.assertEqual(user.email, TEST_EMAIL)
self.assertIsNotNone(user.profile)
# check idempotency
call_command('manage_user', TEST_USERNAME, TEST_EMAIL)
self.assertEqual([(TEST_USERNAME, TEST_EMAIL)], [(u.username, u.email) for u in User.objects.all()])
def test_remove(self):
"""
Ensures that users are removed if they exist and exit cleanly otherwise.
"""
# pylint: disable=no-member
User.objects.create(username=TEST_USERNAME, email=TEST_EMAIL)
self.assertEqual([(TEST_USERNAME, TEST_EMAIL)], [(u.username, u.email) for u in User.objects.all()])
call_command('manage_user', TEST_USERNAME, TEST_EMAIL, '--remove')
self.assertEqual([], list(User.objects.all()))
# check idempotency
call_command('manage_user', TEST_USERNAME, TEST_EMAIL, '--remove')
self.assertEqual([], list(User.objects.all()))
def test_wrong_email(self):
"""
Ensure that the operation is aborted if the username matches an
existing user account but the supplied email doesn't match.
"""
# pylint: disable=no-member
User.objects.create(username=TEST_USERNAME, email=TEST_EMAIL)
with self.assertRaises(CommandError) as exc_context:
call_command('manage_user', TEST_USERNAME, 'other@example.com')
self.assertIn('email addresses do not match', str(exc_context.exception).lower())
self.assertEqual([(TEST_USERNAME, TEST_EMAIL)], [(u.username, u.email) for u in User.objects.all()])
# check that removal uses the same check
with self.assertRaises(CommandError) as exc_context:
call_command('manage_user', TEST_USERNAME, 'other@example.com', '--remove')
self.assertIn('email addresses do not match', str(exc_context.exception).lower())
self.assertEqual([(TEST_USERNAME, TEST_EMAIL)], [(u.username, u.email) for u in User.objects.all()])
@ddt.data(*itertools.product([(True, True), (True, False), (False, True), (False, False)], repeat=2))
@ddt.unpack
def test_bits(self, initial_bits, expected_bits):
"""
Ensure that the 'staff' and 'superuser' bits are set according to the
presence / absence of the associated command options, regardless of
any previous state.
"""
initial_staff, initial_super = initial_bits
User.objects.create(
username=TEST_USERNAME,
email=TEST_EMAIL,
is_staff=initial_staff,
is_superuser=initial_super
)
expected_staff, expected_super = expected_bits
args = [opt for bit, opt in ((expected_staff, '--staff'), (expected_super, '--superuser')) if bit]
call_command('manage_user', TEST_USERNAME, TEST_EMAIL, *args)
user = User.objects.all().first() # pylint: disable=no-member
self.assertEqual(user.is_staff, expected_staff)
self.assertEqual(user.is_superuser, expected_super)
@ddt.data(*itertools.product(('', 'a', 'ab', 'abc'), repeat=2))
@ddt.unpack
def test_groups(self, initial_groups, expected_groups):
"""
Ensures groups assignments are created and deleted idempotently.
"""
groups = {}
for group_name in 'abc':
groups[group_name] = Group.objects.create(name=group_name)
user = User.objects.create(username=TEST_USERNAME, email=TEST_EMAIL)
for group_name in initial_groups:
user.groups.add(groups[group_name])
call_command('manage_user', TEST_USERNAME, TEST_EMAIL, '-g', *expected_groups)
actual_groups = [group.name for group in user.groups.all()]
self.assertEqual(actual_groups, list(expected_groups))
def test_nonexistent_group(self):
"""
Ensures the command does not fail if specified groups cannot be found.
"""
user = User.objects.create(username=TEST_USERNAME, email=TEST_EMAIL)
groups = {}
for group_name in 'abc':
groups[group_name] = Group.objects.create(name=group_name)
user.groups.add(groups[group_name])
call_command('manage_user', TEST_USERNAME, TEST_EMAIL, '-g', 'b', 'c', 'd')
actual_groups = [group.name for group in user.groups.all()]
self.assertEqual(actual_groups, ['b', 'c'])

3
lms/envs/common.py
View File

@@ -2040,9 +2040,6 @@ INSTALLED_APPS = (
# API access administration
'openedx.core.djangoapps.api_admin',
# Management commands used for configuration automation
'edx_management_commands.management_commands',
# Verified Track Content Cohorting
'verified_track_content',

1
requirements/edx/base.txt
View File

@@ -41,7 +41,6 @@ djangorestframework-oauth==1.1.0
edx-ccx-keys==0.1.2
edx-drf-extensions==0.5.1
edx-lint==0.4.3
edx-management-commands==0.1.1
edx-django-oauth2-provider==1.0.3
edx-oauth2-provider==1.0.1
edx-opaque-keys==0.2.1