Changeset - 1de3e5d104bf
[Not reviewed]
default
0 3 0
Mads Kiilerich (mads) - 6 years ago 2020-06-24 17:12:28
mads@kiilerich.com
Grafted from: 0e14bbd2a0e7
tests: consistent use of more helpful raise_exception "fixture"

"Total Crash" was too misleading when debugging problems.
3 files changed with 29 insertions and 32 deletions:
0 comments (0 inline, 0 general)
kallithea/tests/api/api_base.py
Show inline comments
 
@@ -15,79 +15,76 @@
 
"""
 
Tests for the JSON-RPC web api.
 
"""
 

	
 
import os
 
import random
 
import re
 

	
 
import mock
 
import pytest
 

	
 
from kallithea.lib import ext_json
 
from kallithea.lib.auth import AuthUser
 
from kallithea.lib.utils2 import ascii_bytes
 
from kallithea.model.changeset_status import ChangesetStatusModel
 
from kallithea.model.db import ChangesetStatus, PullRequest, RepoGroup, Repository, Setting, Ui, User
 
from kallithea.model.gist import GistModel
 
from kallithea.model.meta import Session
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.scm import ScmModel
 
from kallithea.model.user import UserModel
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.tests import base
 
from kallithea.tests.fixture import Fixture
 
from kallithea.tests.fixture import Fixture, raise_exception
 

	
 

	
 

	
 
API_URL = '/_admin/api'
 
TEST_USER_GROUP = 'test_user_group'
 
TEST_REPO_GROUP = 'test_repo_group'
 

	
 
fixture = Fixture()
 

	
 

	
 
def _build_data(apikey, method, **kw):
 
    """
 
    Builds API data with given random ID
 
    For convenience, the json is returned as str
 
    """
 
    random_id = random.randrange(1, 9999)
 
    return random_id, ext_json.dumps({
 
        "id": random_id,
 
        "api_key": apikey,
 
        "method": method,
 
        "args": kw
 
    })
 

	
 

	
 
jsonify = lambda obj: ext_json.loads(ext_json.dumps(obj))
 

	
 

	
 
def crash(*args, **kwargs):
 
    raise Exception('Total Crash !')
 

	
 

	
 
def api_call(test_obj, params):
 
    response = test_obj.app.post(API_URL, content_type='application/json',
 
                                 params=params)
 
    return response
 

	
 

	
 
## helpers
 
def make_user_group(name=TEST_USER_GROUP):
 
    gr = fixture.create_user_group(name, cur_user=base.TEST_USER_ADMIN_LOGIN)
 
    UserGroupModel().add_user_to_group(user_group=gr,
 
                                       user=base.TEST_USER_ADMIN_LOGIN)
 
    Session().commit()
 
    return gr
 

	
 

	
 
def make_repo_group(name=TEST_REPO_GROUP):
 
    gr = fixture.create_repo_group(name, cur_user=base.TEST_USER_ADMIN_LOGIN)
 
    Session().commit()
 
    return gr
 

	
 

	
 
class _BaseTestApi(object):
 
    REPO = None
 
    REPO_TYPE = None
 
@@ -328,49 +325,49 @@ class _BaseTestApi(object):
 
    def test_api_pull_custom_remote(self):
 
        repo_name = 'test_pull_custom_remote'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 

	
 
        custom_remote_path = os.path.join(Ui.get_by_key('paths', '/').ui_value, self.REPO)
 

	
 
        id_, params = _build_data(self.apikey, 'pull',
 
                                  repoid=repo_name,
 
                                  clone_uri=custom_remote_path)
 
        response = api_call(self, params)
 

	
 
        expected = {'msg': 'Pulled from `%s`' % repo_name,
 
                    'repository': repo_name}
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
        fixture.destroy_repo(repo_name)
 

	
 
    def test_api_rescan_repos(self):
 
        id_, params = _build_data(self.apikey, 'rescan_repos')
 
        response = api_call(self, params)
 

	
 
        expected = {'added': [], 'removed': []}
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(ScmModel, 'repo_scan', crash)
 
    @mock.patch.object(ScmModel, 'repo_scan', raise_exception)
 
    def test_api_rescann_error(self):
 
        id_, params = _build_data(self.apikey, 'rescan_repos', )
 
        response = api_call(self, params)
 

	
 
        expected = 'Error occurred during rescan repositories action'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_create_existing_user(self):
 
        id_, params = _build_data(self.apikey, 'create_user',
 
                                  username=base.TEST_USER_ADMIN_LOGIN,
 
                                  email='test@example.com',
 
                                  password='trololo')
 
        response = api_call(self, params)
 

	
 
        expected = "user `%s` already exist" % base.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=base.TEST_USER_ADMIN_LOGIN + 'new',
 
                                  email=base.TEST_USER_REGULAR_EMAIL,
 
                                  password='trololo')
 
        response = api_call(self, params)
 

	
 
