# create that user anyway

            if user_data['active_from_extern'] is False:

                log.warning("User %s authenticated against %s, but is inactive",

                            username, self.__module__)

                return None

            if self.use_fake_password():

                # Randomize the PW because we don't need it, but don't want

                # them blank either

                passwd = PasswordGenerator().gen_password(length=8)

            log.debug('Updating or creating user info from %s plugin',

                      self.name)

            user = UserModel().create_or_update(

                username=username,

                password=passwd,

                email=user_data["email"],

                firstname=user_data["firstname"],

                lastname=user_data["lastname"],

                active=user_data["active"],

                admin=user_data["admin"],

                extern_name=user_data["extern_name"],

                extern_type=self.name

            )

            # enforce user is just in given groups, all of them has to be ones

            # created from plugins. We store this info in _group_data JSON field

            groups = user_data['groups'] or []

            UserGroupModel().enforce_groups(user, groups, self.name)

            Session().commit()

        return user_data

def importplugin(plugin):

    """

    Imports and returns the authentication plugin in the module named by plugin

    (e.g., plugin='kallithea.lib.auth_modules.auth_internal'). Returns the

    KallitheaAuthPluginBase subclass on success, raises exceptions on failure.

    raises:

        AttributeError -- no KallitheaAuthPlugin class in the module

        TypeError -- if the KallitheaAuthPlugin is not a subclass of ours KallitheaAuthPluginBase

        ImportError -- if we couldn't import the plugin at all

    """

    log.debug("Importing %s", plugin)

    if not plugin.startswith(u'kallithea.lib.auth_modules.auth_'):

        parts = plugin.split(u'.lib.auth_modules.auth_', 1)

        if len(parts) == 2:

            _module, pn = parts

               gist_type=Gist.GIST_PUBLIC, lifetime=-1):

        """

        :param description: description of the gist

        :param owner: user who created this gist

        :param gist_mapping: mapping {filename:{'content':content},...}

        :param gist_type: type of gist private/public

        :param lifetime: in minutes, -1 == forever

        """

        owner = self._get_user(owner)

        gist_id = safe_unicode(unique_id(20))

        lifetime = safe_int(lifetime, -1)

        gist_expires = time.time() + (lifetime * 60) if lifetime != -1 else -1

        log.debug('set GIST expiration date to: %s',

                  time_to_datetime(gist_expires)

                   if gist_expires != -1 else 'forever')

        #create the Database version

        gist = Gist()

        gist.gist_description = description

        gist.gist_access_id = gist_id

        gist.gist_owner = owner.user_id

        gist.gist_expires = gist_expires

        gist.gist_type = safe_unicode(gist_type)

        self.sa.add(gist)

        if gist_type == Gist.GIST_PUBLIC:

            # use DB ID for easy to use GIST ID

            gist_id = safe_unicode(gist.gist_id)

            gist.gist_access_id = gist_id

            self.sa.add(gist)

        gist_repo_path = os.path.join(GIST_STORE_LOC, gist_id)

        log.debug('Creating new %s GIST repo in %s', gist_type, gist_repo_path)

        repo = RepoModel()._create_filesystem_repo(

            repo_name=gist_id, repo_type='hg', repo_group=GIST_STORE_LOC)

        processed_mapping = {}

        for filename in gist_mapping:

            if filename != os.path.basename(filename):

                raise Exception('Filename cannot be inside a directory')

            content = gist_mapping[filename]['content']

            #TODO: expand support for setting explicit lexers

#             if lexer is None:

#                 try:

#                     guess_lexer = pygments.lexers.guess_lexer_for_filename

#                     lexer = guess_lexer(filename,content)

#                 except pygments.util.ClassNotFound:

