Dispatch rescan repositories action. If remove_obsolete is set

RhodeCode will delete repos that are in database but not in the filesystem.

This command can be executed only using api_key belonging to user with admin

rights.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "rescan_repos"

    args :    {

                "remove_obsolete" : "<boolean = Optional(False)>"

              }

OUTPUT::

    id : <id_given_in_input>

    result : "{'added': [<list of names of added repos>],

               'removed': [<list of names of removed repos>]}"

    error :  null

invalidate_cache

----------------

Invalidate cache for repository.

This command can be executed only using api_key belonging to user with admin

rights or regular user that have write or admin or write access to repository.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "invalidate_cache"

    args :    {

                "repoid" : "<reponame or repo_id>"

              }

OUTPUT::

    id : <id_given_in_input>

    result : "Cache for repository `<reponame>` was invalidated: invalidated cache keys: <list_of_cache_keys>"

    error :  null

lock

----

Set locking state on given repository by given user. If userid param is skipped

This command can be executed only using api_key belonging to user with admin

rights or regular user that have admin or write access to repository.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "lock"

    args :    {

                "repoid" : "<reponame or repo_id>"

                "userid" : "<user_id or username = Optional(=apiuser)>",

              }

OUTPUT::

    id : <id_given_in_input>

    result : "User `<username>` set lock state for repo `<reponame>` to `true|false`"

    error :  null

show_ip

-------

Shows IP address as seen from RhodeCode server, together with all

defined IP addresses for given user.

This command can be executed only using api_key belonging to user with admin

rights.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "show_ip"

    args :    {

                "userid" : "<user_id or username>",

              }

OUTPUT::

    id : <id_given_in_input>

    result : {

                 "ip_addr_server": <ip_from_clien>",

                 "user_ips": [

                                {

                                   "ip_addr": "<ip_with_mask>",

                                   "ip_range": ["<start_ip>", "<end_ip>"],

                                },

                                ...

                             ]

             }

    error :  null

get_user

--------

Get's an user by username or user_id, Returns empty result if user is not found.

If userid param is skipped it is set to id of user who is calling this method.

# -*- coding: utf-8 -*-

"""

    rhodecode.controllers.api

    ~~~~~~~~~~~~~~~~~~~~~~~~~

    API controller for RhodeCode

    :created_on: Aug 20, 2011

    :author: marcink

    :copyright: (C) 2011-2012 Marcin Kuzminski <marcin@python-works.com>

    :license: GPLv3, see COPYING for more details.

"""

# This program is free software; you can redistribute it and/or

# modify it under the terms of the GNU General Public License

# as published by the Free Software Foundation; version 2

# of the License or (at your opinion) any later version of the license.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,

# MA  02110-1301, USA.

import traceback

import logging

from rhodecode.controllers.api import JSONRPCController, JSONRPCError

from rhodecode.lib.auth import PasswordGenerator, AuthUser, \

    HasPermissionAllDecorator, HasPermissionAnyDecorator, \

    HasPermissionAnyApi, HasRepoPermissionAnyApi

from rhodecode.lib.utils import map_groups, repo2db_mapper

from rhodecode.model.meta import Session

from rhodecode.model.scm import ScmModel

from rhodecode.model.repo import RepoModel

from rhodecode.model.user import UserModel

from rhodecode.model.users_group import UserGroupModel

from rhodecode.model.permission import PermissionModel

from rhodecode.model.db import Repository, RhodeCodeSetting, UserIpMap

log = logging.getLogger(__name__)

class OptionalAttr(object):

    """

    Special Optional Option that defines other attribute

    """

    def __init__(self, attr_name):

        self.attr_name = attr_name

    def __repr__(self):

        return '<OptionalAttr:%s>' % self.attr_name

    def __call__(self):

        return self

#alias

OAttr = OptionalAttr

