USer management commands

tracim/setup.py

         ],
         'gearbox.commands': [
             'ldap_server = tracim.command.ldap_test_server:LDAPTestServerCommand',
+            'user_create = tracim.command.user:CreateUserCommand',
+            'user_update = tracim.command.user:UpdateUserCommand',
         ]
     },
     dependency_links=[

tracim/tracim/command/__init__.py

-__author__ = 'bux'
+# -*- coding: utf-8 -*-
+import argparse
+import os
+import sys
+
+import transaction
+from gearbox.command import Command
+from paste.deploy import loadapp
+from webtest import TestApp
+
+from tracim.lib.exception import CommandAbortedError
+
+
+class BaseCommand(Command):
+    """ Setup ap at take_action call """
+    auto_setup_app = True
+
+    def run(self, parsed_args):
+        try:
+            super().run(parsed_args)
+        except CommandAbortedError as exc:
+            print(exc)
+
+
+class AppContextCommand(BaseCommand):
+    """
+    Command who initialize app context at beginning of take_action method.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(AppContextCommand, self).__init__(*args, **kwargs)
+
+    @staticmethod
+    def _get_initialized_app_context(parsed_args):
+        """
+        :param parsed_args: parsed args (eg. from take_action)
+        :return: (wsgi_app, test_app)
+        """
+        config_file = parsed_args.config_file
+        config_name = 'config:%s' % config_file
+        here_dir = os.getcwd()
+
+        # Load locals and populate with objects for use in shell
+        sys.path.insert(0, here_dir)
+
+        # Load the wsgi app first so that everything is initialized right
+        wsgi_app = loadapp(config_name, relative_to=here_dir)
+        test_app = TestApp(wsgi_app)
+
+        # Make available the tg.request and other global variables
+        tresponse = test_app.get('/_test_vars')
+
+        return wsgi_app, test_app
+
+    def take_action(self, parsed_args):
+        super(AppContextCommand, self).take_action(parsed_args)
+        if self.auto_setup_app:
+            self._get_initialized_app_context(parsed_args)
+
+    def get_parser(self, prog_name):
+        parser = super(AppContextCommand, self).get_parser(prog_name)
+
+        parser.add_argument("-c", "--config",
+                            help='application config file to read (default: development.ini)',
+                            dest='config_file', default="development.ini")
+        return parser
+
+    def run(self, parsed_args):
+        super().run(parsed_args)
+        transaction.commit()
+
+
+class Extender(argparse.Action):
+    """
+    Copied class from http://stackoverflow.com/a/12461237/801924
+    """
+    def __call__(self, parser, namespace, values, option_strings=None):
+        # Need None here incase `argparse.SUPPRESS` was supplied for `dest`
+        dest = getattr(namespace, self.dest, None)
+        # print dest,self.default,values,option_strings
+        if not hasattr(dest, 'extend') or dest == self.default:
+            dest = []
+            setattr(namespace, self.dest, dest)
+            # if default isn't set to None, this method might be called
+            # with the default as `values` for other arguements which
+            # share this destination.
+            parser.set_defaults(**{self.dest: None})
+        try:
+            dest.extend(values)
+        except ValueError:
+            dest.append(values)

tracim/tracim/command/user.py

+# -*- coding: utf-8 -*-
+import transaction
+from sqlalchemy.exc import IntegrityError
+
+from tracim.command import AppContextCommand, Extender
+from tracim.lib.exception import AlreadyExistError, CommandAbortedError
+from tracim.lib.group import GroupApi
+from tracim.lib.user import UserApi
+from tracim.model import DBSession, User
+
+
+class UserCommand(AppContextCommand):
+
+    ACTION_CREATE = 'create'
+    ACTION_UPDATE = 'update'
+
+    action = NotImplemented
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._session = DBSession
+        self._transaction = transaction
+        self._user_api = UserApi(None)
+        self._group_api = GroupApi(None)
+
+    def get_parser(self, prog_name):
+        parser = super().get_parser(prog_name)
+
+        parser.add_argument(
+            "-l",
+            "--login",
+            help='User login (email)',
+            dest='login',
+            required=True
+        )
+
+        parser.add_argument(
+            "-p",
+            "--password",
+            help='User password',
+            dest='password',
+            required=False,
+            default=None
+        )
+
+        parser.add_argument(
+            "-u",
+            "--update",
+            help='Update user password if exist',
+            dest='update',
+            action='store_true'
+        )
+
+        parser.add_argument(
+            "-g",
+            "--add-to-group",
+            help='Add user to group',
+            dest='add_to_group',
+            nargs='*',
+            action=Extender,
+            default=[],
+        )
+
+        parser.add_argument(
+            "-rmg",
+            "--remove-from-group",
+            help='Remove user from group',
+            dest='remove_from_group',
+            nargs='*',
+            action=Extender,
+            default=[],
+        )
+
+        return parser
+
+    def _user_exist(self, login):
+        return self._user_api.user_with_email_exists(login)
+
+    def _get_group(self, name):
+        return self._group_api.get_one_with_name(name)
+
+    def _add_user_to_named_group(self, user, group_name):
+        group = self._get_group(group_name)
+        if user not in group.users:
+            group.users.append(user)
+        self._session.flush()
+
+    def _remove_user_from_named_group(self, user, group_name):
+        group = self._get_group(group_name)
+        if user in group.users:
+            group.users.remove(user)
+        self._session.flush()
+
+    def _create_user(self, login, password, **kwargs):
+        if not password:
+            raise CommandAbortedError("You must provide -p/--password parameter")
+
+        try:
+            user = User(email=login, password=password, **kwargs)
+            self._session.add(user)
+            self._session.flush()
+        except IntegrityError:
+            self._session.rollback()
+            raise AlreadyExistError()
+
+        return user
+
+    def _update_password_for_login(self, login, password):
+        user = self._user_api.get_one_by_email(login)
+        user.password = password
+        self._session.flush()
+        transaction.commit()
+
+    def take_action(self, parsed_args):
+        super().take_action(parsed_args)
+
+        user = self._proceed_user(parsed_args)
+        self._proceed_groups(user, parsed_args)
+
+        print("User created/updated")
+
+    def _proceed_user(self, parsed_args):
+        if self.action == self.ACTION_CREATE:
+            try:
+                user = self._create_user(login=parsed_args.login, password=parsed_args.password)
+            except AlreadyExistError:
+                raise CommandAbortedError("Error: User already exist (use `user update` command instead)")
+        else:
+            if parsed_args.password:
+                self._update_password_for_login(login=parsed_args.login, password=parsed_args.password)
+            user = self._user_api.get_one_by_email(parsed_args.login)
+
+        return user
+
+    def _proceed_groups(self, user, parsed_args):
+        # User always in "users" group
+        self._add_user_to_named_group(user, 'users')
+
+        for group_name in parsed_args.add_to_group:
+            self._add_user_to_named_group(user, group_name)
+
+        for group_name in parsed_args.remove_from_group:
+            self._remove_user_from_named_group(user, group_name)
+
+
+class CreateUserCommand(UserCommand):
+    action = UserCommand.ACTION_CREATE
+
+
+class UpdateUserCommand(UserCommand):
+    action = UserCommand.ACTION_UPDATE

tracim/tracim/lib/exception.py

+# -*- coding: utf-8 -*-
+
+
+class TracimError(Exception):
+    pass
+
+
+class AlreadyExistError(TracimError):
+    pass
+
+
+class CommandError(TracimError):
+    pass
+
+
+class CommandAbortedError(CommandError):
+    pass

tracim/tracim/lib/group.py

 
     def get_one(self, group_id) -> Group:
         return self._base_query().filter(Group.group_id==group_id).one()
+
+    def get_one_with_name(self, group_name) -> Group:
+        return self._base_query().filter(Group.group_name==group_name).one()

tracim/tracim/tests/__init__.py

 # -*- coding: utf-8 -*-
 """Unit and functional test suite for tracim."""
-
-from os import getcwd, path
+import argparse
+from os import getcwd
 
 import ldap3
+import tg
+import transaction
+from gearbox.commands.setup_app import SetupAppCommand
 from ldap_test import LdapServer
 from nose.tools import ok_
 from paste.deploy import loadapp
-from webtest import TestApp
-
-from gearbox.commands.setup_app import SetupAppCommand
-
-import tg
-from tg import config
-from tg.util import Bunch
-
 from sqlalchemy.engine import reflection
-
 from sqlalchemy.schema import DropConstraint
 from sqlalchemy.schema import DropSequence
 from sqlalchemy.schema import DropTable
 from sqlalchemy.schema import MetaData
 from sqlalchemy.schema import Sequence
 from sqlalchemy.schema import Table
-
-import transaction
+from tg import config
+from tg.util import Bunch
+from webtest import TestApp
 from who_ldap import make_connection
 
+from tracim.command import BaseCommand
 from tracim.lib.base import logger
 from tracim.model import DBSession
 
     def tearDown(self):
         transaction.commit()
 
+
+class TestCommand(TestStandard):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        # We disable app loading from commands classes
+        BaseCommand.auto_setup_app = False
+        # Hack parser object to test conditions
+        BaseCommand.get_parser = self._get_test_parser()
+
+    def _get_test_parser(self):
+        def get_parser(self, prog_name):
+            parser = ArgumentParser(
+                description=self.get_description(),
+                prog=prog_name,
+                add_help=False
+            )
+            return parser
+        return get_parser
+
+    def _execute_command(self, command_class, command_name, sub_argv):
+        parser = argparse.ArgumentParser()
+        command = command_class(self.app, parser)
+        cmd_parser = command.get_parser(command_name)
+        parsed_args = cmd_parser.parse_args(sub_argv)
+        return command.run(parsed_args)
+
+    def setUp(self):
+        super().setUp()
+        # Ensure app config is loaded
+        self.app.get('/_test_vars')
+
+
 class TestController(object):
     """Base functional test case for the controllers.
 
                 ok_(False, "Cannot establish connection with LDAP test server")
             else:
                 ok_(True)
+
+
+class ArgumentParser(argparse.ArgumentParser):
+    def exit(self, status=0, message=None):
+        raise Exception(message)

tracim/tracim/tests/command/user.py

+from nose.tools import eq_
+from nose.tools import ok_
+
+from tracim.command.user import CreateUserCommand, UpdateUserCommand
+from tracim.model import DBSession, Group
+from tracim.model.auth import User
+from tracim.tests import TestCommand
+
+
+class TestUserCommand(TestCommand):
+
+    def test_create(self):
+        self._create_user('new-user@algoo.fr', 'toor')
+
+    def test_update_password(self):
+        self._create_user('new-user@algoo.fr', 'toor')
+        self._execute_command(
+            CreateUserCommand,
+            'gearbox user update',
+            ['-l', 'new-user@algoo.fr', '-p', 'new_password']
+        )
+        user = DBSession.query(User).filter(User.email == 'new-user@algoo.fr').one()
+        user.validate_password('new_password')
+
+    def test_create_with_group(self):
+        more_args = ['--add-to-group', 'managers', '--add-to-group', 'administrators']
+        self._create_user('new-user@algoo.fr', 'toor', more_args=more_args)
+        user = DBSession.query(User).filter(User.email == 'new-user@algoo.fr').one()
+        group_managers = DBSession.query(Group).filter(Group.group_name == 'managers').one()
+        group_administrators = DBSession.query(Group).filter(Group.group_name == 'administrators').one()
+
+        ok_(user in group_managers.users)
+        ok_(user in group_administrators.users)
+
+    def test_change_groups(self):
+        # create an user in managers group
+        more_args = ['--add-to-group', 'managers']
+        self._create_user('new-user@algoo.fr', 'toor', more_args=more_args)
+        user = DBSession.query(User).filter(User.email == 'new-user@algoo.fr').one()
+        group_managers = DBSession.query(Group).filter(Group.group_name == 'managers').one()
+        group_administrators = DBSession.query(Group).filter(Group.group_name == 'administrators').one()
+
+        ok_(user in group_managers.users)
+        ok_(user not in group_administrators.users)
+
+        # Update him and add to administrators group
+        add_to_admins_argvs = ['-l', 'new-user@algoo.fr', '--add-to-group', 'administrators']
+        self._execute_command(UpdateUserCommand, 'gearbox user update', add_to_admins_argvs)
+        user = DBSession.query(User).filter(User.email == 'new-user@algoo.fr').one()
+        group_managers = DBSession.query(Group).filter(Group.group_name == 'managers').one()
+        group_administrators = DBSession.query(Group).filter(Group.group_name == 'administrators').one()
+
+        ok_(user in group_managers.users)
+        ok_(user in group_administrators.users)
+
+        # remove him from administrators group
+        remove_from_admins_argvs = ['-l', 'new-user@algoo.fr', '--remove-from-group', 'administrators']
+        self._execute_command(UpdateUserCommand, 'gearbox user update', remove_from_admins_argvs)
+        user = DBSession.query(User).filter(User.email == 'new-user@algoo.fr').one()
+        group_managers = DBSession.query(Group).filter(Group.group_name == 'managers').one()
+        group_administrators = DBSession.query(Group).filter(Group.group_name == 'administrators').one()
+
+        ok_(user in group_managers.users)
+        ok_(user not in group_administrators.users)
+
+    def _create_user(self, email, password, more_args=[]):
+        args = ['-l', email, '-p', password]
+        args.extend(more_args)
+
+        self._check_user_exist(email, exist=False)
+        self._execute_command(CreateUserCommand, 'gearbox user create', args)
+        self._check_user_exist(email, exist=True)
+
+        user = DBSession.query(User).filter(User.email == email).one()
+        user.validate_password(password)
+
+    @staticmethod
+    def _check_user_exist(email, exist=True):
+        eq_(exist, bool(DBSession.query(User).filter(User.email == email).count()))
+