@@ -418,83 +415,83 @@ class _BaseTestApi(object):
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user(usr.user_id)
 

	
 
    def test_api_create_user_with_extern_name(self):
 
        username = 'test_new_api_user_passwordless'
 
        email = username + "@example.com"
 

	
 
        id_, params = _build_data(self.apikey, 'create_user',
 
                                  username=username,
 
                                  email=email, extern_name='internal')
 
        response = api_call(self, params)
 

	
 
        usr = User.get_by_username(username)
 
        ret = dict(
 
            msg='created new user `%s`' % username,
 
            user=jsonify(usr.get_api_data())
 
        )
 
        try:
 
            expected = ret
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user(usr.user_id)
 

	
 
    @mock.patch.object(UserModel, 'create_or_update', crash)
 
    @mock.patch.object(UserModel, 'create_or_update', raise_exception)
 
    def test_api_create_user_when_exception_happened(self):
 

	
 
        username = 'test_new_api_user'
 
        email = username + "@example.com"
 

	
 
        id_, params = _build_data(self.apikey, 'create_user',
 
                                  username=username,
 
                                  email=email,
 
                                  password='trololo')
 
        response = api_call(self, params)
 
        expected = 'failed to create user `%s`' % username
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_delete_user(self):
 
        usr = UserModel().create_or_update(username='test_user',
 
                                           password='qweqwe',
 
                                           email='u232@example.com',
 
                                           firstname='u1', lastname='u1')
 
        Session().commit()
 
        username = usr.username
 
        email = usr.email
 
        usr_id = usr.user_id
 
        ## DELETE THIS USER NOW
 

	
 
        id_, params = _build_data(self.apikey, 'delete_user',
 
                                  userid=username, )
 
        response = api_call(self, params)
 

	
 
        ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),
 
               'user': None}
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(UserModel, 'delete', crash)
 
    @mock.patch.object(UserModel, 'delete', raise_exception)
 
    def test_api_delete_user_when_exception_happened(self):
 
        usr = UserModel().create_or_update(username='test_user',
 
                                           password='qweqwe',
 
                                           email='u232@example.com',
 
                                           firstname='u1', lastname='u1')
 
        Session().commit()
 
        username = usr.username
 

	
 
        id_, params = _build_data(self.apikey, 'delete_user',
 
                                  userid=username, )
 
        response = api_call(self, params)
 
        ret = 'failed to delete user ID:%s %s' % (usr.user_id,
 
                                                  usr.username)
 
        expected = ret
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,expected', [
 
        ('firstname', 'new_username'),
 
        ('lastname', 'new_username'),
 
        ('email', 'new_username'),
 
        ('admin', True),
 
        ('admin', False),
 
        ('extern_type', 'ldap'),
 
        ('extern_type', None),
 
@@ -540,49 +537,49 @@ class _BaseTestApi(object):
 
    def test_api_update_user_by_user_id(self):
 
        usr = User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
 
        ret = jsonify(usr.get_api_data())
 
        id_, params = _build_data(self.apikey, 'update_user',
 
                                  userid=usr.user_id)
 

	
 
        response = api_call(self, params)
 
        ret = {
 
            'msg': 'updated user ID:%s %s' % (
 
                usr.user_id, base.TEST_USER_ADMIN_LOGIN),
 
            'user': ret
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_update_user_default_user(self):
 
        usr = User.get_default_user()
 
        id_, params = _build_data(self.apikey, 'update_user',
 
                                  userid=usr.user_id)
 

	
 
        response = api_call(self, params)
 
        expected = 'editing default user is forbidden'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(UserModel, 'update_user', crash)
 
    @mock.patch.object(UserModel, 'update_user', raise_exception)
 
    def test_api_update_user_when_exception_happens(self):
 
        usr = User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
 
        ret = jsonify(usr.get_api_data())
 
        id_, params = _build_data(self.apikey, 'update_user',
 
                                  userid=usr.user_id)
 

	
 
        response = api_call(self, params)
 
        ret = 'failed to update user `%s`' % usr.user_id
 

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

	
 
    def test_api_get_repo(self):
 
        new_group = 'some_new_group'
 
        make_user_group(new_group)
 
        RepoModel().grant_user_group_permission(repo=self.REPO,
 
                                                group_name=new_group,
 
                                                perm='repository.read')
 
        Session().commit()
 
        id_, params = _build_data(self.apikey, 'get_repo',
 
                                  repoid=self.REPO)
 
        response = api_call(self, params)
 
        assert "tags" not in response.json['result']
 
        assert 'pull_requests' not in response.json['result']
 
@@ -999,49 +996,49 @@ class _BaseTestApi(object):
 

	
 
    def test_api_create_repo_exists(self):
 
        repo_name = self.REPO
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,)
 
        response = api_call(self, params)
 
        expected = "repo `%s` already exist" % repo_name
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_create_repo_dot_dot(self):
 
        # it is only possible to create repositories in existing repo groups - and '..' can't be used
 
        group_name = '%s/..' % TEST_REPO_GROUP
 
        repo_name = '%s/%s' % (group_name, 'could-be-outside')
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,)
 
        response = api_call(self, params)
 
        expected = 'repo group `%s` not found' % group_name
 
        self._compare_error(id_, expected, given=response.body)
 
        fixture.destroy_repo(repo_name)
 

	
 
    @mock.patch.object(RepoModel, 'create', crash)
 
    @mock.patch.object(RepoModel, 'create', raise_exception)
 
    def test_api_create_repo_exception_occurred(self):
 
        repo_name = 'api-repo'
 
        id_, params = _build_data(self.apikey, 'create_repo',
 
                                  repo_name=repo_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
                                  repo_type=self.REPO_TYPE,)
 
        response = api_call(self, params)
 
        expected = 'failed to create repository `%s`' % repo_name
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('changing_attr,updates', [
 
        ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
 
        ('description', {'description': 'new description'}),
 
        ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail
 
        ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a mis-feature - it would bypass access control
 
        ('clone_uri', {'clone_uri': None}),
 
        ('landing_rev', {'landing_rev': 'branch:master'}),
 
        ('enable_statistics', {'enable_statistics': True}),
 
        ('enable_downloads', {'enable_downloads': True}),
 
        ('name', {'name': 'new_repo_name'}),
 
        ('repo_group', {'group': 'test_group_for_update'}),
 
    ])
 
    def test_api_update_repo(self, changing_attr, updates):
 
        repo_name = 'api_update_me'
 
