^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Revoke permission for a user group on the given repository.

This command can only be executed using the api_key of a user with admin rights.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method  : "revoke_user_group_permission"

    args:     {

                "repoid" : "<reponame or repo_id>"

                "usersgroupid" : "<user group id or name>"

              }

OUTPUT::

    id : <id_given_in_input>

    result: {

              "msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",

              "success": true

            }

    error:  null

API access for web views

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

API access can also be turned on for each web view in Kallithea that is

decorated with the ``@LoginRequired`` decorator. Some views use

``@LoginRequired(api_access=True)`` and are always available. By default only

RSS/Atom feed views are enabled. Other views are

only available if they have been whitelisted. Edit the

``api_access_controllers_whitelist`` option in your .ini file and define views

that should have API access enabled.

For example, to enable API access to patch/diff, raw file and archive::

    api_access_controllers_whitelist =

        ChangesetController:changeset_patch,

        ChangesetController:changeset_raw,

        FilesController:raw,

        FilesController:archivefile

After this change, a Kallithea view can be accessed without login using

bearer authentication, by including this header with the request::

    Authentication: Bearer <api_key>

"""

import time

import traceback

import logging

from sqlalchemy import or_

from tg import request

from kallithea.controllers.api import JSONRPCController, JSONRPCError

from kallithea.lib.auth import (

    PasswordGenerator, AuthUser, HasPermissionAnyDecorator,

    HasPermissionAnyDecorator, HasPermissionAny, HasRepoPermissionLevel,

    HasRepoGroupPermissionLevel, HasUserGroupPermissionLevel)

from kallithea.lib.utils import map_groups, repo2db_mapper

from kallithea.lib.utils2 import (

    str2bool, time_to_datetime, safe_int, Optional, OAttr)

from kallithea.model.meta import Session

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.scm import ScmModel, UserGroupList

from kallithea.model.repo import RepoModel

from kallithea.model.user import UserModel

from kallithea.model.user_group import UserGroupModel

from kallithea.model.gist import GistModel

from kallithea.model.db import (

    Repository, Setting, UserIpMap, Permission, User, Gist,

    RepoGroup, UserGroup)

from kallithea.lib.compat import json

from kallithea.lib.exceptions import (

    DefaultUserException, UserGroupsAssignedException)

log = logging.getLogger(__name__)

def store_update(updates, attr, name):

    """

    Stores param in updates dict if it's not instance of Optional

    allows easy updates of passed in params

    """

    if not isinstance(attr, Optional):

        updates[name] = attr

def get_user_or_error(userid):

    """

    Get user by id or name or return JsonRPCError if not found

    :param userid:

    """

    user = UserModel().get_user(userid)

    if user is None:

        raise JSONRPCError("user `%s` does not exist" % (userid,))

    return user

          id : <id_given_in_input>

          result : null

          error :  {

            "failed to delete gist ID:<gist_id>"

          }

        """

        gist = get_gist_or_error(gistid)

        if not HasPermissionAny('hg.admin')():

            if gist.owner_id != request.authuser.user_id:

                raise JSONRPCError('gist `%s` does not exist' % (gistid,))

        try:

            GistModel().delete(gist)

            Session().commit()

            return dict(

                msg='deleted gist ID:%s' % (gist.gist_access_id,),

                gist=None

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError('failed to delete gist ID:%s'

                               % (gist.gist_access_id,))

    comment_id = Column('changeset_comment_id', Integer(), ForeignKey('changeset_comments.comment_id'), nullable=False)

    modified_at = Column(DateTime(), nullable=False, default=datetime.datetime.now)

    version = Column(Integer(), nullable=False, default=0)

    pull_request_id = Column(Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=True)

    author = relationship('User')

    repo = relationship('Repository')

    comment = relationship('ChangesetComment')

    pull_request = relationship('PullRequest')

    def __unicode__(self):

        return u"<%s('%s:%s')>" % (

            self.__class__.__name__,

            self.status, self.author

        )

    @classmethod

    def get_status_lbl(cls, value):

        return cls.STATUSES_DICT.get(value)

    @property

    def status_lbl(self):

        return ChangesetStatus.get_status_lbl(self.status)

class PullRequest(Base, BaseDbModel):

    __tablename__ = 'pull_requests'

    __table_args__ = (

        Index('pr_org_repo_id_idx', 'org_repo_id'),

        Index('pr_other_repo_id_idx', 'other_repo_id'),

        _table_args_default_dict,

    )

    # values for .status

    STATUS_NEW = u'new'

    STATUS_CLOSED = u'closed'

    pull_request_id = Column(Integer(), primary_key=True)

    title = Column(Unicode(255), nullable=False)

    description = Column(UnicodeText(), nullable=False)

    status = Column(Unicode(255), nullable=False, default=STATUS_NEW) # only for closedness, not approve/reject/etc

    created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    updated_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    owner_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)

    _revisions = Column('revisions', UnicodeText(), nullable=False)

    org_repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)

    org_ref = Column(Unicode(255), nullable=False)

    other_repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)

    @mock.patch.object(GistModel, 'delete', crash)

    def test_api_delete_gist_exception_occurred(self):

        gist_id = fixture.create_gist().gist_access_id

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

                                  gistid=gist_id)

        response = api_call(self, params)

        expected = 'failed to delete gist ID:%s' % (gist_id,)

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

    def test_api_get_ip(self):

        id_, params = _build_data(self.apikey, 'get_ip')

        response = api_call(self, params)

        expected = {

            'server_ip_addr': '0.0.0.0',

            'user_ips': []

        }

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

    def test_api_get_server_info(self):

        id_, params = _build_data(self.apikey, 'get_server_info')

        response = api_call(self, params)

        expected = Setting.get_server_info()

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

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