#                     lexer = 'text'

        lifetime = safe_int(lifetime, -1)

        if lifetime == 0:  # preserve old value

            gist_expires = gist.gist_expires

        else:

            gist_expires = time.time() + (lifetime * 60) if lifetime != -1 else -1

        #calculate operation type based on given data

        gist_mapping_op = {}

        for k, v in gist_mapping.items():

            # add, mod, del

            if not v['org_filename'] and v['filename']:

                op = 'add'

            elif v['org_filename'] and not v['filename']:

                op = 'del'

            else:

                op = 'mod'

            v['op'] = op

            gist_mapping_op[k] = v

        gist.gist_description = description

        gist.gist_expires = gist_expires

        gist.owner = owner

        gist.gist_type = gist_type

        message = 'updated file'

        message += 's: ' if len(gist_mapping) > 1 else ': '

        message += ', '.join([x for x in gist_mapping])

        #fake Kallithea Repository object

        fake_repo = AttributeDict(dict(

            repo_name=gist_repo.path,

            scm_instance_no_cache=lambda: gist_repo,

        ))

        self._store_metadata(gist_repo, gist.gist_id, gist.gist_access_id,

                             owner.user_id, gist.gist_type, gist.gist_expires)

        ScmModel().update_nodes(

            user=owner.user_id,

            repo=fake_repo,

            message=message,

            nodes=gist_mapping_op,

            trigger_push_hook=False

        )

        return gist

            check_allowed_create_user

        _fd = form_data

        user_data = {

            'username': _fd['username'],

            'password': _fd['password'],

            'email': _fd['email'],

            'firstname': _fd['firstname'],

            'lastname': _fd['lastname'],

            'active': _fd['active'],

            'admin': False

        }

        # raises UserCreationError if it's not allowed

        check_allowed_create_user(user_data, cur_user)

        from kallithea.lib.auth import get_crypt_password

        new_user = User()

        for k, v in form_data.items():

            if k == 'password':

                v = get_crypt_password(v)

            if k == 'firstname':

                k = 'name'

            setattr(new_user, k, v)

        new_user.api_key = generate_api_key()

        log_create_user(new_user.get_dict(), cur_user)

        return new_user

    def create_or_update(self, username, password, email, firstname=u'',

                         lastname=u'', active=True, admin=False,

                         extern_type=None, extern_name=None, cur_user=None):

        """

        Creates a new instance if not found, or updates current one

        :param username:

        :param password:

        :param email:

        :param active:

        :param firstname:

        :param lastname:

        :param active:

        :param admin:

        :param extern_name:

        :param extern_type:

        :param cur_user:

        """

        if not cur_user:

            cur_user = getattr(get_current_authuser(), 'username', None)

        try:

            new_user.username = username

            new_user.admin = admin

            new_user.email = email

            new_user.active = active

            new_user.extern_name = safe_str(extern_name) \

                if extern_name else None

            new_user.extern_type = safe_str(extern_type) \

                if extern_type else None

            new_user.name = firstname

            new_user.lastname = lastname

            if not edit:

                new_user.api_key = generate_api_key()

            # set password only if creating an user or password is changed

            password_change = new_user.password and \

                not check_password(password, new_user.password)

            if not edit or password_change:

                reason = 'new password' if edit else 'new user'

                log.debug('Updating password reason=>%s', reason)

                new_user.password = get_crypt_password(password) \

                    if password else ''

            if not edit:

                log_create_user(new_user.get_dict(), cur_user)

            return new_user

        except (DatabaseError,):

            log.error(traceback.format_exc())

            raise

    def create_registration(self, form_data):

        from kallithea.model.notification import NotificationModel

        import kallithea.lib.helpers as h

        form_data['admin'] = False

        form_data['extern_type'] = User.DEFAULT_AUTH_TYPE

        form_data['extern_name'] = ''

        new_user = self.create(form_data)

        # notification to admins

        subject = _('New user registration')

        body = (

            u'New user registration\n'

            '---------------------\n'

            '- Username: {user.username}\n'

            '- Full Name: {user.full_name}\n'

            '- Email: {user.email}\n'

            ).format(user=new_user)

        edit_url = h.canonical_url('edit_user', id=new_user.user_id)

        email_kwargs = {

            'registered_user_url': edit_url,

            'new_username': new_user.username,

            'new_email': new_user.email,

            'new_full_name': new_user.full_name}

        NotificationModel().create(created_by=new_user, subject=subject,

                                   body=body, recipients=None,

                                   type_=Notification.TYPE_REGISTRATION,

                                   email_kwargs=email_kwargs)

    def update(self, user_id, form_data, skip_attrs=None):

        from kallithea.lib.auth import get_crypt_password

        skip_attrs = skip_attrs or []

        user = self.get(user_id, cache=False)

            new_user_group.users_group_name = name

            new_user_group.user_group_description = description

            new_user_group.users_group_active = active

            if group_data:

                new_user_group.group_data = group_data

            self.sa.add(new_user_group)

            perm_obj = self._create_default_perms(new_user_group)

            self.sa.add(perm_obj)

            self.grant_user_permission(user_group=new_user_group,

                                       user=owner, perm='usergroup.admin')

            return new_user_group

        except Exception:

            log.error(traceback.format_exc())

            raise

    def update(self, user_group, form_data):

        try:

            user_group = self._get_user_group(user_group)

            for k, v in form_data.items():

                if k == 'users_group_members':

                    members_list = []

                    if v:

                        v = [v] if isinstance(v, basestring) else v

                        for u_id in set(v):

                            member = UserGroupMember(user_group.users_group_id, u_id)

                            members_list.append(member)

                setattr(user_group, k, v)

        except Exception:

            log.error(traceback.format_exc())

            raise

    def delete(self, user_group, force=False):

        """

        Deletes user group, unless force flag is used

        raises exception if there are members in that group, else deletes

        group and users

        :param user_group:

        :param force:

        """

        user_group = self._get_user_group(user_group)

        try:

            # check if this group is not assigned to repo

            assigned_groups = UserGroupRepoToPerm.query() \

                .filter(UserGroupRepoToPerm.users_group == user_group).all()

            assigned_groups = [x.repository.repo_name for x in assigned_groups]

            if assigned_groups and not force:

                raise UserGroupsAssignedException(

                    'User Group assigned to %s' % ", ".join(assigned_groups))

            self.sa.delete(user_group)