class Optional(object):

    """

    Defines an optional parameter::

        param = param.getval() if isinstance(param, Optional) else param

        param = param() if isinstance(param, Optional) else param

    is equivalent of::

        param = Optional.extract(param)

    """

    def __init__(self, type_):

        self.type_ = type_

    def __repr__(self):

        return '<Optional:%s>' % self.type_.__repr__()

    def __call__(self):

        return self.getval()

    @HasPermissionAllDecorator('hg.admin')

    def rescan_repos(self, apiuser, remove_obsolete=Optional(False)):

        """

        Dispatch rescan repositories action. If remove_obsolete is set

        than also delete repos that are in database but not in the filesystem.

        aka "clean zombies"

        :param apiuser:

        :param remove_obsolete:

        """

        try:

            rm_obsolete = Optional.extract(remove_obsolete)

            added, removed = repo2db_mapper(ScmModel().repo_scan(),

                                            remove_obsolete=rm_obsolete)

            return {'added': added, 'removed': removed}

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'Error occurred during rescan repositories action'

            )

    def invalidate_cache(self, apiuser, repoid):

        """

        Dispatch cache invalidation action on given repo

        :param apiuser:

        :param repoid:

        """

        repo = get_repo_or_error(repoid)

        if HasPermissionAnyApi('hg.admin')(user=apiuser) is False:

            # check if we have admin permission for this repo !

            if HasRepoPermissionAnyApi('repository.admin',

                                       'repository.write')(user=apiuser,

                                            repo_name=repo.repo_name) is False:

                raise JSONRPCError('repository `%s` does not exist' % (repoid))

        try:

            invalidated_keys = ScmModel().mark_for_invalidation(repo.repo_name)

            Session().commit()

            return ('Cache for repository `%s` was invalidated: '

                    'invalidated cache keys: %s' % (repoid, invalidated_keys))

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'Error occurred during cache invalidation action'

            )

        """

        Set locking state on particular repository by given user, if

        this command is runned by non-admin account userid is set to user

        who is calling this method

        :param apiuser:

        :param repoid:

        :param userid:

        :param locked:

        """

        repo = get_repo_or_error(repoid)

        if HasPermissionAnyApi('hg.admin')(user=apiuser):

            pass

        elif HasRepoPermissionAnyApi('repository.admin',

                                     'repository.write')(user=apiuser,

                                                         repo_name=repo.repo_name):

            #make sure normal user does not pass someone else userid,

            #he is not allowed to do that

            if not isinstance(userid, Optional) and userid != apiuser.user_id:

                raise JSONRPCError(

                    'userid is not the same as your user'

                )

        else:

            raise JSONRPCError('repository `%s` does not exist' % (repoid))

        if isinstance(userid, Optional):

            userid = apiuser.user_id

        user = get_user_or_error(userid)

            else:

    @HasPermissionAllDecorator('hg.admin')

    def show_ip(self, apiuser, userid):

        """

        Shows IP address as seen from RhodeCode server, together with all

        defined IP addresses for given user

        :param apiuser:

        :param userid:

        """

        user = get_user_or_error(userid)

        ips = UserIpMap.query().filter(UserIpMap.user == user).all()

        return dict(

            ip_addr_server=self.ip_addr,

            user_ips=ips

        )

    def get_user(self, apiuser, userid=Optional(OAttr('apiuser'))):

        """"

        Get a user by username, or userid, if userid is given

        :param apiuser:

        :param userid:

        """

        if HasPermissionAnyApi('hg.admin')(user=apiuser) is False:

            #make sure normal user does not pass someone else userid,

            #he is not allowed to do that

            if not isinstance(userid, Optional) and userid != apiuser.user_id:

                raise JSONRPCError(

                    'userid is not the same as your user'

                )

        if isinstance(userid, Optional):

            userid = apiuser.user_id

        user = get_user_or_error(userid)

        data = user.get_api_data()

        data['permissions'] = AuthUser(user_id=user.user_id).permissions

        return data

    @HasPermissionAllDecorator('hg.admin')

    def get_users(self, apiuser):

        """"

        Get all users

        :param apiuser:

        """

        :param repo_name:

        """

        from rhodecode.lib.utils import is_valid_repo

        return is_valid_repo(repo_name, cls.base_path())

    def get_api_data(self):

        """

        Common function for generating repo api data

        """

        repo = self

        data = dict(

            repo_id=repo.repo_id,

            repo_name=repo.repo_name,

            repo_type=repo.repo_type,

            clone_uri=repo.clone_uri,

            private=repo.private,

            created_on=repo.created_on,

            description=repo.description,

            landing_rev=repo.landing_rev,

            owner=repo.user.username,

            fork_of=repo.fork.repo_name if repo.fork else None,

            enable_statistics=repo.enable_statistics,

            enable_locking=repo.enable_locking,

            enable_downloads=repo.enable_downloads,

            last_changeset=repo.changeset_cache

        )

        rc_config = RhodeCodeSetting.get_app_settings()

        repository_fields = str2bool(rc_config.get('rhodecode_repository_fields'))

        if repository_fields:

            for f in self.extra_fields:

                data[f.field_key_prefixed] = f.field_value

        return data

    @classmethod

    def lock(cls, repo, user_id):

        repo.locked = [user_id, time.time()]

        Session().add(repo)

        Session().commit()

    @classmethod

    def unlock(cls, repo):

        repo.locked = None

        Session().add(repo)

        Session().commit()

    @property

    def last_db_change(self):

        return self.updated_on

    def clone_url(self, **override):

        from pylons import url

        from urlparse import urlparse

        import urllib

        parsed_url = urlparse(url('home', qualified=True))

        default_clone_uri = '%(scheme)s://%(user)s%(pass)s%(netloc)s%(prefix)s%(path)s'

        decoded_path = safe_unicode(urllib.unquote(parsed_url.path))

        args = {

           'user': '',

           'pass': '',

           'scheme': parsed_url.scheme,

           'netloc': parsed_url.netloc,

           'prefix': decoded_path,

           'path': self.repo_name

        }

        args.update(override)

        return default_clone_uri % args

    #==========================================================================

    # SCM PROPERTIES

    #==========================================================================

    def get_changeset(self, rev=None):

        return get_changeset_safe(self.scm_instance, rev)

    def get_landing_changeset(self):

        """

        Returns landing changeset, or if that doesn't exist returns the tip

        """

        cs = self.get_changeset(self.landing_rev) or self.get_changeset()

        return cs

    def update_changeset_cache(self, cs_cache=None):

        """

        Update cache of last changeset for repository, keys should be::

            short_id

            raw_id

            revision

            message

            date

            author

            response = api_call(self, params)

            expected = ('User `%s` set lock state for repo `%s` to `%s`'

                       % (self.TEST_USER_LOGIN, repo_name, True))

            self._compare_ok(id_, expected, given=response.body)

        finally:

            destroy_repo(repo_name)

    def test_api_lock_repo_lock_aquire_non_admin_with_userid(self):

        repo_name = 'api_delete_me'

        create_repo(repo_name, self.REPO_TYPE, owner=self.TEST_USER_LOGIN)

        try:

            id_, params = _build_data(self.apikey_regular, 'lock',

                                      userid=TEST_USER_ADMIN_LOGIN,

                                      repoid=repo_name,

                                      locked=True)

            response = api_call(self, params)

            expected = 'userid is not the same as your user'

            self._compare_error(id_, expected, given=response.body)

        finally:

            destroy_repo(repo_name)

    def test_api_lock_repo_lock_aquire_non_admin_not_his_repo(self):

        id_, params = _build_data(self.apikey_regular, 'lock',

                                  repoid=self.REPO,

                                  locked=True)

        response = api_call(self, params)

        expected = 'repository `%s` does not exist' % (self.REPO)

        self._compare_error(id_, expected, given=response.body)

    def test_api_lock_repo_lock_release(self):

        id_, params = _build_data(self.apikey, 'lock',

                                  userid=TEST_USER_ADMIN_LOGIN,

                                  repoid=self.REPO,

                                  locked=False)

        response = api_call(self, params)

        expected = ('User `%s` set lock state for repo `%s` to `%s`'

                   % (TEST_USER_ADMIN_LOGIN, self.REPO, False))

        self._compare_ok(id_, expected, given=response.body)

    def test_api_lock_repo_lock_aquire_optional_userid(self):

        id_, params = _build_data(self.apikey, 'lock',

                                  repoid=self.REPO,

                                  locked=True)

        response = api_call(self, params)

        expected = ('User `%s` set lock state for repo `%s` to `%s`'

                   % (TEST_USER_ADMIN_LOGIN, self.REPO, True))

        self._compare_ok(id_, expected, given=response.body)

    @mock.patch.object(Repository, 'lock', crash)

    def test_api_lock_error(self):

        id_, params = _build_data(self.apikey, 'lock',

                                  userid=TEST_USER_ADMIN_LOGIN,

                                  repoid=self.REPO,

                                  locked=True)

        response = api_call(self, params)

        expected = 'Error occurred locking repository `%s`' % self.REPO

        self._compare_error(id_, expected, given=response.body)

    def test_api_create_existing_user(self):

        id_, params = _build_data(self.apikey, 'create_user',

                                  username=TEST_USER_ADMIN_LOGIN,

                                  email='test@foo.com',

                                  password='trololo')

        response = api_call(self, params)

        expected = "user `%s` already exist" % TEST_USER_ADMIN_LOGIN

        self._compare_error(id_, expected, given=response.body)

    def test_api_create_user_with_existing_email(self):

        id_, params = _build_data(self.apikey, 'create_user',

                                  username=TEST_USER_ADMIN_LOGIN + 'new',

                                  email=TEST_USER_REGULAR_EMAIL,

                                  password='trololo')

        response = api_call(self, params)

        expected = "email `%s` already exist" % TEST_USER_REGULAR_EMAIL

        self._compare_error(id_, expected, given=response.body)

    def test_api_create_user(self):

        username = 'test_new_api_user'

        email = username + "@foo.com"

        id_, params = _build_data(self.apikey, 'create_user',

                                  username=username,

                                  email=email,

                                  password='trololo')

        response = api_call(self, params)

        usr = UserModel().get_by_username(username)

        ret = dict(

            msg='created new user `%s`' % username,

            user=jsonify(usr.get_api_data())

        )

        expected = ret