# 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, either version 3 of the License, or

# (at your option) any later version.

#

# 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, see <http://www.gnu.org/licenses/>.

from kallithea.tests.api.api_base import _BaseTestApi

class TestGitApi(_BaseTestApi, TestController):

    REPO = GIT_REPO

    REPO_TYPE = 'git'

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

# 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, either version 3 of the License, or

# (at your option) any later version.

#

# 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, see <http://www.gnu.org/licenses/>.

from kallithea.tests.api.api_base import _BaseTestApi

class TestHgApi(_BaseTestApi, TestController):

    REPO = HG_REPO

    REPO_TYPE = 'hg'

from kallithea.model.db import Notification, User, UserNotification

from kallithea.model.meta import Session

from kallithea.lib.utils2 import safe_str

log = logging.getLogger(__name__)

skipif = pytest.mark.skipif

parametrize = pytest.mark.parametrize

# Hack: These module global values MUST be set to actual values before running any tests. This is currently done by conftest.py.

url = None

testapp = None

__all__ = [

    'skipif', 'parametrize', 'environ', 'url', 'TestController',

    'ldap_lib_installed', 'pam_lib_installed', 'invalidate_all_caches',

    'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',

    'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',

    'TEST_USER_ADMIN_EMAIL', 'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',

    'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',

    'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',

    'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',

    'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',

]

# Invoke websetup with the current config file

# SetupCommand('setup-app').run([config_file])

environ = {}

#SOME GLOBALS FOR TESTS

TESTS_TMP_PATH = os.path.join(tempfile.gettempdir(), 'rc_test_%s' % tempfile._RandomNameSequence().next())

TEST_USER_ADMIN_LOGIN = 'test_admin'

TEST_USER_ADMIN_PASS = 'test12'

TEST_USER_ADMIN_EMAIL = 'test_admin@example.com'

TEST_USER_REGULAR_LOGIN = 'test_regular'

TEST_USER_REGULAR_PASS = 'test12'

TEST_USER_REGULAR_EMAIL = 'test_regular@example.com'

TEST_USER_REGULAR2_LOGIN = 'test_regular2'

TEST_USER_REGULAR2_PASS = 'test12'

TEST_USER_REGULAR2_EMAIL = 'test_regular2@example.com'

HG_REPO = u'vcs_test_hg'

GIT_REPO = u'vcs_test_git'

NEW_HG_REPO = u'vcs_test_hg_new'

NEW_GIT_REPO = u'vcs_test_git_new'

HG_FORK = u'vcs_test_hg_fork'

GIT_FORK = u'vcs_test_git_fork'

## VCS

SCM_TESTS = ['hg', 'git']

uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))

GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'

TEST_GIT_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)

TEST_GIT_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)

TEST_GIT_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)

HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'

TEST_HG_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)

TEST_HG_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)

TEST_HG_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)

TEST_DIR = tempfile.gettempdir()

TEST_REPO_PREFIX = 'vcs-test'

# cached repos if any !

# comment out to get some other repos from bb or github

GIT_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)

HG_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)

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

# 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, either version 3 of the License, or

# (at your option) any later version.

#

# 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, see <http://www.gnu.org/licenses/>.

"""

Helpers for fixture generation

"""

import logging

import os

import shutil

import tarfile

from os.path import dirname

from kallithea.model.meta import Session

from kallithea.model.repo import RepoModel

from kallithea.model.user import UserModel

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.user_group import UserGroupModel

from kallithea.model.gist import GistModel

from kallithea.model.scm import ScmModel

from kallithea.lib.db_manage import DbManage

from kallithea.lib.vcs.backends.base import EmptyChangeset

from kallithea.tests.base import invalidate_all_caches, GIT_REPO, HG_REPO, TESTS_TMP_PATH, TEST_USER_ADMIN_LOGIN

log = logging.getLogger(__name__)

FIXTURES = os.path.join(dirname(dirname(os.path.abspath(__file__))), 'tests', 'fixtures')

def error_function(*args, **kwargs):

    raise Exception('Total Crash !')

class Fixture(object):

    def __init__(self):

        pass

    def anon_access(self, status):

        """

        Context manager for controlling anonymous access.

        Anon access will be set and committed, but restored again when exiting the block.

        if newfile:

            nodes = {

                filename: {

                    'content': content

                }

            }

            cs = ScmModel().create_nodes(

                user=TEST_USER_ADMIN_LOGIN, repo=repo,

                message=message,

                nodes=nodes,

                parent_cs=_cs,

                author=author,

            )

        else:

            cs = ScmModel().commit_change(

                repo=repo.scm_instance, repo_name=repo.repo_name,

                cs=parent, user=TEST_USER_ADMIN_LOGIN,

                author=author,

                message=message,

                content=content,

                f_path=filename

            )

        return cs

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

# Global test environment setup

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

def create_test_env(repos_test_path, config):

    """

    Makes a fresh database and

    install test repository into tmp dir

    """

    # PART ONE create db

    dbconf = config['sqlalchemy.url']

    log.debug('making test db %s', dbconf)

    # create test dir if it doesn't exist

    if not os.path.isdir(repos_test_path):

        log.debug('Creating testdir %s', repos_test_path)

        os.makedirs(repos_test_path)

    dbmanage = DbManage(dbconf=dbconf, root=config['here'],

                        tests=True)

    dbmanage.create_tables(override=True)

    # for tests dynamically set new root paths based on generated content