@@ -1119,49 +1116,49 @@ class _BaseTestApi(object):
 
        fixture.create_repo(repo_name)
 
        updates = {'group': 'test_group_for_update'}
 
        id_, params = _build_data(self.apikey, 'update_repo',
 
                                  repoid=repo_name, **updates)
 
        response = api_call(self, params)
 
        try:
 
            expected = 'repository group `%s` does not exist' % updates['group']
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_update_repo_regular_user_not_allowed(self):
 
        repo_name = 'admin_owned'
 
        fixture.create_repo(repo_name)
 
        updates = {'description': 'something else'}
 
        id_, params = _build_data(self.apikey_regular, 'update_repo',
 
                                  repoid=repo_name, **updates)
 
        response = api_call(self, params)
 
        try:
 
            expected = 'repository `%s` does not exist' % repo_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    @mock.patch.object(RepoModel, 'update', crash)
 
    @mock.patch.object(RepoModel, 'update', raise_exception)
 
    def test_api_update_repo_exception_occurred(self):
 
        repo_name = 'api_update_me'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        id_, params = _build_data(self.apikey, 'update_repo',
 
                                  repoid=repo_name, owner=base.TEST_USER_ADMIN_LOGIN,)
 
        response = api_call(self, params)
 
        try:
 
            expected = 'failed to update repo `%s`' % repo_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_update_repo_regular_user_change_repo_name(self):
 
        repo_name = 'admin_owned'
 
        new_repo_name = 'new_repo_name'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        RepoModel().grant_user_permission(repo=repo_name,
 
                                          user=self.TEST_USER_LOGIN,
 
                                          perm='repository.admin')
 
        UserModel().revoke_perm('default', 'hg.create.repository')
 
        UserModel().grant_perm('default', 'hg.create.none')
 
        updates = {'name': new_repo_name}
 
        id_, params = _build_data(self.apikey_regular, 'update_repo',
 
                                  repoid=repo_name, **updates)
 
@@ -1243,49 +1240,49 @@ class _BaseTestApi(object):
 
            'success': True
 
        }
 
        try:
 
            expected = ret
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_delete_repo_by_non_admin_no_permission(self):
 
        repo_name = 'api_delete_me'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        try:
 
            id_, params = _build_data(self.apikey_regular, 'delete_repo',
 
                                      repoid=repo_name, )
 
            response = api_call(self, params)
 
            expected = 'repository `%s` does not exist' % (repo_name)
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_delete_repo_exception_occurred(self):
 
        repo_name = 'api_delete_me'
 
        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
        try:
 
            with mock.patch.object(RepoModel, 'delete', crash):
 
            with mock.patch.object(RepoModel, 'delete', raise_exception):
 
                id_, params = _build_data(self.apikey, 'delete_repo',
 
                                          repoid=repo_name, )
 
                response = api_call(self, params)
 

	
 
                expected = 'failed to delete repository `%s`' % repo_name
 
                self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(repo_name)
 

	
 
    def test_api_fork_repo(self):
 
        fork_name = 'api-repo-fork'
 
        id_, params = _build_data(self.apikey, 'fork_repo',
 
                                  repoid=self.REPO,
 
                                  fork_name=fork_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
        )
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
 
                                                     fork_name),
 
            'success': True,
 
            'task': None,
 
        }
 
