# -*- coding: utf-8 -*-
import argparse
from pyramid.scripting import AppEnvironment
import transaction
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import NoResultFound
from tracim import CFG
from tracim.command import AppContextCommand
from tracim.command import Extender
from tracim.exceptions import UserAlreadyExistError
from tracim.exceptions import GroupDoesNotExist
from tracim.exceptions import NotificationNotSend
from tracim.exceptions import BadCommandError
from tracim.lib.core.group import GroupApi
from tracim.lib.core.user import UserApi
from tracim.models import User
from tracim.models import Group
class UserCommand(AppContextCommand):
ACTION_CREATE = 'create'
ACTION_UPDATE = 'update'
action = NotImplemented
def get_description(self) -> str:
return '''Create or update user.'''
def get_parser(self, prog_name: str) -> argparse.ArgumentParser:
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(
"-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=[],
)
parser.add_argument(
"--send-email",
help='Send mail to user',
dest='send_email',
required=False,
action='store_true',
default=False,
)
return parser
def _user_exist(self, login: str) -> User:
return self._user_api.user_with_email_exists(login)
def _get_group(self, name: str) -> Group:
groups_availables = [group.group_name
for group in self._group_api.get_all()]
if name not in groups_availables:
msg = "Group '{}' does not exist, choose a group name in : ".format(name) # nopep8
for group in groups_availables:
msg+= "'{}',".format(group)
self._session.rollback()
raise GroupDoesNotExist(msg)
return self._group_api.get_one_with_name(name)
def _add_user_to_named_group(
self,
user: str,
group_name: str
) -> None:
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: User,
group_name: str
) -> None:
group = self._get_group(group_name)
if user in group.users:
group.users.remove(user)
self._session.flush()
def _create_user(
self,
login: str,
password: str,
do_notify: bool,
**kwargs
) -> User:
if not password:
if self._password_required():
raise BadCommandError(
"You must provide -p/--password parameter"
)
password = ''
try:
user = self._user_api.create_user(
email=login,
password=password,
do_save=True,
do_notify=do_notify,
)
# TODO - G.M - 04-04-2018 - [Caldav] Check this code
# # We need to enable radicale if it not already done
# daemons = DaemonsManager()
# daemons.run('radicale', RadicaleDaemon)
self._user_api.execute_created_user_actions(user)
except IntegrityError:
self._session.rollback()
raise UserAlreadyExistError()
except NotificationNotSend as exception:
self._session.rollback()
raise exception
return user
def _update_password_for_login(self, login: str, password: str) -> None:
user = self._user_api.get_one_by_email(login)
user.password = password
self._session.flush()
transaction.commit()
def take_app_action(
self,
parsed_args: argparse.Namespace,
app_context: AppEnvironment
) -> None:
# TODO - G.M - 05-04-2018 -Refactor this in order
# to not setup object var outside of __init__ .
self._session = app_context['request'].dbsession
self._app_config = app_context['registry'].settings['CFG']
self._user_api = UserApi(
current_user=None,
session=self._session,
config=self._app_config,
)
self._group_api = GroupApi(
current_user=None,
session=self._session,
config=self._app_config,
)
user = self._proceed_user(parsed_args)
self._proceed_groups(user, parsed_args)
print("User created/updated")
def _proceed_user(self, parsed_args: argparse.Namespace) -> User:
self._check_context(parsed_args)
if self.action == self.ACTION_CREATE:
try:
user = self._create_user(
login=parsed_args.login,
password=parsed_args.password,
do_notify=parsed_args.send_email,
)
except UserAlreadyExistError:
raise UserAlreadyExistError("Error: User already exist (use `user update` command instead)")
except NotificationNotSend:
raise NotificationNotSend("Error: Cannot send email notification, user not created.")
# TODO - G.M - 04-04-2018 - [Email] Check this code
# if parsed_args.send_email:
# email_manager = get_email_manager()
# email_manager.notify_created_account(
# user=user,
# password=parsed_args.password,
# )
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: User,
parsed_args: argparse.Namespace
) -> None:
# 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)
def _password_required(self) -> bool:
# TODO - G.M - 04-04-2018 - [LDAP] Check this code
# if config.get('auth_type') == LDAPAuth.name:
# return False
return True
def _check_context(self, parsed_args: argparse.Namespace) -> None:
# TODO - G.M - 04-04-2018 - [LDAP] Check this code
# if config.get('auth_type') == LDAPAuth.name:
# auth_instance = config.get('auth_instance')
# if not auth_instance.ldap_auth.user_exist(parsed_args.login):
# raise LDAPUserUnknown(
# "LDAP is enabled and user with login/email \"%s\" not found in LDAP" % parsed_args.login
# )
pass
class CreateUserCommand(UserCommand):
action = UserCommand.ACTION_CREATE
class UpdateUserCommand(UserCommand):
action = UserCommand.ACTION_UPDATE
class LDAPUserUnknown(BadCommandError):
pass