from kallithea.model.meta import Session

from kallithea.model.db import Setting

name = 'spam-setting-name'

def test_passing_list_setting_value_results_in_string_valued_setting():

    assert Setting.get_by_name(name) is None

    setting = Setting.create_or_update(name, ['spam', 'eggs'])

    try:

        assert Setting.get_by_name(name) is not None

        # Quirk: list value is stringified.

        assert Setting.get_by_name(name).app_settings_value \

               == "['spam', 'eggs']"

        assert Setting.get_by_name(name).app_settings_type == 'unicode'

    finally:

        Session().delete(setting)

def test_list_valued_setting_creation_requires_manual_value_formatting():

    assert Setting.get_by_name(name) is None

    # Quirk: need manual formatting of list setting value.

    setting = Setting.create_or_update(name, 'spam,eggs', type='list')

    try:

        assert setting.app_settings_value == ['spam', 'eggs']

    finally:

        Session().delete(setting)

def test_list_valued_setting_update():

    assert Setting.get_by_name(name) is None

    setting = Setting.create_or_update(name, 'spam', type='list')

    try:

        assert setting.app_settings_value == [u'spam']

        # Assign back setting value.

        setting.app_settings_value = setting.app_settings_value

        # Quirk: value is stringified on write and listified on read.

        assert setting.app_settings_value == ["[u'spam']"]

        setting.app_settings_value = setting.app_settings_value

        assert setting.app_settings_value == ["[u\"[u'spam']\"]"]

    finally:

        Session().delete(setting)