@@ -1391,49 +1388,49 @@ class _BaseTestApi(object):
 
                                      repoid=self.REPO,
 
                                      fork_name=fork_name,
 
                                      owner=base.TEST_USER_ADMIN_LOGIN,
 
            )
 
            response = api_call(self, params)
 

	
 
            expected = "fork `%s` already exist" % fork_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_repo(fork_name)
 

	
 
    def test_api_fork_repo_repo_exists(self):
 
        fork_name = self.REPO
 

	
 
        id_, params = _build_data(self.apikey, 'fork_repo',
 
                                  repoid=self.REPO,
 
                                  fork_name=fork_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
        )
 
        response = api_call(self, params)
 

	
 
        expected = "repo `%s` already exist" % fork_name
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'create_fork', crash)
 
    @mock.patch.object(RepoModel, 'create_fork', raise_exception)
 
    def test_api_fork_repo_exception_occurred(self):
 
        fork_name = 'api-repo-fork'
 
        id_, params = _build_data(self.apikey, 'fork_repo',
 
                                  repoid=self.REPO,
 
                                  fork_name=fork_name,
 
                                  owner=base.TEST_USER_ADMIN_LOGIN,
 
        )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to fork repository `%s` as `%s`' % (self.REPO,
 
                                                               fork_name)
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_get_user_group(self):
 
        id_, params = _build_data(self.apikey, 'get_user_group',
 
                                  usergroupid=TEST_USER_GROUP)
 
        response = api_call(self, params)
 

	
 
        user_group = UserGroupModel().get_group(TEST_USER_GROUP)
 
        members = []
 
        for user in user_group.members:
 
            user = user.user
 
            members.append(user.get_api_data())
 

	
 
@@ -1463,157 +1460,157 @@ class _BaseTestApi(object):
 
        group_name = 'some_new_group'
 
        id_, params = _build_data(self.apikey, 'create_user_group',
 
                                  group_name=group_name)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'created new user group `%s`' % group_name,
 
            'user_group': jsonify(UserGroupModel() \
 
                .get_by_name(group_name) \
 
                .get_api_data())
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
        fixture.destroy_user_group(group_name)
 

	
 
    def test_api_get_user_group_that_exist(self):
 
        id_, params = _build_data(self.apikey, 'create_user_group',
 
                                  group_name=TEST_USER_GROUP)
 
        response = api_call(self, params)
 

	
 
        expected = "user group `%s` already exist" % TEST_USER_GROUP
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(UserGroupModel, 'create', crash)
 
    @mock.patch.object(UserGroupModel, 'create', raise_exception)
 
    def test_api_get_user_group_exception_occurred(self):
 
        group_name = 'exception_happens'
 
        id_, params = _build_data(self.apikey, 'create_user_group',
 
                                  group_name=group_name)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to create group `%s`' % group_name
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('changing_attr,updates', [
 
        ('group_name', {'group_name': 'new_group_name'}),
 
        ('group_name', {'group_name': 'test_group_for_update'}),
 
        ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
 
        ('active', {'active': False}),
 
        ('active', {'active': True}),
 
    ])
 
    def test_api_update_user_group(self, changing_attr, updates):
 
        gr_name = 'test_group_for_update'
 
        user_group = fixture.create_user_group(gr_name)
 
        try:
 
            id_, params = _build_data(self.apikey, 'update_user_group',
 
                                      usergroupid=gr_name, **updates)
 
            response = api_call(self, params)
 
            expected = {
 
               'msg': 'updated user group ID:%s %s' % (user_group.users_group_id,
 
                                                     user_group.users_group_name),
 
               'user_group': user_group.get_api_data()
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            if changing_attr == 'group_name':
 
                # switch to updated name for proper cleanup
 
                gr_name = updates['group_name']
 
            fixture.destroy_user_group(gr_name)
 

	
 
    @mock.patch.object(UserGroupModel, 'update', crash)
 
    @mock.patch.object(UserGroupModel, 'update', raise_exception)
 
    def test_api_update_user_group_exception_occurred(self):
 
        gr_name = 'test_group'
 
        fixture.create_user_group(gr_name)
 
        try:
 
            id_, params = _build_data(self.apikey, 'update_user_group',
 
                                      usergroupid=gr_name)
 
            response = api_call(self, params)
 
            expected = 'failed to update user group `%s`' % gr_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_add_user_to_user_group(self):
 
        gr_name = 'test_group'
 
        fixture.create_user_group(gr_name)
 
        try:
 
            id_, params = _build_data(self.apikey, 'add_user_to_user_group',
 
                                      usergroupid=gr_name,
 
                                      userid=base.TEST_USER_ADMIN_LOGIN)
 
            response = api_call(self, params)
 
            expected = {
 
            'msg': 'added member `%s` to user group `%s`' % (
 
                    base.TEST_USER_ADMIN_LOGIN, gr_name),
 
            'success': True
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_add_user_to_user_group_that_doesnt_exist(self):
 
        id_, params = _build_data(self.apikey, 'add_user_to_user_group',
 
                                  usergroupid='false-group',
 
                                  userid=base.TEST_USER_ADMIN_LOGIN)
 
        response = api_call(self, params)
 

	
 
        expected = 'user group `%s` does not exist' % 'false-group'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(UserGroupModel, 'add_user_to_group', crash)
 
    @mock.patch.object(UserGroupModel, 'add_user_to_group', raise_exception)
 
    def test_api_add_user_to_user_group_exception_occurred(self):
 
        gr_name = 'test_group'
 
        fixture.create_user_group(gr_name)
 
        try:
 
            id_, params = _build_data(self.apikey, 'add_user_to_user_group',
 
                                      usergroupid=gr_name,
 
                                      userid=base.TEST_USER_ADMIN_LOGIN)
 
            response = api_call(self, params)
 
            expected = 'failed to add member to user group `%s`' % gr_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_remove_user_from_user_group(self):
 
        gr_name = 'test_group_3'
 
        gr = fixture.create_user_group(gr_name)
 
        UserGroupModel().add_user_to_group(gr, user=base.TEST_USER_ADMIN_LOGIN)
 
        Session().commit()
 
        try:
 
            id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
 
                                      usergroupid=gr_name,
 
                                      userid=base.TEST_USER_ADMIN_LOGIN)
 
            response = api_call(self, params)
 
            expected = {
 
                'msg': 'removed member `%s` from user group `%s`' % (
 
                    base.TEST_USER_ADMIN_LOGIN, gr_name
 
                ),
 
                'success': True}
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    @mock.patch.object(UserGroupModel, 'remove_user_from_group', crash)
 
    @mock.patch.object(UserGroupModel, 'remove_user_from_group', raise_exception)
 
    def test_api_remove_user_from_user_group_exception_occurred(self):
 
        gr_name = 'test_group_3'
 
        gr = fixture.create_user_group(gr_name)
 
        UserGroupModel().add_user_to_group(gr, user=base.TEST_USER_ADMIN_LOGIN)
 
        Session().commit()
 
        try:
 
            id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
 
                                      usergroupid=gr_name,
 
                                      userid=base.TEST_USER_ADMIN_LOGIN)
 
            response = api_call(self, params)
 
            expected = 'failed to remove member from user group `%s`' % gr_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_delete_user_group(self):
 
        gr_name = 'test_group'
 
        ugroup = fixture.create_user_group(gr_name)
 
        gr_id = ugroup.users_group_id
 
        try:
 
            id_, params = _build_data(self.apikey, 'delete_user_group',
 
                                      usergroupid=gr_name)
 
            response = api_call(self, params)
 
            expected = {
 
@@ -1630,203 +1627,203 @@ class _BaseTestApi(object):
 
        ugroup = fixture.create_user_group(gr_name)
 
        gr_id = ugroup.users_group_id
 

	
 
        ugr_to_perm = RepoModel().grant_user_group_permission(self.REPO, gr_name, 'repository.write')
 
        msg = 'User Group assigned to %s' % ugr_to_perm.repository.repo_name
 

	
 
        try:
 
            id_, params = _build_data(self.apikey, 'delete_user_group',
 
                                      usergroupid=gr_name)
 
            response = api_call(self, params)
 
            expected = msg
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            if UserGroupModel().get_by_name(gr_name):
 
                fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_delete_user_group_exception_occurred(self):
 
        gr_name = 'test_group'
 
        ugroup = fixture.create_user_group(gr_name)
 
        gr_id = ugroup.users_group_id
 
        id_, params = _build_data(self.apikey, 'delete_user_group',
 
                                  usergroupid=gr_name)
 

	
 
        try:
 
            with mock.patch.object(UserGroupModel, 'delete', crash):
 
            with mock.patch.object(UserGroupModel, 'delete', raise_exception):
 
                response = api_call(self, params)
 
                expected = 'failed to delete user group ID:%s %s' % (gr_id, gr_name)
 
                self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    @base.parametrize('name,perm', [
 
        ('none', 'repository.none'),
 
        ('read', 'repository.read'),
 
        ('write', 'repository.write'),
 
        ('admin', 'repository.admin'),
 
    ])
 
    def test_api_grant_user_permission(self, name, perm):
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
 
                perm, base.TEST_USER_ADMIN_LOGIN, self.REPO
 
            ),
 
            'success': True
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_grant_user_permission_wrong_permission(self):
 
        perm = 'haha.no.permission'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'permission `%s` does not exist' % perm
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'grant_user_permission', crash)
 
    @mock.patch.object(RepoModel, 'grant_user_permission', raise_exception)
 
    def test_api_grant_user_permission_exception_when_adding(self):
 
        perm = 'repository.read'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
 
            base.TEST_USER_ADMIN_LOGIN, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_revoke_user_permission(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN, )
 
        response = api_call(self, params)
 

	
 
        expected = {
 
            'msg': 'Revoked perm for user: `%s` in repo: `%s`' % (
 
                base.TEST_USER_ADMIN_LOGIN, self.REPO
 
            ),
 
            'success': True
 
        }
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'revoke_user_permission', crash)
 
    @mock.patch.object(RepoModel, 'revoke_user_permission', raise_exception)
 
    def test_api_revoke_user_permission_exception_when_adding(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN, )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
 
            base.TEST_USER_ADMIN_LOGIN, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,perm', [
 
        ('none', 'repository.none'),
 
        ('read', 'repository.read'),
 
        ('write', 'repository.write'),
 
        ('admin', 'repository.admin'),
 
    ])
 
    def test_api_grant_user_group_permission(self, name, perm):
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Granted perm: `%s` for user group: `%s` in repo: `%s`' % (
 
                perm, TEST_USER_GROUP, self.REPO
 
            ),
 
            'success': True
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_grant_user_group_permission_wrong_permission(self):
 
        perm = 'haha.no.permission'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'permission `%s` does not exist' % perm
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'grant_user_group_permission', crash)
 
    @mock.patch.object(RepoModel, 'grant_user_group_permission', raise_exception)
 
    def test_api_grant_user_group_permission_exception_when_adding(self):
 
        perm = 'repository.read'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user group: `%s` in repo: `%s`' % (
 
            TEST_USER_GROUP, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_revoke_user_group_permission(self):
 
        RepoModel().grant_user_group_permission(repo=self.REPO,
 
                                                group_name=TEST_USER_GROUP,
 
                                                perm='repository.read')
 
        Session().commit()
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP, )
 
        response = api_call(self, params)
 

	
 
        expected = {
 
            'msg': 'Revoked perm for user group: `%s` in repo: `%s`' % (
 
                TEST_USER_GROUP, self.REPO
 
            ),
 
            'success': True
 
        }
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'revoke_user_group_permission', crash)
 
    @mock.patch.object(RepoModel, 'revoke_user_group_permission', raise_exception)
 
    def test_api_revoke_user_group_permission_exception_when_adding(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP, )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user group: `%s` in repo: `%s`' % (
 
            TEST_USER_GROUP, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,perm,apply_to_children', [
 
        ('none', 'group.none', 'none'),
 
        ('read', 'group.read', 'none'),
 
        ('write', 'group.write', 'none'),
 
        ('admin', 'group.admin', 'none'),
 

	
 
        ('none', 'group.none', 'all'),
 
        ('read', 'group.read', 'all'),
 
        ('write', 'group.write', 'all'),
 
        ('admin', 'group.admin', 'all'),
 

	
 
        ('none', 'group.none', 'repos'),
 
@@ -1886,49 +1883,49 @@ class _BaseTestApi(object):
 
            ret = {
 
                'msg': 'Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (
 
                    perm, apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
                ),
 
                'success': True
 
            }
 
            expected = ret
 
            self._compare_ok(id_, expected, given=response.body)
 
        else:
 
            expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
 
            self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_grant_user_permission_to_repo_group_wrong_permission(self):
 
        perm = 'haha.no.permission'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission_to_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'permission `%s` does not exist' % perm
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoGroupModel, 'grant_user_permission', crash)
 
    @mock.patch.object(RepoGroupModel, 'grant_user_permission', raise_exception)
 
    def test_api_grant_user_permission_to_repo_group_exception_when_adding(self):
 
        perm = 'group.read'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission_to_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user: `%s` in repo group: `%s`' % (
 
            base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,apply_to_children', [
 
        ('none', 'none'),
 
        ('all', 'all'),
 
        ('repos', 'repos'),
 
        ('groups', 'groups'),
 
    ])
 
    def test_api_revoke_user_permission_from_repo_group(self, name, apply_to_children):
 
        RepoGroupModel().grant_user_permission(repo_group=TEST_REPO_GROUP,
 
                                               user=base.TEST_USER_ADMIN_LOGIN,
 
                                               perm='group.read',)
 
@@ -1971,49 +1968,49 @@ class _BaseTestApi(object):
 
        if grant_admin:
 
            RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
 
                                                   self.TEST_USER_LOGIN,
 
                                                   'group.admin')
 
            Session().commit()
 

	
 
        id_, params = _build_data(self.apikey_regular,
 
                                  'revoke_user_permission_from_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN,
 
                                  apply_to_children=apply_to_children,)
 
        response = api_call(self, params)
 
        if access_ok:
 
            expected = {
 
                'msg': 'Revoked perm (recursive:%s) for user: `%s` in repo group: `%s`' % (
 
                    apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
                ),
 
                'success': True
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        else:
 
            expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
 
            self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoGroupModel, 'revoke_user_permission', crash)
 
    @mock.patch.object(RepoGroupModel, 'revoke_user_permission', raise_exception)
 
    def test_api_revoke_user_permission_from_repo_group_exception_when_adding(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_permission_from_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  userid=base.TEST_USER_ADMIN_LOGIN, )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user: `%s` in repo group: `%s`' % (
 
            base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,perm,apply_to_children', [
 
        ('none', 'group.none', 'none'),
 
        ('read', 'group.read', 'none'),
 
        ('write', 'group.write', 'none'),
 
        ('admin', 'group.admin', 'none'),
 

	
 
        ('none', 'group.none', 'all'),
 
        ('read', 'group.read', 'all'),
 
        ('write', 'group.write', 'all'),
 
        ('admin', 'group.admin', 'all'),
 

	
 
        ('none', 'group.none', 'repos'),
 
@@ -2075,49 +2072,49 @@ class _BaseTestApi(object):
 
            ret = {
 
                'msg': 'Granted perm: `%s` (recursive:%s) for user group: `%s` in repo group: `%s`' % (
 
                    perm, apply_to_children, TEST_USER_GROUP, TEST_REPO_GROUP
 
                ),
 
                'success': True
 
            }
 
            expected = ret
 
            self._compare_ok(id_, expected, given=response.body)
 
        else:
 
            expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
 
            self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_grant_user_group_permission_to_repo_group_wrong_permission(self):
 
        perm = 'haha.no.permission'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission_to_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'permission `%s` does not exist' % perm
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoGroupModel, 'grant_user_group_permission', crash)
 
    @mock.patch.object(RepoGroupModel, 'grant_user_group_permission', raise_exception)
 
    def test_api_grant_user_group_permission_exception_when_adding_to_repo_group(self):
 
        perm = 'group.read'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission_to_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user group: `%s` in repo group: `%s`' % (
 
            TEST_USER_GROUP, TEST_REPO_GROUP
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @base.parametrize('name,apply_to_children', [
 
        ('none', 'none'),
 
        ('all', 'all'),
 
        ('repos', 'repos'),
 
        ('groups', 'groups'),
 
    ])
 
    def test_api_revoke_user_group_permission_from_repo_group(self, name, apply_to_children):
 
        RepoGroupModel().grant_user_group_permission(repo_group=TEST_REPO_GROUP,
 
                                                     group_name=TEST_USER_GROUP,
 
                                                     perm='group.read',)
 
@@ -2159,49 +2156,49 @@ class _BaseTestApi(object):
 
        if grant_admin:
 
            RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
 
                                                   self.TEST_USER_LOGIN,
 
                                                   'group.admin')
 
            Session().commit()
 

	
 
        id_, params = _build_data(self.apikey_regular,
 
                                  'revoke_user_group_permission_from_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  apply_to_children=apply_to_children,)
 
        response = api_call(self, params)
 
        if access_ok:
 
            expected = {
 
                'msg': 'Revoked perm (recursive:%s) for user group: `%s` in repo group: `%s`' % (
 
                    apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
 
                ),
 
                'success': True
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        else:
 
            expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
 
            self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoGroupModel, 'revoke_user_group_permission', crash)
 
    @mock.patch.object(RepoGroupModel, 'revoke_user_group_permission', raise_exception)
 
    def test_api_revoke_user_group_permission_from_repo_group_exception_when_adding(self):
 
        id_, params = _build_data(self.apikey, 'revoke_user_group_permission_from_repo_group',
 
                                  repogroupid=TEST_REPO_GROUP,
 
                                  usergroupid=TEST_USER_GROUP,)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user group: `%s` in repo group: `%s`' % (
 
            TEST_USER_GROUP, TEST_REPO_GROUP
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_get_gist(self):
 
        gist = fixture.create_gist()
 
        gist_id = gist.gist_access_id
 
        gist_created_on = gist.created_on
 
        id_, params = _build_data(self.apikey, 'get_gist',
 
                                  gistid=gist_id, )
 
        response = api_call(self, params)
 

	
 
        expected = {
 
            'access_id': gist_id,
 
            'created_on': gist_created_on,
 
            'description': 'new-gist',
 
            'expires': -1.0,
 
@@ -2280,81 +2277,81 @@ class _BaseTestApi(object):
 
        expected = 'userid is not the same as your user'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_create_gist(self):
 
        id_, params = _build_data(self.apikey_regular, 'create_gist',
 
                                  lifetime=10,
 
                                  description='foobar-gist',
 
                                  gist_type='public',
 
                                  files={'foobar': {'content': 'foo'}})
 
        response = api_call(self, params)
 
        expected = {
 
            'gist': {
 
                'access_id': response.json['result']['gist']['access_id'],
 
                'created_on': response.json['result']['gist']['created_on'],
 
                'description': 'foobar-gist',
 
                'expires': response.json['result']['gist']['expires'],
 
                'gist_id': response.json['result']['gist']['gist_id'],
 
                'type': 'public',
 
                'url': response.json['result']['gist']['url']
 
            },
 
            'msg': 'created new gist'
 
        }
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(GistModel, 'create', crash)
 
    @mock.patch.object(GistModel, 'create', raise_exception)
 
    def test_api_create_gist_exception_occurred(self):
 
        id_, params = _build_data(self.apikey_regular, 'create_gist',
 
                                  files={})
 
        response = api_call(self, params)
 
        expected = 'failed to create gist'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_delete_gist(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 = {'gist': None, 'msg': 'deleted gist ID:%s' % gist_id}
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_delete_gist_regular_user(self):
 
        gist_id = fixture.create_gist(owner=self.TEST_USER_LOGIN).gist_access_id
 
        id_, params = _build_data(self.apikey_regular, 'delete_gist',
 
                                  gistid=gist_id)
 
        response = api_call(self, params)
 
        expected = {'gist': None, 'msg': 'deleted gist ID:%s' % gist_id}
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_delete_gist_regular_user_no_permission(self):
 
        gist_id = fixture.create_gist().gist_access_id
 
        id_, params = _build_data(self.apikey_regular, 'delete_gist',
 
                                  gistid=gist_id)
 
        response = api_call(self, params)
 
        expected = 'gist `%s` does not exist' % (gist_id,)
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(GistModel, 'delete', crash)
 
    @mock.patch.object(GistModel, 'delete', raise_exception)
 
    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)
 

	
 
    def test_api_get_changesets(self):
kallithea/tests/fixture.py
Show inline comments
 
@@ -29,50 +29,50 @@ from tg.util.webtest import test_context
 
from kallithea.lib import helpers
 
from kallithea.lib.auth import AuthUser
 
from kallithea.lib.db_manage import DbManage
 
from kallithea.lib.vcs.backends.base import EmptyChangeset
 
from kallithea.model.changeset_status import ChangesetStatusModel
 
from kallithea.model.comment import ChangesetCommentsModel
 
from kallithea.model.db import ChangesetStatus, Gist, RepoGroup, Repository, User, UserGroup
 
from kallithea.model.gist import GistModel
 
from kallithea.model.meta import Session
 
from kallithea.model.pull_request import CreatePullRequestAction  # , CreatePullRequestIterationAction, PullRequestModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.scm import ScmModel
 
from kallithea.model.user import UserModel
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.tests.base import (GIT_REPO, HG_REPO, IP_ADDR, TEST_USER_ADMIN_EMAIL, TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN, TESTS_TMP_PATH,
 
                                  invalidate_all_caches)
 

	
 

	
 
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 !')
 
def raise_exception(*args, **kwargs):
 
    raise Exception('raise_exception raised exception')
 

	
 

	
 
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.
 

	
 
        Usage:
 

	
 
        fixture = Fixture()
 
        with fixture.anon_access(False):
 
            stuff
 
        """
 

	
 
        class context(object):
 
            def __enter__(self):
 
                anon = User.get_default_user()
 
                self._before = anon.active
 
                anon.active = status
kallithea/tests/functional/test_admin_repos.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
import os
 
import urllib.parse
 

	
 
import mock
 
import pytest
 

	
 
from kallithea.lib import vcs
 
from kallithea.model import db
 
from kallithea.model.db import Permission, Repository, Ui, User, UserRepoToPerm
 
from kallithea.model.meta import Session
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.user import UserModel
 
from kallithea.tests import base
 
from kallithea.tests.fixture import Fixture, error_function
 
from kallithea.tests.fixture import Fixture, raise_exception
 

	
 

	
 
fixture = Fixture()
 

	
 

	
 
def _get_permission_for_user(user, repo):
 
    perm = UserRepoToPerm.query() \
 
                .filter(UserRepoToPerm.repository ==
 
                        Repository.get_by_repo_name(repo)) \
 
                .filter(UserRepoToPerm.user == User.get_by_username(user)) \
 
                .all()
 
    return perm
 

	
 

	
 
class _BaseTestCase(base.TestController):
 
    """
 
    Write all tests here
 
    """
 
    REPO = None
 
    REPO_TYPE = None
 
    NEW_REPO = None
 
    OTHER_TYPE_REPO = None
 
    OTHER_TYPE = None
 

	
 
@@ -571,49 +571,49 @@ class _BaseTestCase(base.TestController)
 
        # disable on regular user
 
        user_model.revoke_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
 
        user_model.grant_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.create.none')
 
        user_model.revoke_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
 
        user_model.grant_perm(base.TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
 
        Session().commit()
 

	
 

	
 
        user = User.get(usr['user_id'])
 

	
 
        repo_name = self.NEW_REPO + 'no_perms'
 
        description = 'description for newly created repo'
 
        response = self.app.post(base.url('repos'),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_name=repo_name,
 
                                                repo_type=self.REPO_TYPE,
 
                                                repo_description=description,
 
                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
 

	
 
        response.mustcontain('<span class="error-message">Invalid value</span>')
 

	
 
        RepoModel().delete(repo_name)
 
        Session().commit()
 

	
 
    @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
 
    @mock.patch.object(RepoModel, '_create_filesystem_repo', raise_exception)
 
    def test_create_repo_when_filesystem_op_fails(self):
 
        self.log_user()
 
        repo_name = self.NEW_REPO
 
        description = 'description for newly created repo'
 

	
 
        response = self.app.post(base.url('repos'),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_name=repo_name,
 
                                                repo_type=self.REPO_TYPE,
 
                                                repo_description=description,
 
                                                _session_csrf_secret_token=self.session_csrf_secret_token()))
 

	
 
        self.checkSessionFlash(response,
 
                               'Error creating repository %s' % repo_name)
 
        # repo must not be in db
 
        repo = Repository.get_by_repo_name(repo_name)
 
        assert repo is None
 

	
 
        # repo must not be in filesystem !
 
        assert not os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name))
 

	
 

	
 
class TestAdminReposControllerGIT(_BaseTestCase):
 
    REPO = base.GIT_REPO
0 comments (0 inline, 0 general)