kallithea Changeset - e2d7830a0438

Changeset - e2d7830a0438

Parent rev.

Child rev.

[Not reviewed]

default

0 3 0

Mads Kiilerich - 3 years ago 2023-03-05 12:02:55
mads@kiilerich.com

ruff: Do not assign a `lambda` expression, use a `def`

3 files changed with 9 insertions and 9 deletions:

kallithea/tests/api/api_base.py

kallithea/tests/other/test_libs.py

kallithea/tests/vcs/test_git.py

0 comments (0 inline, 0 general)

kallithea/tests/api/api_base.py

➞

Show inline comments

 # -*- 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/>.
 """
 Tests for the JSON-RPC web api.
 """
 import os
 import random
 import re
 import string
 from typing import Sized
 import mock
 import pytest
 from webtest import TestApp
 from kallithea.lib import ext_json
 from kallithea.lib.auth import AuthUser
 from kallithea.lib.utils2 import ascii_bytes
 from kallithea.model import db, meta
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.gist import GistModel
 from kallithea.model.pull_request import 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 import base
 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 jsonify(obj):
     return ext_json.loads(ext_json.dumps(obj))
 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)
     meta.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)
     meta.Session().commit()
     return gr
 class _BaseTestApi(object):
     app: TestApp  # assigned by app_fixture in subclass TestController mixin
     # assigned in subclass:
     REPO: str
     REPO_TYPE: str
     TEST_REVISION: str
     TEST_PR_SRC: str
     TEST_PR_DST: str
     TEST_PR_REVISIONS: Sized
     @classmethod
     def setup_class(cls):
         cls.usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         cls.apikey = cls.usr.api_key
         cls.test_user = UserModel().create_or_update(
             username='test-api',
             password='test',
             email='test@example.com',
             firstname='first',
             lastname='last'
+        )
         meta.Session().commit()
         cls.TEST_USER_LOGIN = cls.test_user.username
         cls.apikey_regular = cls.test_user.api_key
     @classmethod
     def teardown_class(cls):
         pass
     def setup_method(self, method):
         make_user_group()
         make_repo_group()
     def teardown_method(self, method):
         fixture.destroy_user_group(TEST_USER_GROUP)
         fixture.destroy_gists()
         fixture.destroy_repo_group(TEST_REPO_GROUP)
     def _compare_ok(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': None,
             'result': expected
         })
         given = ext_json.loads(given)
         assert expected == given, (expected, given)
     def _compare_error(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': expected,
             'result': None
         })
         given = ext_json.loads(given)
         assert expected == given, (expected, given)
     def test_api_wrong_key(self):
         id_, params = _build_data('trololo', 'get_user')
         response = api_call(self, params)
         expected = 'Invalid API key'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param_args_null(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param_args_bad(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_args_is_null(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
         assert response.status == '200 OK'
     def test_api_args_is_bad(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
         assert response.status == '200 OK'
     def test_api_args_different_args(self):
         expected = {
             'ascii_letters': string.ascii_letters,
             'ws': string.whitespace,
             'printables': string.printable
+        }
         id_, params = _build_data(self.apikey, 'test', args=expected)
         response = api_call(self, params)
         assert response.status == '200 OK'
         self._compare_ok(id_, expected, response.body)
     def test_api_get_users(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         response = api_call(self, params)
         ret_all = []
         _users = db.User.query().filter_by(is_default_user=False) \
             .order_by(db.User.username).all()
         for usr in _users:
             ret = usr.get_api_data()
             ret_all.append(jsonify(ret))
         expected = ret_all
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid=base.TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_that_does_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid='trololo')
         response = api_call(self, params)
         expected = "user `%s` does not exist" % 'trololo'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid(self):
         id_, params = _build_data(self.apikey, 'get_user')
         response = api_call(self, params)
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user')
         response = api_call(self, params)
         usr = db.User.get_by_username(self.TEST_USER_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_with_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user',
                                   userid=self.TEST_USER_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
     def test_api_pull_remote(self):
         # Note: pulling from local repos is a mis-feature - it will bypass access control
         # ... but ok, if the path already has been set in the database
         repo_name = 'test_pull'
         r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         # hack around that clone_uri can't be set to to a local path
         # (as shown by test_api_create_repo_clone_uri_local)
         r.clone_uri = os.path.join(db.Ui.get_by_key('paths', '/').ui_value, self.REPO)
         meta.Session().commit()
         pre_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in db.Repository.query().filter(db.Repository.repo_name == repo_name)]
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=repo_name,)
         response = api_call(self, params)
         expected = {'msg': 'Pulled from `%s`' % repo_name,
                     'repository': repo_name}
         self._compare_ok(id_, expected, given=response.body)
         post_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in db.Repository.query().filter(db.Repository.repo_name == repo_name)]
         fixture.destroy_repo(repo_name)
         assert pre_cached_tip != post_cached_tip
     def test_api_pull_fork(self):
         fork_name = 'fork'
         fixture.create_fork(self.REPO, fork_name)
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=fork_name,)
         response = api_call(self, params)
         expected = {'msg': 'Pulled from `%s`' % fork_name,
                     'repository': fork_name}
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_pull_error_no_remote_no_fork(self):
         # should fail because no clone_uri is set
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=self.REPO, )
         response = api_call(self, params)
         expected = 'Unable to pull changes from `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     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(db.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', 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)
         expected = "email `%s` already exist" % base.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 + "@example.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         usr = db.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)
     def test_api_create_user_without_password(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@example.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email)
         response = api_call(self, params)
         usr = db.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)
     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 = db.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', 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')
         meta.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', 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')
         meta.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),
         ('extern_name', 'test'),
         ('extern_name', None),
         ('active', False),
         ('active', True),
         ('password', 'newpass'),
     ])
     def test_api_update_user(self, name, expected):
         usr = db.User.get_by_username(self.TEST_USER_LOGIN)
         kw = {name: expected,
               'userid': usr.user_id}
         id_, params = _build_data(self.apikey, 'update_user', **kw)
         response = api_call(self, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, self.TEST_USER_LOGIN),
             'user': jsonify(db.User \
                 .get_by_username(self.TEST_USER_LOGIN) \
                 .get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_no_changed_params(self):
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=base.TEST_USER_ADMIN_LOGIN)
         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_by_user_id(self):
         usr = db.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 = db.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', raise_exception)
     def test_api_update_user_when_exception_happens(self):
         usr = db.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')
         meta.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']
         repo = RepoModel().get_by_repo_name(self.REPO)
         ret = repo.get_api_data()
         members = []
         followers = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user = user.user
             user_data = {'name': user.username, 'type': "user",
                          'permission': perm}
             members.append(user_data)
         for user_group in repo.users_group_to_perm:
             perm = user_group.permission.permission_name
             user_group = user_group.users_group
             user_group_data = {'name': user_group.users_group_name,
                                'type': "user_group", 'permission': perm}
             members.append(user_group_data)
         for user in repo.followers:
             followers.append(user.user.get_api_data())
         ret['members'] = members
         ret['followers'] = followers
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_user_group(new_group)
         id_, params = _build_data(self.apikey, 'get_repo', repoid=self.REPO,
                                   with_revision_names=True,
                                   with_pullrequests=True)
         response = api_call(self, params)
         assert "v0.2.0" in response.json['result']['tags']
         assert 'pull_requests' in response.json['result']
     @base.parametrize('grant_perm', [
         ('repository.admin'),
         ('repository.write'),
         ('repository.read'),
     ])
     def test_api_get_repo_by_non_admin(self, grant_perm):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         meta.Session().commit()
         id_, params = _build_data(self.apikey_regular, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(self.REPO)
         assert len(repo.repo_to_perm) >= 2  # make sure we actually are testing something - probably the default 2 permissions, possibly more
         expected = repo.get_api_data()
         members = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user_obj = user.user
             user_data = {'name': user_obj.username, 'type': "user",
                          'permission': perm}
             members.append(user_data)
         for user_group in repo.users_group_to_perm:
             perm = user_group.permission.permission_name
             user_group_obj = user_group.users_group
             user_group_data = {'name': user_group_obj.users_group_name,
                                'type': "user_group", 'permission': perm}
             members.append(user_group_data)
         expected['members'] = members
         followers = []
         for user in repo.followers:
             followers.append(user.user.get_api_data())
         expected['followers'] = followers
         try:
             self._compare_ok(id_, expected, given=response.body)
         finally:
             RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
     def test_api_get_repo_by_non_admin_no_permission_to_repo(self):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=db.User.DEFAULT_USER_NAME,
                                           perm='repository.none')
         try:
             RepoModel().grant_user_permission(repo=self.REPO,
                                               user=self.TEST_USER_LOGIN,
                                               perm='repository.none')
             id_, params = _build_data(self.apikey_regular, 'get_repo',
                                       repoid=self.REPO)
             response = api_call(self, params)
             expected = 'repository `%s` does not exist' % (self.REPO)
             self._compare_error(id_, expected, given=response.body)
         finally:
             RepoModel().grant_user_permission(repo=self.REPO,
                                               user=db.User.DEFAULT_USER_NAME,
                                               perm='repository.read')
     def test_api_get_repo_that_doesn_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid='no-such-repo')
         response = api_call(self, params)
         ret = 'repository `%s` does not exist' % 'no-such-repo'
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repos(self):
         id_, params = _build_data(self.apikey, 'get_repos')
         response = api_call(self, params)
         expected = jsonify([
             repo.get_api_data()
             for repo in db.Repository.query()
         ])
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_repos_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_repos')
         response = api_call(self, params)
         expected = jsonify([
             repo.get_api_data()
             for repo in AuthUser(dbuser=db.User.get_by_username(self.TEST_USER_LOGIN)).get_all_user_repos()
         ])
         self._compare_ok(id_, expected, given=response.body)
     @base.parametrize('name,ret_type', [
         ('all', 'all'),
         ('dirs', 'dirs'),
         ('files', 'files'),
     ])
     def test_api_get_repo_nodes(self, name, ret_type):
         rev = 'tip'
         path = '/'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = response.json['result']
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_revisions(self):
         rev = 'i-dont-exist'
         path = '/'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path, )
         response = api_call(self, params)
         expected = 'failed to get repo: `%s` nodes' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_path(self):
         rev = 'tip'
         path = '/idontexits'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path, )
         response = api_call(self, params)
         expected = 'failed to get repo: `%s` nodes' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_ret_type(self):
         rev = 'tip'
         path = '/'
         ret_type = 'error'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         expected = ('ret_type must be one of %s'
                     % (','.join(sorted(['files', 'dirs', 'all']))))
         self._compare_error(id_, expected, given=response.body)
     @base.parametrize('name,ret_type,grant_perm', [
         ('all', 'all', 'repository.write'),
         ('dirs', 'dirs', 'repository.admin'),
         ('files', 'files', 'repository.read'),
     ])
     def test_api_get_repo_nodes_by_regular_user(self, name, ret_type, grant_perm):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         meta.Session().commit()
         rev = 'tip'
         path = '/'
         id_, params = _build_data(self.apikey_regular, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = response.json['result']
         try:
             self._compare_ok(id_, expected, given=response.body)
         finally:
             RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
     def test_api_create_repo(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)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     @base.parametrize('repo_name', [
         '',
         '.',
         '..',
         ':',
         '/',
         '<test>',
     ])
     def test_api_create_repo_bad_names(self, repo_name):
         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)
         if repo_name == '/':
             expected = "repo group `` not found"
             self._compare_error(id_, expected, given=response.body)
         else:
             expected = "failed to create repository `%s`" % repo_name
             self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_clone_uri_local(self):
         # cloning from local repos was a mis-feature - it would bypass access control
         # TODO: introduce other test coverage of actual remote cloning
         clone_uri = os.path.join(base.TESTS_TMP_PATH, self.REPO)
         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,
                                   clone_uri=clone_uri,
+        )
         response = api_call(self, params)
         expected = "failed to create repository `%s`" % repo_name
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_and_repo_group(self):
         repo_group_name = 'my_gr'
         repo_name = '%s/api-repo' % repo_group_name
         # repo creation can no longer also create repo group
         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' % repo_group_name
         self._compare_error(id_, expected, given=response.body)
         assert RepoModel().get_by_repo_name(repo_name) is None
         # create group before creating repo
         rg = fixture.create_repo_group(repo_group_name)
         meta.Session().commit()
         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 = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
+        }
         self._compare_ok(id_, expected, given=response.body)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         fixture.destroy_repo(repo_name)
         fixture.destroy_repo_group(repo_group_name)
     def test_api_create_repo_in_repo_group_without_permission(self):
         repo_group_basename = 'api-repo-repo'
         repo_group_name = '%s/%s' % (TEST_REPO_GROUP, repo_group_basename)
         repo_name = '%s/api-repo' % repo_group_name
         top_group = db.RepoGroup.get_by_group_name(TEST_REPO_GROUP)
         assert top_group
         rg = fixture.create_repo_group(repo_group_basename, parent_group_id=top_group)
         meta.Session().commit()
         RepoGroupModel().grant_user_permission(repo_group_name,
                                                self.TEST_USER_LOGIN,
                                                'group.none')
         meta.Session().commit()
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         # API access control match Web access control:
         expected = 'no permission to create repo in test_repo_group/api-repo-repo'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
         fixture.destroy_repo_group(repo_group_name)
     def test_api_create_repo_unknown_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=owner,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_repo_dont_specify_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_by_non_admin(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_by_non_admin_specify_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
                                   owner=owner)
         response = api_call(self, params)
         expected = 'Only Kallithea admin can specify `owner` param'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     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', 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'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         if changing_attr == 'repo_group':
             fixture.create_repo_group(updates['group'])
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         if changing_attr == 'name':
             repo_name = updates['name']
         if changing_attr == 'repo_group':
             repo_name = '/'.join([updates['group'], repo_name])
         try:
             if changing_attr == 'clone_uri' and updates['clone_uri']:
                 expected = 'failed to update repo `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
             else:
                 expected = {
                     'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
                     'repository': repo.get_api_data()
+                }
                 self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             if changing_attr == 'repo_group':
                 fixture.destroy_repo_group(updates['group'])
     @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_group_repo(self, changing_attr, updates):
         group_name = 'lololo'
         fixture.create_repo_group(group_name)
         repo_name = '%s/api_update_me' % group_name
         repo = fixture.create_repo(repo_name, repo_group=group_name, repo_type=self.REPO_TYPE)
         if changing_attr == 'repo_group':
             fixture.create_repo_group(updates['group'])
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         if changing_attr == 'name':
             repo_name = '%s/%s' % (group_name, updates['name'])
         if changing_attr == 'repo_group':
             repo_name = '/'.join([updates['group'], repo_name.rsplit('/', 1)[-1]])
         try:
             if changing_attr == 'clone_uri' and updates['clone_uri']:
                 expected = 'failed to update repo `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
             else:
                 expected = {
                     'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
                     'repository': repo.get_api_data()
+                }
                 self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             if changing_attr == 'repo_group':
                 fixture.destroy_repo_group(updates['group'])
         fixture.destroy_repo_group(group_name)
     def test_api_update_repo_repo_group_does_not_exist(self):
         repo_name = 'admin_owned'
         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', 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_top_level_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)
         response = api_call(self, params)
         try:
             expected = 'no permission to create (or move) top level repositories'
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             fixture.destroy_repo(new_repo_name)
     def test_api_update_repo_regular_user_change_repo_name_allowed(self):
         repo_name = 'admin_owned'
         new_repo_name = 'new_repo_name'
         repo = 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.none')
         UserModel().grant_perm('default', 'hg.create.repository')
         updates = {'name': new_repo_name}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = {
                 'msg': 'updated repo ID:%s %s' % (repo.repo_id, new_repo_name),
                 'repository': repo.get_api_data()
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             fixture.destroy_repo(new_repo_name)
     def test_api_update_repo_regular_user_change_owner(self):
         repo_name = 'admin_owned'
         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')
         updates = {'owner': base.TEST_USER_ADMIN_LOGIN}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = 'Only Kallithea admin can specify `owner` param'
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_delete_repo(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         id_, params = _build_data(self.apikey, 'delete_repo',
                                   repoid=repo_name, )
         response = api_call(self, params)
         ret = {
             'msg': 'Deleted repository `%s`' % repo_name,
             '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(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         id_, params = _build_data(self.apikey_regular, 'delete_repo',
                                   repoid=repo_name, )
         response = api_call(self, params)
         ret = {
             'msg': 'Deleted repository `%s`' % repo_name,
             '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', 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,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     @base.parametrize('fork_name', [
         'api-repo-fork',
         '%s/api-repo-fork' % TEST_REPO_GROUP,
     ])
     def test_api_fork_repo_non_admin(self, fork_name):
         RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                self.TEST_USER_LOGIN,
                                                'group.write')
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
+        )
         response = api_call(self, params)
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_fork_repo_non_admin_specify_owner(self):
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
+        )
         response = api_call(self, params)
         expected = 'Only Kallithea admin can specify `owner` param'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_fork_repo_non_admin_no_permission_to_fork(self):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=db.User.DEFAULT_USER_NAME,
                                           perm='repository.none')
         fork_name = 'api-repo-fork'
         try:
             id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                       repoid=self.REPO,
                                       fork_name=fork_name,
+            )
             response = api_call(self, params)
             expected = 'repository `%s` does not exist' % (self.REPO)
             self._compare_error(id_, expected, given=response.body)
         finally:
             RepoModel().grant_user_permission(repo=self.REPO,
                                               user=db.User.DEFAULT_USER_NAME,
                                               perm='repository.read')
             fixture.destroy_repo(fork_name)
     @base.parametrize('name,perm', [
         ('read', 'repository.read'),
         ('write', 'repository.write'),
         ('admin', 'repository.admin'),
     ])
     def test_api_fork_repo_non_admin_no_create_repo_permission(self, name, perm):
         fork_name = 'api-repo-fork'
         # regardless of base repository permission, forking is disallowed
         # when repository creation is disabled
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=perm)
         UserModel().revoke_perm('default', 'hg.create.repository')
         UserModel().grant_perm('default', 'hg.create.none')
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
+        )
         response = api_call(self, params)
         expected = 'no permission to create top level repo'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_fork_repo_unknown_owner(self):
         fork_name = 'api-repo-fork'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=owner,
+        )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
     def test_api_fork_repo_fork_exists(self):
         fork_name = 'api-repo-fork'
         fixture.create_fork(self.REPO, fork_name)
         try:
             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 = "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', 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())
         ret = user_group.get_api_data()
         ret['members'] = members
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_groups(self):
         gr_name = 'test_user_group2'
         make_user_group(gr_name)
         try:
             id_, params = _build_data(self.apikey, 'get_user_groups', )
             response = api_call(self, params)
             expected = []
             for gr_name in [TEST_USER_GROUP, 'test_user_group2']:
                 user_group = UserGroupModel().get_group(gr_name)
                 ret = user_group.get_api_data()
                 expected.append(ret)
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user_group(gr_name)
     def test_api_create_user_group(self):
         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', 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', 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', 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)
         meta.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', 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)
         meta.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 = {
                 'user_group': None,
                 'msg': 'deleted user group ID:%s %s' % (gr_id, gr_name)
+            }
             self._compare_ok(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_that_is_assigned(self):
         gr_name = 'test_group'
         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', 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', 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', 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', 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')
         meta.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', 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'),
         ('read', 'group.read', 'repos'),
         ('write', 'group.write', 'repos'),
         ('admin', 'group.admin', 'repos'),
         ('none', 'group.none', 'groups'),
         ('read', 'group.read', 'groups'),
         ('write', 'group.write', 'groups'),
         ('admin', 'group.admin', 'groups'),
     ])
     def test_api_grant_user_permission_to_repo_group(self, name, perm, apply_to_children):
         id_, params = _build_data(self.apikey,
                                   'grant_user_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   userid=base.TEST_USER_ADMIN_LOGIN,
                                   perm=perm, apply_to_children=apply_to_children)
         response = api_call(self, params)
         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)
     @base.parametrize('name,perm,apply_to_children,grant_admin,access_ok', [
         ('none_fails', 'group.none', 'none', False, False),
         ('read_fails', 'group.read', 'none', False, False),
         ('write_fails', 'group.write', 'none', False, False),
         ('admin_fails', 'group.admin', 'none', False, False),
         # with granted perms
         ('none_ok', 'group.none', 'none', True, True),
         ('read_ok', 'group.read', 'none', True, True),
         ('write_ok', 'group.write', 'none', True, True),
         ('admin_ok', 'group.admin', 'none', True, True),
     ])
     def test_api_grant_user_permission_to_repo_group_by_regular_user(
             self, name, perm, apply_to_children, grant_admin, access_ok):
         if grant_admin:
             RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                    self.TEST_USER_LOGIN,
                                                    'group.admin')
             meta.Session().commit()
         id_, params = _build_data(self.apikey_regular,
                                   'grant_user_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   userid=base.TEST_USER_ADMIN_LOGIN,
                                   perm=perm, apply_to_children=apply_to_children)
         response = api_call(self, params)
         if access_ok:
             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', 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',)
         meta.Session().commit()
         id_, params = _build_data(self.apikey,
                                   '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)
         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)
     @base.parametrize('name,apply_to_children,grant_admin,access_ok', [
         ('none', 'none', False, False),
         ('all', 'all', False, False),
         ('repos', 'repos', False, False),
         ('groups', 'groups', False, False),
         # after granting admin rights
         ('none', 'none', False, False),
         ('all', 'all', False, False),
         ('repos', 'repos', False, False),
         ('groups', 'groups', False, False),
     ])
     def test_api_revoke_user_permission_from_repo_group_by_regular_user(
             self, name, apply_to_children, grant_admin, access_ok):
         RepoGroupModel().grant_user_permission(repo_group=TEST_REPO_GROUP,
                                                user=base.TEST_USER_ADMIN_LOGIN,
                                                perm='group.read',)
         meta.Session().commit()
         if grant_admin:
             RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                    self.TEST_USER_LOGIN,
                                                    'group.admin')
             meta.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', 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'),
         ('read', 'group.read', 'repos'),
         ('write', 'group.write', 'repos'),
         ('admin', 'group.admin', 'repos'),
         ('none', 'group.none', 'groups'),
         ('read', 'group.read', 'groups'),
         ('write', 'group.write', 'groups'),
         ('admin', 'group.admin', 'groups'),
     ])
     def test_api_grant_user_group_permission_to_repo_group(self, name, perm, apply_to_children):
         id_, params = _build_data(self.apikey,
                                   'grant_user_group_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   usergroupid=TEST_USER_GROUP,
                                   perm=perm,
                                   apply_to_children=apply_to_children,)
         response = api_call(self, params)
         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)
     @base.parametrize('name,perm,apply_to_children,grant_admin,access_ok', [
         ('none_fails', 'group.none', 'none', False, False),
         ('read_fails', 'group.read', 'none', False, False),
         ('write_fails', 'group.write', 'none', False, False),
         ('admin_fails', 'group.admin', 'none', False, False),
         # with granted perms
         ('none_ok', 'group.none', 'none', True, True),
         ('read_ok', 'group.read', 'none', True, True),
         ('write_ok', 'group.write', 'none', True, True),
         ('admin_ok', 'group.admin', 'none', True, True),
     ])
     def test_api_grant_user_group_permission_to_repo_group_by_regular_user(
             self, name, perm, apply_to_children, grant_admin, access_ok):
         if grant_admin:
             RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                    self.TEST_USER_LOGIN,
                                                    'group.admin')
             meta.Session().commit()
         id_, params = _build_data(self.apikey_regular,
                                   'grant_user_group_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   usergroupid=TEST_USER_GROUP,
                                   perm=perm,
                                   apply_to_children=apply_to_children,)
         response = api_call(self, params)
         if access_ok:
             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', 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',)
         meta.Session().commit()
         id_, params = _build_data(self.apikey,
                                   '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)
         expected = {
             'msg': 'Revoked perm (recursive:%s) for user group: `%s` in repo group: `%s`' % (
                 apply_to_children, TEST_USER_GROUP, TEST_REPO_GROUP
             ),
             'success': True
+        }
         self._compare_ok(id_, expected, given=response.body)
     @base.parametrize('name,apply_to_children,grant_admin,access_ok', [
         ('none', 'none', False, False),
         ('all', 'all', False, False),
         ('repos', 'repos', False, False),
         ('groups', 'groups', False, False),
         # after granting admin rights
         ('none', 'none', False, False),
         ('all', 'all', False, False),
         ('repos', 'repos', False, False),
         ('groups', 'groups', False, False),
     ])
     def test_api_revoke_user_group_permission_from_repo_group_by_regular_user(
             self, name, apply_to_children, grant_admin, access_ok):
         RepoGroupModel().grant_user_permission(repo_group=TEST_REPO_GROUP,
                                                user=base.TEST_USER_ADMIN_LOGIN,
                                                perm='group.read',)
         meta.Session().commit()
         if grant_admin:
             RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                    self.TEST_USER_LOGIN,
                                                    'group.admin')
             meta.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', 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,
             'gist_id': int(gist_id),
             'type': 'public',
             'url': 'http://localhost:80/_admin/gists/%s' % gist_id
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_gist_that_does_not_exist(self):
         id_, params = _build_data(self.apikey_regular, 'get_gist',
                                   gistid='12345', )
         response = api_call(self, params)
         expected = 'gist `%s` does not exist' % ('12345',)
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_gist_private_gist_without_permission(self):
         gist = fixture.create_gist()
         gist_id = gist.gist_access_id
         gist_created_on = gist.created_on
         id_, params = _build_data(self.apikey_regular, 'get_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)
     def test_api_get_gists(self):
         fixture.create_gist()
         fixture.create_gist()
         id_, params = _build_data(self.apikey, 'get_gists')
         response = api_call(self, params)
         expected = response.json
         assert len(response.json['result']) == 2
         #self._compare_ok(id_, expected, given=response.body)
     def test_api_get_gists_regular_user(self):
         # by admin
         fixture.create_gist()
         fixture.create_gist()
         # by reg user
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
         id_, params = _build_data(self.apikey_regular, 'get_gists')
         response = api_call(self, params)
         expected = response.json
         assert len(response.json['result']) == 3
         #self._compare_ok(id_, expected, given=response.body)
     def test_api_get_gists_only_for_regular_user(self):
         # by admin
         fixture.create_gist()
         fixture.create_gist()
         # by reg user
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
         id_, params = _build_data(self.apikey, 'get_gists',
                                   userid=self.TEST_USER_LOGIN)
         response = api_call(self, params)
         expected = response.json
         assert len(response.json['result']) == 3
         #self._compare_ok(id_, expected, given=response.body)
     def test_api_get_gists_regular_user_with_different_userid(self):
         id_, params = _build_data(self.apikey_regular, 'get_gists',
                                   userid=base.TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         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', 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', 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 = db.Setting.get_server_info()
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_changesets(self):
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, start=0, end=2)
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert len(result) == 3
         assert 'message' in result[0]
         assert 'added' not in result[0]
     def test_api_get_changesets_with_max_revisions(self):
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, start_date="2011-02-24T00:00:00", max_revisions=10)
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert len(result) == 10
         assert 'message' in result[0]
         assert 'added' not in result[0]
     def test_api_get_changesets_with_branch(self):
         if self.REPO == 'vcs_test_hg':
             branch = 'stable'
         else:
             pytest.skip("skipping due to missing branches in git test repo")
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, branch_name=branch, start_date="2011-02-24T00:00:00")
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert len(result) == 5
         assert 'message' in result[0]
         assert 'added' not in result[0]
     def test_api_get_changesets_with_file_list(self):
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, start_date="2010-04-07T23:30:30", end_date="2010-04-08T00:31:14", with_file_list=True)
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert len(result) == 3
         assert 'message' in result[0]
         assert 'added' in result[0]
     def test_api_get_changeset(self):
         review = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")
         id_, params = _build_data(self.apikey, 'get_changeset',
                                   repoid=self.REPO, raw_id=self.TEST_REVISION)
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert result["raw_id"] == self.TEST_REVISION
         assert "reviews" not in result
     def test_api_get_changeset_with_reviews(self):
         reviewobjs = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")
         id_, params = _build_data(self.apikey, 'get_changeset',
                                   repoid=self.REPO, raw_id=self.TEST_REVISION,
                                   with_reviews=True)
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert result["raw_id"] == self.TEST_REVISION
         assert "reviews" in result
         assert len(result["reviews"]) == 1
         review = result["reviews"][0]
         expected = {
             'status': 'approved',
             'modified_at': reviewobjs[0].modified_at.replace(microsecond=0).isoformat(),
             'reviewer': 'test_admin',
+        }
         assert review == expected
     def test_api_get_changeset_that_does_not_exist(self):
         """ Fetch changeset status for non-existant changeset.
         revision id is the above git hash used in the test above with the
         last 3 nibbles replaced with 0xf.  Should not exist for git _or_ hg.
         """
         id_, params = _build_data(self.apikey, 'get_changeset',
                                   repoid=self.REPO, raw_id = '7ab37bc680b4aa72c34d07b230c866c28e9fcfff')
         response = api_call(self, params)
         expected = 'Changeset %s does not exist' % ('7ab37bc680b4aa72c34d07b230c866c28e9fcfff',)
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_changeset_without_permission(self):
         review = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")
         RepoModel().revoke_user_permission(repo=self.REPO, user=self.TEST_USER_LOGIN)
         RepoModel().revoke_user_permission(repo=self.REPO, user="default")
         id_, params = _build_data(self.apikey_regular, 'get_changeset',
                                   repoid=self.REPO, raw_id=self.TEST_REVISION)
         response = api_call(self, params)
         expected = 'Access denied to repo %s' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'get test')
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey,
             "method": 'get_pullrequest',
             "args": {"pullrequest_id": pull_request_id},
         }))
         response = api_call(self, params)
         pullrequest = db.PullRequest().get(pull_request_id)
         expected = {
             "status": "new",
             "pull_request_id": pull_request_id,
             "description": "No description",
             "url": "/%s/pull-request/%s/_/%s" % (self.REPO, pull_request_id, "stable"),
             "reviewers": [{"username": "test_regular"}],
             "org_repo_url": "http://localhost:80/%s" % self.REPO,
             "org_ref_parts": ["branch", "stable", self.TEST_PR_SRC],
             "other_ref_parts": ["branch", "default", self.TEST_PR_DST],
             "comments": [{"username": base.TEST_USER_ADMIN_LOGIN, "text": "",
                          "comment_id": pullrequest.comments[0].comment_id}],
             "owner": base.TEST_USER_ADMIN_LOGIN,
             "statuses": [{"status": "under_review", "reviewer": base.TEST_USER_ADMIN_LOGIN, "modified_at": "2000-01-01T00:00:00"} for i in range(0, len(self.TEST_PR_REVISIONS))],
             "title": "get test",
             "revisions": self.TEST_PR_REVISIONS,
             "created_on": "2000-01-01T00:00:00",
             "updated_on": "2000-01-01T00:00:00",
+        }
         self._compare_ok(random_id, expected,
                          given=re.sub(br"\d\d\d\d\-\d\d\-\d\dT\d\d\:\d\d\:\d\d",
                                       b"2000-01-01T00:00:00", response.body))
     def test_api_close_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'close test')
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "close_pr": True},
         }))
         response = api_call(self, params)
         self._compare_ok(random_id, True, given=response.body)
         pullrequest = db.PullRequest().get(pull_request_id)
         assert pullrequest.comments[-1].text == ''
         assert pullrequest.status == db.PullRequest.STATUS_CLOSED
         assert pullrequest.is_closed() == True
     def test_api_status_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, "status test")
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN).api_key,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "status": db.ChangesetStatus.STATUS_APPROVED},
         }))
         response = api_call(self, params)
         pullrequest = db.PullRequest().get(pull_request_id)
         self._compare_error(random_id, "No permission to change pull request status. User needs to be admin, owner or reviewer.", given=response.body)
         assert db.ChangesetStatus.STATUS_UNDER_REVIEW == ChangesetStatusModel().calculate_pull_request_result(pullrequest)[2]
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN).api_key,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "status": db.ChangesetStatus.STATUS_APPROVED},
         }))
         response = api_call(self, params)
         self._compare_ok(random_id, True, given=response.body)
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.ChangesetStatus.STATUS_APPROVED == ChangesetStatusModel().calculate_pull_request_result(pullrequest)[2]
     def test_api_comment_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, "comment test")
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "comment_msg": "Looks good to me"},
         }))
         response = api_call(self, params)
         self._compare_ok(random_id, True, given=response.body)
         pullrequest = db.PullRequest().get(pull_request_id)
         assert pullrequest.comments[-1].text == 'Looks good to me'
     def test_api_edit_reviewers_add_single(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         expected = { 'added': [base.TEST_USER_REGULAR2_LOGIN], 'already_present': [], 'removed': [] }
         self._compare_ok(random_id, expected, given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_add_nonexistent(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": 999},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "user `999` does not exist", given=response.body)
     def test_api_edit_reviewers_add_multiple(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {
                 "pull_request_id": pull_request_id,
                 "add": [ self.TEST_USER_LOGIN, base.TEST_USER_REGULAR2_LOGIN ]
             },
         }))
         response = api_call(self, params)
         # list order depends on python sorting hash, which is randomized
         assert set(ext_json.loads(response.body)['result']['added']) == set([base.TEST_USER_REGULAR2_LOGIN, self.TEST_USER_LOGIN])
         assert set(ext_json.loads(response.body)['result']['already_present']) == set()
         assert set(ext_json.loads(response.body)['result']['removed']) == set()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(self.TEST_USER_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_add_already_present(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {
                 "pull_request_id": pull_request_id,
                 "add": [ base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR2_LOGIN ]
             },
         }))
         response = api_call(self, params)
         expected = { 'added': [base.TEST_USER_REGULAR2_LOGIN],
                      'already_present': [base.TEST_USER_REGULAR_LOGIN],
                      'removed': [],
+                   }
         self._compare_ok(random_id, expected, given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_add_closed(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         PullRequestModel().close_pull_request(pull_request_id)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "Cannot edit reviewers of a closed pull request.", given=response.body)
     def test_api_edit_reviewers_add_not_owner(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "No permission to edit reviewers of this pull request. User needs to be admin or pull request owner.", given=response.body)
     def test_api_edit_reviewers_remove_single(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR_LOGIN},
         }))
         response = api_call(self, params)
         expected = { 'added': [],
                      'already_present': [],
                      'removed': [base.TEST_USER_REGULAR_LOGIN],
+                   }
         self._compare_ok(random_id, expected, given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_remove_nonexistent(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": 999},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "user `999` does not exist", given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_remove_nonpresent(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         # NOTE: no explicit indication that removed user was not even a reviewer
         expected = { 'added': [],
                      'already_present': [],
                      'removed': [base.TEST_USER_REGULAR2_LOGIN],
+                   }
         self._compare_ok(random_id, expected, given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_remove_multiple(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         prr = db.PullRequestReviewer(db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN), pullrequest)
         meta.Session().add(prr)
         meta.Session().commit()
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": [ base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR2_LOGIN ] },
         }))
         response = api_call(self, params)
         # list order depends on python sorting hash, which is randomized
         assert set(ext_json.loads(response.body)['result']['added']) == set()
         assert set(ext_json.loads(response.body)['result']['already_present']) == set()
         assert set(ext_json.loads(response.body)['result']['removed']) == set([base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR2_LOGIN])
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_remove_closed(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         PullRequestModel().close_pull_request(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR_LOGIN},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "Cannot edit reviewers of a closed pull request.", given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_remove_not_owner(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         pullrequest.owner = db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR_LOGIN},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "No permission to edit reviewers of this pull request. User needs to be admin or pull request owner.", given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_add_remove_single(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id,
                      "add": base.TEST_USER_REGULAR2_LOGIN,
                      "remove": base.TEST_USER_REGULAR_LOGIN
                     },
         }))
         response = api_call(self, params)
         expected = { 'added': [base.TEST_USER_REGULAR2_LOGIN],
                      'already_present': [],
                      'removed': [base.TEST_USER_REGULAR_LOGIN],
+                   }
         self._compare_ok(random_id, expected, given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_add_remove_multiple(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         prr = db.PullRequestReviewer(db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN), pullrequest)
         meta.Session().add(prr)
         meta.Session().commit()
         assert db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id,
                      "add": [ base.TEST_USER_REGULAR2_LOGIN ],
                      "remove": [ base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_ADMIN_LOGIN ],
                     },
         }))
         response = api_call(self, params)
         # list order depends on python sorting hash, which is randomized
         assert set(ext_json.loads(response.body)['result']['added']) == set([base.TEST_USER_REGULAR2_LOGIN])
         assert set(ext_json.loads(response.body)['result']['already_present']) == set()
         assert set(ext_json.loads(response.body)['result']['removed']) == set([base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_ADMIN_LOGIN])
         assert db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN) not in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_invalid_params(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         pullrequest.owner = db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "Invalid request. Neither 'add' nor 'remove' is specified.", given=response.body)
         assert ext_json.loads(response.body)['result'] is None

kallithea/tests/other/test_libs.py

➞

Show inline comments

 # -*- 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/>.
 """
 kallithea.tests.other.test_libs
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Package for testing various lib/helper functions in kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jun 9, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import datetime
 import hashlib
 import re
 import mock
 import routes
 from dateutil import relativedelta
 from tg import request
 from tg.util.webtest import test_context
 import kallithea.lib.helpers as h
 from kallithea.lib import webutils
 from kallithea.lib.utils2 import AttributeDict, get_clone_url, safe_bytes
 from kallithea.model import db
 from kallithea.tests import base
 proto = 'http'
 TEST_URLS = [
     ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
      '%s://127.0.0.1:8080' % proto),
     ('%s://example.com' % proto, ['%s://' % proto, 'example.com'],
      '%s://example.com' % proto),
     ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',
                                                 '8080'],
      '%s://example.com:8080' % proto),
+]
 proto = 'https'
 TEST_URLS += [
     ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
      '%s://127.0.0.1:8080' % proto),
     ('%s://example.com' % proto, ['%s://' % proto, 'example.com'],
      '%s://example.com' % proto),
     ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',
                                                 '8080'],
      '%s://example.com:8080' % proto),
+]
 class TestLibs(base.TestController):
     @base.parametrize('test_url,expected,expected_creds', TEST_URLS)
     def test_uri_filter(self, test_url, expected, expected_creds):
         from kallithea.lib.utils2 import uri_filter
         assert uri_filter(test_url) == expected
     @base.parametrize('test_url,expected,expected_creds', TEST_URLS)
     def test_credentials_filter(self, test_url, expected, expected_creds):
         from kallithea.lib.utils2 import credentials_filter
         assert credentials_filter(test_url) == expected_creds
     @base.parametrize('str_bool,expected', [
                            ('t', True),
                            ('true', True),
                            ('y', True),
                            ('yes', True),
                            ('on', True),
                            ('1', True),
                            ('Y', True),
                            ('yeS', True),
                            ('Y', True),
                            ('TRUE', True),
                            ('T', True),
                            ('False', False),
                            ('F', False),
                            ('FALSE', False),
                            ('0', False),
     ])
     def test_asbool(self, str_bool, expected):
         from kallithea.lib.utils2 import asbool
         assert asbool(str_bool) == expected
     def test_mention_extractor(self):
         sample = (
             "@first hi there @world here's my email username@example.com "
             "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "
             "@UPPER    @cAmEL @2one_more22 @john please see this http://org.pl "
             "@marian.user just do it @marco-polo and next extract @marco_polo "
             "user.dot  hej ! not-needed maril@example.com"
+        )
         expected = set([
             '2one_more22', 'first', 'lukaszb', 'one', 'one_more22', 'UPPER', 'cAmEL', 'john',
             'marian.user', 'marco-polo', 'marco_polo', 'world'])
         assert expected == set(webutils.extract_mentioned_usernames(sample))
     @base.parametrize('age_args,expected', [
         (dict(), 'just now'),
         (dict(seconds= -1), '1 second ago'),
         (dict(seconds= -60 * 2), '2 minutes ago'),
         (dict(hours= -1), '1 hour ago'),
         (dict(hours= -24), '1 day ago'),
         (dict(hours= -24 * 5), '5 days ago'),
         (dict(months= -1), '1 month ago'),
         (dict(months= -1, days= -2), '1 month and 2 days ago'),
         (dict(months= -1, days= -20), '1 month and 19 days ago'),
         (dict(years= -1, months= -1), '1 year and 1 month ago'),
         (dict(years= -1, months= -10), '1 year and 10 months ago'),
         (dict(years= -2, months= -4), '2 years and 4 months ago'),
         (dict(years= -2, months= -11), '2 years and 11 months ago'),
         (dict(years= -3, months= -2), '3 years and 2 months ago'),
     ])
     def test_age(self, age_args, expected):
         from kallithea.lib.webutils import age
         with test_context(self.app):
             n = datetime.datetime(year=2012, month=5, day=17)
             delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
             assert age(n + delt(**age_args), now=n) == expected
             assert age(n + relativedelta.relativedelta(**age_args), now=n) == expected
     @base.parametrize('age_args,expected', [
         (dict(), 'just now'),
         (dict(seconds= -1), '1 second ago'),
         (dict(seconds= -60 * 2), '2 minutes ago'),
         (dict(hours= -1), '1 hour ago'),
         (dict(hours= -24), '1 day ago'),
         (dict(hours= -24 * 5), '5 days ago'),
         (dict(months= -1), '1 month ago'),
         (dict(months= -1, days= -2), '1 month ago'),
         (dict(months= -1, days= -20), '1 month ago'),
         (dict(years= -1, months= -1), '13 months ago'),
         (dict(years= -1, months= -10), '22 months ago'),
         (dict(years= -2, months= -4), '2 years ago'),
         (dict(years= -2, months= -11), '3 years ago'),
         (dict(years= -3, months= -2), '3 years ago'),
         (dict(years= -4, months= -8), '5 years ago'),
     ])
     def test_age_short(self, age_args, expected):
         from kallithea.lib.webutils import age
         with test_context(self.app):
             n = datetime.datetime(year=2012, month=5, day=17)
             delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
             assert age(n + delt(**age_args), show_short_version=True, now=n) == expected
             assert age(n + relativedelta.relativedelta(**age_args), show_short_version=True, now=n) == expected
     @base.parametrize('age_args,expected', [
         (dict(), 'just now'),
         (dict(seconds=1), 'in 1 second'),
         (dict(seconds=60 * 2), 'in 2 minutes'),
         (dict(hours=1), 'in 1 hour'),
         (dict(hours=24), 'in 1 day'),
         (dict(hours=24 * 5), 'in 5 days'),
         (dict(months=1), 'in 1 month'),
         (dict(months=1, days=1), 'in 1 month and 1 day'),
         (dict(years=1, months=1), 'in 1 year and 1 month')
     ])
     def test_age_in_future(self, age_args, expected):
         from kallithea.lib.webutils import age
         with test_context(self.app):
             n = datetime.datetime(year=2012, month=5, day=17)
             delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
             assert age(n + delt(**age_args), now=n) == expected
             assert age(n + relativedelta.relativedelta(**age_args), now=n) == expected
     def test_tag_extractor(self):
         sample = (
             "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"
             "[requires] [stale] [see<>=>] [see => http://example.com]"
             "[requires => url] [lang => python] [just a tag]"
             "[,d] [ => ULR ] [obsolete] [desc]]"
+        )
         res = webutils.urlify_text(sample, stylize=True)
         assert '<div class="label label-meta" data-tag="tag">tag</div>' in res
         assert '<div class="label label-meta" data-tag="obsolete">obsolete</div>' in res
         assert '<div class="label label-meta" data-tag="stale">stale</div>' in res
         assert '<div class="label label-meta" data-tag="lang">python</div>' in res
         assert '<div class="label label-meta" data-tag="requires">requires =&gt; <a href="/url">url</a></div>' in res
         assert '<div class="label label-meta" data-tag="tag">tag</div>' in res
     def test_alternative_gravatar(self):
         _md5 = lambda s: hashlib.md5(safe_bytes(s)).hexdigest()
         def _md5(s):
             return hashlib.md5(safe_bytes(s)).hexdigest()
         # mock tg.tmpl_context
         def fake_tmpl_context(_url):
             _c = AttributeDict()
             _c.visual = AttributeDict()
             _c.visual.use_gravatar = True
             _c.visual.gravatar_url = _url
             return _c
         with mock.patch('kallithea.lib.webutils.url.current', lambda *a, **b: 'https://example.com'):
             fake = fake_tmpl_context(_url='http://example.com/{email}')
             with mock.patch('kallithea.lib.helpers.c', fake):
                     assert webutils.url.current() == 'https://example.com'
                     grav = h.gravatar_url(email_address='test@example.com', size=24)
                     assert grav == 'http://example.com/test@example.com'
             fake = fake_tmpl_context(_url='http://example.com/{email}')
             with mock.patch('kallithea.lib.helpers.c', fake):
                 grav = h.gravatar_url(email_address='test@example.com', size=24)
                 assert grav == 'http://example.com/test@example.com'
             fake = fake_tmpl_context(_url='http://example.com/{md5email}')
             with mock.patch('kallithea.lib.helpers.c', fake):
                 em = 'test@example.com'
                 grav = h.gravatar_url(email_address=em, size=24)
                 assert grav == 'http://example.com/%s' % (_md5(em))
             fake = fake_tmpl_context(_url='http://example.com/{md5email}/{size}')
             with mock.patch('kallithea.lib.helpers.c', fake):
                 em = 'test@example.com'
                 grav = h.gravatar_url(email_address=em, size=24)
                 assert grav == 'http://example.com/%s/%s' % (_md5(em), 24)
             fake = fake_tmpl_context(_url='{scheme}://{netloc}/{md5email}/{size}')
             with mock.patch('kallithea.lib.helpers.c', fake):
                 em = 'test@example.com'
                 grav = h.gravatar_url(email_address=em, size=24)
                 assert grav == 'https://example.com/%s/%s' % (_md5(em), 24)
     @base.parametrize('clone_uri_tmpl,repo_name,username,prefix,expected', [
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', None, '', 'http://vps1:8000/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', 'username', '', 'http://username@vps1:8000/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', None, '/prefix', 'http://vps1:8000/prefix/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', 'user', '/prefix', 'http://user@vps1:8000/prefix/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', 'username', '/prefix', 'http://username@vps1:8000/prefix/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', 'user', '/prefix/', 'http://user@vps1:8000/prefix/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', 'username', '/prefix/', 'http://username@vps1:8000/prefix/group/repo1'),
         ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', None, '', 'http://vps1:8000/_23'),
         ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', 'username', '', 'http://username@vps1:8000/_23'),
         ('http://{user}@{netloc}/_{repoid}', 'group/repo1', 'username', '', 'http://username@vps1:8000/_23'),
         ('http://{netloc}/_{repoid}', 'group/repo1', 'username', '', 'http://vps1:8000/_23'),
         ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', 'username', '', 'https://username@proxy1.example.com/group/repo1'),
         ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', None, '', 'https://proxy1.example.com/group/repo1'),
         ('https://proxy1.example.com/{user}/{repo}', 'group/repo1', 'username', '', 'https://proxy1.example.com/username/group/repo1'),
     ])
     def test_clone_url_generator(self, clone_uri_tmpl, repo_name, username, prefix, expected):
         clone_url = get_clone_url(clone_uri_tmpl=clone_uri_tmpl, prefix_url='http://vps1:8000' + prefix,
                                   repo_name=repo_name, repo_id=23, username=username)
         assert clone_url == expected
     def _quick_url(self, text, tmpl="""<a class="changeset_hash" href="%s">%s</a>""", url_=None):
         """
         Changes `some text url[foo]` => `some text <a href="/">foo</a>
         :param text:
         """
         # quickly change expected url[] into a link
         url_pattern = re.compile(r'(?:url\[)(.+?)(?:\])')
         def url_func(match_obj):
             _url = match_obj.groups()[0]
             return tmpl % (url_ or '/repo_name/changeset/%s' % _url, _url)
         return url_pattern.sub(url_func, text)
     @base.parametrize('sample,expected', [
       ("",
        ""),
       ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",
        """git-svn-id: <a href="https://svn.apache.org/repos/asf/libcloud/trunk@1441655">https://svn.apache.org/repos/asf/libcloud/trunk@1441655</a> 13f79535-47bb-0310-9956-ffa450edef68"""),
       ("from rev 000000000000",
        """from rev url[000000000000]"""),
       ("from rev 000000000000123123 also rev 000000000000",
        """from rev url[000000000000123123] also rev url[000000000000]"""),
       ("this should-000 00",
        """this should-000 00"""),
       ("longtextffffffffff rev 123123123123",
        """longtextffffffffff rev url[123123123123]"""),
       ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",
        """rev ffffffffffffffffffffffffffffffffffffffffffffffffff"""),
       ("ffffffffffff some text traalaa",
        """url[ffffffffffff] some text traalaa"""),
        ("""Multi line
        123123123123
        some text 123123123123
        sometimes !
        """,
        """Multi line<br/>"""
        """       url[123123123123]<br/>"""
        """       some text url[123123123123]<br/>"""
        """       sometimes !"""),
     ])
     def test_urlify_text(self, sample, expected):
         expected = self._quick_url(expected)
         with mock.patch('kallithea.lib.webutils.UrlGenerator.__call__',
             lambda self, name, **kwargs: dict(changeset_home='/%(repo_name)s/changeset/%(revision)s')[name] % kwargs,
         ):
             assert webutils.urlify_text(sample, 'repo_name') == expected
     @base.parametrize('sample,expected,url_', [
       ("",
        "",
        ""),
       ("https://svn.apache.org/repos",
        """url[https://svn.apache.org/repos]""",
        "https://svn.apache.org/repos"),
       ("http://svn.apache.org/repos",
        """url[http://svn.apache.org/repos]""",
        "http://svn.apache.org/repos"),
       ("from rev a also rev http://google.com",
        """from rev a also rev url[http://google.com]""",
        "http://google.com"),
       ("http://imgur.com/foo.gif inline http://imgur.com/foo.gif ending http://imgur.com/foo.gif",
        """url[http://imgur.com/foo.gif] inline url[http://imgur.com/foo.gif] ending url[http://imgur.com/foo.gif]""",
        "http://imgur.com/foo.gif"),
       ("""Multi line
        https://foo.bar.example.com
        some text lalala""",
        """Multi line<br/>"""
        """       url[https://foo.bar.example.com]<br/>"""
        """       some text lalala""",
        "https://foo.bar.example.com"),
       ("@mention @someone",
        """<b>@mention</b> <b>@someone</b>""",
        ""),
       ("deadbeefcafe 123412341234",
        """<a class="changeset_hash" href="/repo_name/changeset/deadbeefcafe">deadbeefcafe</a> <a class="changeset_hash" href="/repo_name/changeset/123412341234">123412341234</a>""",
        ""),
       ("We support * markup for *bold* markup of *single or multiple* words, "
        "*a bit @like http://slack.com*. "
        "The first * must come after whitespace and not be followed by whitespace, "
        "contain anything but * and newline until the next *, "
        "which must not come after whitespace "
        "and not be followed by * or alphanumerical *characters*.",
        """We support * markup for <b>*bold*</b> markup of <b>*single or multiple*</b> words, """
        """<b>*a bit <b>@like</b> <a href="http://slack.com">http://slack.com</a>*</b>. """
        """The first * must come after whitespace and not be followed by whitespace, """
        """contain anything but * and newline until the next *, """
        """which must not come after whitespace """
        """and not be followed by * or alphanumerical <b>*characters*</b>.""",
        "-"),
       ("HTML escaping: <abc> 'single' \"double\" &pointer",
        "HTML escaping: &lt;abc&gt; &#39;single&#39; &quot;double&quot; &amp;pointer",
        "-"),
       # tags are covered by test_tag_extractor
     ])
     def test_urlify_test(self, sample, expected, url_):
         expected = self._quick_url(expected,
                                    tmpl="""<a href="%s">%s</a>""", url_=url_)
         with mock.patch('kallithea.lib.webutils.UrlGenerator.__call__',
             lambda self, name, **kwargs: dict(changeset_home='/%(repo_name)s/changeset/%(revision)s')[name] % kwargs,
         ):
             assert webutils.urlify_text(sample, 'repo_name', stylize=True) == expected
     @base.parametrize('sample,expected', [
       ("deadbeefcafe @mention, and http://foo.bar/ yo",
        """<a class="changeset_hash" href="/repo_name/changeset/deadbeefcafe">deadbeefcafe</a>"""
        """<a class="message-link" href="#the-link"> <b>@mention</b>, and </a>"""
        """<a href="http://foo.bar/">http://foo.bar/</a>"""
        """<a class="message-link" href="#the-link"> yo</a>"""),
     ])
     def test_urlify_link(self, sample, expected):
         with mock.patch('kallithea.lib.webutils.UrlGenerator.__call__',
             lambda self, name, **kwargs: dict(changeset_home='/%(repo_name)s/changeset/%(revision)s')[name] % kwargs,
         ):
             assert webutils.urlify_text(sample, 'repo_name', link_='#the-link') == expected
     @base.parametrize('issue_pat,issue_server,issue_sub,sample,expected', [
         (r'#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'issue #123 and issue#456',
             """issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
             """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/456">#456</a>"""),
         (r'(?:\s*#)(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'issue #123 and issue#456',
             """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
             """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/456">#456</a>"""),
         # to require whitespace before the issue reference, one may be tempted to use \b...
         (r'\bPR(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'issue PR123 and issuePR456',
             """issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
             """issuePR456"""),
         # ... but it turns out that \b does not work well in combination with '#': the expectations
         # are reversed from what is actually happening.
         (r'\b#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'issue #123 and issue#456',
             """issue #123 and """
             """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/456">#456</a>"""),
         # ... so maybe try to be explicit? Unfortunately the whitespace before the issue
         # reference is not retained, again, because it is part of the pattern.
         (r'(?:^|\s)#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             '#15 and issue #123 and issue#456',
             """<a class="issue-tracker-link" href="http://foo/repo_name/issue/15">#15</a> and """
             """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
             """issue#456"""),
         # ... instead, use lookbehind assertions.
         (r'(?:^|(?<=\s))#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             '#15 and issue #123 and issue#456',
             """<a class="issue-tracker-link" href="http://foo/repo_name/issue/15">#15</a> and """
             """issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
             """issue#456"""),
         (r'(?:pullrequest|pull request|PR|pr) ?#?(\d+)', 'http://foo/{repo}/issue/\\1', 'PR#\\1',
             'fixed with pullrequest #1, pull request#2, PR 3, pr4',
             """fixed with <a class="issue-tracker-link" href="http://foo/repo_name/issue/1">PR#1</a>, """
             """<a class="issue-tracker-link" href="http://foo/repo_name/issue/2">PR#2</a>, """
             """<a class="issue-tracker-link" href="http://foo/repo_name/issue/3">PR#3</a>, """
             """<a class="issue-tracker-link" href="http://foo/repo_name/issue/4">PR#4</a>"""),
         (r'#(\d+)', 'http://foo/{repo}/issue/\\1', 'PR\\1',
             'interesting issue #123',
             """interesting issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">PR123</a>"""),
         (r'BUG\d{5}', 'https://bar/{repo}/\\1', '\\1',
             'silly me, I did not parenthesize the id, BUG12345.',
             """silly me, I did not parenthesize the id, <a class="issue-tracker-link" href="https://bar/repo_name/\\1">BUG12345</a>."""),
         (r'BUG(\d{5})', 'https://bar/{repo}/', 'BUG\\1',
             'silly me, the URL does not contain id, BUG12345.',
             """silly me, the URL does not contain id, <a class="issue-tracker-link" href="https://bar/repo_name/">BUG12345</a>."""),
         (r'(PR-\d+)', 'http://foo/{repo}/issue/\\1', '',
             'interesting issue #123, err PR-56',
             """interesting issue #123, err <a class="issue-tracker-link" href="http://foo/repo_name/issue/PR-56">PR-56</a>"""),
         (r'#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             "some 'standard' text with apostrophes",
             """some &#39;standard&#39; text with apostrophes"""),
         (r'#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             "some 'standard' issue #123",
             """some &#39;standard&#39; issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a>"""),
         (r'#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'an issue   #123       with extra whitespace',
             """an issue   <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a>       with extra whitespace"""),
         (r'(?:\s*#)(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
             'an issue   #123       with extra whitespace',
             """an issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a>       with extra whitespace"""),
         # invalid issue pattern
         (r'(PR\d+', 'http://foo/{repo}/issue/{id}', '',
             'PR135',
             """PR135"""),
         # other character than #
         (r'(?:^|(?<=\s))\$(\d+)', 'http://foo/{repo}/issue/\\1', '',
             'empty issue_sub $123 and issue$456',
             """empty issue_sub <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">$123</a> and """
             """issue$456"""),
         # named groups
         (r'(PR|pullrequest|pull request) ?(?P<sitecode>BRU|CPH|BER)-(?P<id>\d+)', r'http://foo/\g<sitecode>/pullrequest/\g<id>/', r'PR-\g<sitecode>-\g<id>',
             'pullrequest CPH-789 is similar to PRBRU-747',
             """<a class="issue-tracker-link" href="http://foo/CPH/pullrequest/789/">PR-CPH-789</a> is similar to """
             """<a class="issue-tracker-link" href="http://foo/BRU/pullrequest/747/">PR-BRU-747</a>"""),
     ])
     def test_urlify_issues(self, issue_pat, issue_server, issue_sub, sample, expected):
         config_stub = {
             'sqlalchemy.url': 'foo',
             'issue_pat': issue_pat,
             'issue_server_link': issue_server,
             'issue_sub': issue_sub,
+        }
         # force recreation of lazy function
         with mock.patch('kallithea.lib.webutils._urlify_issues_f', None):
             with mock.patch('kallithea.CONFIG', config_stub):
                 assert webutils.urlify_text(sample, 'repo_name') == expected
     @base.parametrize('sample,expected', [
         ('abc X5', 'abc <a class="issue-tracker-link" href="http://main/repo_name/main/5/">#5</a>'),
         ('abc pullrequest #6 xyz', 'abc <a class="issue-tracker-link" href="http://pr/repo_name/pr/6">PR#6</a> xyz'),
         ('pull request7 #', '<a class="issue-tracker-link" href="http://pr/repo_name/pr/7">PR#7</a> #'),
         ('look PR9 and pr #11', 'look <a class="issue-tracker-link" href="http://pr/repo_name/pr/9">PR#9</a> and <a class="issue-tracker-link" href="http://pr/repo_name/pr/11">PR#11</a>'),
         ('pullrequest#10 solves issue 9', '<a class="issue-tracker-link" href="http://pr/repo_name/pr/10">PR#10</a> solves <a class="issue-tracker-link" href="http://bug/repo_name/bug/9">bug#9</a>'),
         ('issue FAIL67', 'issue <a class="issue-tracker-link" href="http://fail/repo_name/67">FAIL67</a>'),
         ('issue FAILMORE89', 'issue FAILMORE89'), # no match because absent prefix
     ])
     def test_urlify_issues_multiple_issue_patterns(self, sample, expected):
         config_stub = {
             'sqlalchemy.url': r'foo',
             'issue_pat': r'X(\d+)',
             'issue_server_link': r'http://main/{repo}/main/\1/',
             'issue_sub': r'#\1',
             'issue_pat_pr': r'(?:pullrequest|pull request|PR|pr) ?#?(\d+)',
             'issue_server_link_pr': r'http://pr/{repo}/pr/\1',
             'issue_sub_pr': r'PR#\1',
             'issue_pat_bug': r'(?:BUG|bug|issue) ?#?(\d+)',
             'issue_server_link_bug': r'http://bug/{repo}/bug/\1',
             'issue_sub_bug': r'bug#\1',
             'issue_pat_empty_prefix': r'FAIL(\d+)',
             'issue_server_link_empty_prefix': r'http://fail/{repo}/\1',
             'issue_sub_empty_prefix': r'',
             'issue_pat_absent_prefix': r'FAILMORE(\d+)',
             'issue_server_link_absent_prefix': r'http://failmore/{repo}/\1',
+        }
         # force recreation of lazy function
         with mock.patch('kallithea.lib.webutils._urlify_issues_f', None):
             with mock.patch('kallithea.CONFIG', config_stub):
                 assert webutils.urlify_text(sample, 'repo_name') == expected
     @base.parametrize('test,expected', [
       ("", None),
       ("/_2", None),
       ("_2", 2),
       ("_2/", None),
     ])
     def test_get_permanent_id(self, test, expected):
         from kallithea.lib.utils import _get_permanent_id
         extracted = _get_permanent_id(test)
         assert extracted == expected, 'url:%s, got:`%s` expected: `%s`' % (test, base._test, expected)
     @base.parametrize('test,expected', [
       ("", ""),
       ("/", "/"),
       ("/_ID", '/_ID'),
       ("ID", "ID"),
       ("_ID", 'NAME'),
       ("_ID/", 'NAME/'),
       ("_ID/1/2", 'NAME/1/2'),
       ("_IDa", '_IDa'),
     ])
     def test_fix_repo_id_name(self, test, expected):
         repo = db.Repository.get_by_repo_name(base.HG_REPO)
         test = test.replace('ID', str(repo.repo_id))
         expected = expected.replace('NAME', repo.repo_name).replace('ID', str(repo.repo_id))
         from kallithea.lib.utils import fix_repo_id_name
         replaced = fix_repo_id_name(test)
         assert replaced == expected, 'url:%s, got:`%s` expected: `%s`' % (test, replaced, expected)
     @base.parametrize('canonical,test,expected', [
         ('http://www.example.org/', '/abc/xyz', 'http://www.example.org/abc/xyz'),
         ('http://www.example.org', '/abc/xyz', 'http://www.example.org/abc/xyz'),
         ('http://www.example.org', '/abc/xyz/', 'http://www.example.org/abc/xyz/'),
         ('http://www.example.org', 'abc/xyz/', 'http://www.example.org/abc/xyz/'),
         ('http://www.example.org', 'about', 'http://www.example.org/about-page'),
         ('http://www.example.org/repos/', 'abc/xyz/', 'http://www.example.org/repos/abc/xyz/'),
         ('http://www.example.org/kallithea/repos/', 'abc/xyz/', 'http://www.example.org/kallithea/repos/abc/xyz/'),
     ])
     def test_canonical_url(self, canonical, test, expected):
         # setup url(), used by canonical_url
         m = routes.Mapper()
         m.connect('about', '/about-page')
         url = routes.URLGenerator(m, {'HTTP_HOST': 'http_host.example.org'})
         config_mock = {
             'canonical_url': canonical,
+        }
         with test_context(self.app):
             request.environ['routes.url'] = url
             with mock.patch('kallithea.CONFIG', config_mock):
                 assert webutils.canonical_url(test) == expected
     @base.parametrize('canonical,expected', [
         ('http://www.example.org', 'www.example.org'),
         ('http://www.example.org/repos/', 'www.example.org'),
         ('http://www.example.org/kallithea/repos/', 'www.example.org'),
     ])
     def test_canonical_hostname(self, canonical, expected):
         # setup url(), used by canonical_hostname
         m = routes.Mapper()
         url = routes.URLGenerator(m, {'HTTP_HOST': 'http_host.example.org'})
         config_mock = {
             'canonical_url': canonical,
+        }
         with test_context(self.app):
             request.environ['routes.url'] = url
             with mock.patch('kallithea.CONFIG', config_mock):
                 assert webutils.canonical_hostname() == expected

kallithea/tests/vcs/test_git.py

➞

Show inline comments

 import datetime
 import os
 import sys
 import mock
 import pytest
 from kallithea.lib.vcs.backends.git import GitChangeset, GitRepository
 from kallithea.lib.vcs.exceptions import NodeDoesNotExistError, RepositoryError, VCSError
 from kallithea.lib.vcs.nodes import DirNode, FileNode, NodeKind, NodeState
 from kallithea.model.scm import ScmModel
 from kallithea.tests.vcs.base import _BackendTestMixin
 from kallithea.tests.vcs.conf import TEST_GIT_REPO, TEST_GIT_REPO_CLONE, TESTS_TMP_PATH, get_new_dir
 class TestGitRepository(object):
     def __check_for_existing_repo(self):
         if os.path.exists(TEST_GIT_REPO_CLONE):
             pytest.fail('Cannot test git clone repo as location %s already '
                       'exists. You should manually remove it first.'
                       % TEST_GIT_REPO_CLONE)
     def setup_method(self):
         self.repo = GitRepository(TEST_GIT_REPO)
     def test_wrong_repo_path(self):
         wrong_repo_path = os.path.join(TESTS_TMP_PATH, 'errorrepo')
         with pytest.raises(RepositoryError):
             GitRepository(wrong_repo_path)
     def test_git_cmd_injection(self):
         repo_inject_path = 'file:/%s; echo "Cake";' % TEST_GIT_REPO
         with pytest.raises(RepositoryError):
             # Should fail because URL will contain the parts after ; too
             GitRepository(get_new_dir('injection-repo'), src_url=repo_inject_path, update_after_clone=True, create=True)
         with pytest.raises(RepositoryError):
             # Should fail on direct clone call, which as of this writing does not happen outside of class
             clone_fail_repo = GitRepository(get_new_dir('injection-repo'), create=True)
             clone_fail_repo.clone(repo_inject_path, update_after_clone=True,)
         # Verify correct quoting of evil characters that should work on posix file systems
         if sys.platform == 'win32':
             # windows does not allow '"' in dir names
             # and some versions of the git client don't like ` and '
             tricky_path = get_new_dir("tricky-path-repo-$")
         else:
             tricky_path = get_new_dir("tricky-path-repo-$'\"`")
         successfully_cloned = GitRepository(tricky_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)
         # Repo should have been created
         assert not successfully_cloned._repo.bare
         if sys.platform == 'win32':
             # windows does not allow '"' in dir names
             # and some versions of the git client don't like ` and '
             tricky_path_2 = get_new_dir("tricky-path-2-repo-$")
         else:
             tricky_path_2 = get_new_dir("tricky-path-2-repo-$'\"`")
         successfully_cloned2 = GitRepository(tricky_path_2, src_url=tricky_path, bare=True, create=True)
         # Repo should have been created and thus used correct quoting for clone
         assert successfully_cloned2._repo.bare
         # Should pass because URL has been properly quoted
         successfully_cloned.pull(tricky_path_2)
         successfully_cloned2.fetch(tricky_path)
     def test_repo_create_with_spaces_in_path(self):
         repo_path = get_new_dir("path with spaces")
         repo = GitRepository(repo_path, src_url=None, bare=True, create=True)
         # Repo should have been created
         assert repo._repo.bare
     def test_repo_clone(self):
         self.__check_for_existing_repo()
         repo = GitRepository(TEST_GIT_REPO)
         repo_clone = GitRepository(TEST_GIT_REPO_CLONE,
             src_url=TEST_GIT_REPO, create=True, update_after_clone=True)
         assert len(repo.revisions) == len(repo_clone.revisions)
         # Checking hashes of changesets should be enough
         for changeset in repo.get_changesets():
             raw_id = changeset.raw_id
             assert raw_id == repo_clone.get_changeset(raw_id).raw_id
     def test_repo_clone_with_spaces_in_path(self):
         repo_path = get_new_dir("path with spaces")
         successfully_cloned = GitRepository(repo_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)
         # Repo should have been created
         assert not successfully_cloned._repo.bare
         successfully_cloned.pull(TEST_GIT_REPO)
         self.repo.fetch(repo_path)
     def test_repo_clone_without_create(self):
         with pytest.raises(RepositoryError):
             GitRepository(TEST_GIT_REPO_CLONE + '_wo_create', src_url=TEST_GIT_REPO)
     def test_repo_clone_with_update(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_with_update'
         repo_clone = GitRepository(clone_path,
             create=True, src_url=TEST_GIT_REPO, update_after_clone=True)
         assert len(repo.revisions) == len(repo_clone.revisions)
         # check if current workdir was updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         assert os.path.isfile(fpath) == True, 'Repo was cloned and updated but file %s could not be found' % fpath
     def test_repo_clone_without_update(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_without_update'
         repo_clone = GitRepository(clone_path,
             create=True, src_url=TEST_GIT_REPO, update_after_clone=False)
         assert len(repo.revisions) == len(repo_clone.revisions)
         # check if current workdir was *NOT* updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         # Make sure it's not bare repo
         assert not repo_clone._repo.bare
         assert os.path.isfile(fpath) == False, 'Repo was cloned and updated but file %s was found' % fpath
     def test_repo_clone_into_bare_repo(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_bare.git'
         repo_clone = GitRepository(clone_path, create=True,
             src_url=repo.path, bare=True)
         assert repo_clone._repo.bare
     def test_create_repo_is_not_bare_by_default(self):
         repo = GitRepository(get_new_dir('not-bare-by-default'), create=True)
         assert not repo._repo.bare
     def test_create_bare_repo(self):
         repo = GitRepository(get_new_dir('bare-repo'), create=True, bare=True)
         assert repo._repo.bare
     def test_revisions(self):
         # there are 112 revisions (by now)
         # so we can assume they would be available from now on
         subset = set([
             'c1214f7e79e02fc37156ff215cd71275450cffc3',
             '38b5fe81f109cb111f549bfe9bb6b267e10bc557',
             'fa6600f6848800641328adbf7811fd2372c02ab2',
             '102607b09cdd60e2793929c4f90478be29f85a17',
             '49d3fd156b6f7db46313fac355dca1a0b94a0017',
             '2d1028c054665b962fa3d307adfc923ddd528038',
             'd7e0d30fbcae12c90680eb095a4f5f02505ce501',
             'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
             '8430a588b43b5d6da365400117c89400326e7992',
             'd955cd312c17b02143c04fa1099a352b04368118',
             'f67b87e5c629c2ee0ba58f85197e423ff28d735b',
             'add63e382e4aabc9e1afdc4bdc24506c269b7618',
             'f298fe1189f1b69779a4423f40b48edf92a703fc',
             'bd9b619eb41994cac43d67cf4ccc8399c1125808',
             '6e125e7c890379446e98980d8ed60fba87d0f6d1',
             'd4a54db9f745dfeba6933bf5b1e79e15d0af20bd',
             '0b05e4ed56c802098dfc813cbe779b2f49e92500',
             '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
             '45223f8f114c64bf4d6f853e3c35a369a6305520',
             'ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
             'f5ea29fc42ef67a2a5a7aecff10e1566699acd68',
             '27d48942240f5b91dfda77accd2caac94708cc7d',
             '622f0eb0bafd619d2560c26f80f09e3b0b0d78af',
             'e686b958768ee96af8029fe19c6050b1a8dd3b2b'])
         assert subset.issubset(set(self.repo.revisions))
     def test_slicing(self):
         # 4 1 5 10 95
         for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),
                                  (10, 20, 10), (5, 100, 95)]:
             revs = list(self.repo[sfrom:sto])
             assert len(revs) == size
             assert revs[0] == self.repo.get_changeset(sfrom)
             assert revs[-1] == self.repo.get_changeset(sto - 1)
     def test_branches(self):
         # TODO: Need more tests here
         # Removed (those are 'remotes' branches for cloned repo)
         #assert 'master' in self.repo.branches
         #assert 'gittree' in self.repo.branches
         #assert 'web-branch' in self.repo.branches
         for name, id in self.repo.branches.items():
             assert isinstance(self.repo.get_changeset(id), GitChangeset)
     def test_tags(self):
         # TODO: Need more tests here
         assert 'v0.1.1' in self.repo.tags
         assert 'v0.1.2' in self.repo.tags
         for name, id in self.repo.tags.items():
             assert isinstance(self.repo.get_changeset(id), GitChangeset)
     def _test_single_changeset_cache(self, revision):
         chset = self.repo.get_changeset(revision)
         assert revision in self.repo.changesets
         assert chset is self.repo.changesets[revision]
     def test_initial_changeset(self):
         id = self.repo.revisions[0]
         init_chset = self.repo.get_changeset(id)
         assert init_chset.message == 'initial import\n'
         assert init_chset.author == 'Marcin Kuzminski <marcin@python-blog.com>'
         for path in ('vcs/__init__.py',
                      'vcs/backends/BaseRepository.py',
                      'vcs/backends/__init__.py'):
             assert isinstance(init_chset.get_node(path), FileNode)
         for path in ('', 'vcs', 'vcs/backends'):
             assert isinstance(init_chset.get_node(path), DirNode)
         with pytest.raises(NodeDoesNotExistError):
             init_chset.get_node(path='foobar')
         node = init_chset.get_node('vcs/')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
         node = init_chset.get_node('vcs')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
         node = init_chset.get_node('vcs/__init__.py')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.FILE
     def test_not_existing_changeset(self):
         with pytest.raises(RepositoryError):
             self.repo.get_changeset('f' * 40)
     def test_changeset10(self):
         chset10 = self.repo.get_changeset(self.repo.revisions[9])
         readme = b"""===
 VCS
 ===
 Various Version Control System management abstraction layer for Python.
 Introduction
 ------------
 TODO: To be written...
 """
         node = chset10.get_node('README.rst')
         assert node.kind == NodeKind.FILE
         assert node.content == readme
 class TestGitChangeset(object):
     def setup_method(self):
         self.repo = GitRepository(TEST_GIT_REPO)
     def test_default_changeset(self):
         tip = self.repo.get_changeset()
         assert tip == self.repo.get_changeset(None)
         assert tip == self.repo.get_changeset('tip')
     def test_root_node(self):
         tip = self.repo.get_changeset()
         assert tip.root is tip.get_node('')
     def test_lazy_fetch(self):
         """
         Test if changeset's nodes expands and are cached as we walk through
         the revision. This test is somewhat hard to write as order of tests
         is a key here. Written by running command after command in a shell.
         """
         commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         assert commit_id in self.repo.revisions
         chset = self.repo.get_changeset(commit_id)
         assert len(chset.nodes) == 0
         root = chset.root
         assert len(chset.nodes) == 1
         assert len(root.nodes) == 8
         # accessing root.nodes updates chset.nodes
         assert len(chset.nodes) == 9
         docs = root.get_node('docs')
         # we haven't yet accessed anything new as docs dir was already cached
         assert len(chset.nodes) == 9
         assert len(docs.nodes) == 8
         # accessing docs.nodes updates chset.nodes
         assert len(chset.nodes) == 17
         assert docs is chset.get_node('docs')
         assert docs is root.nodes[0]
         assert docs is root.dirs[0]
         assert docs is chset.get_node('docs')
     def test_nodes_with_changeset(self):
         commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         chset = self.repo.get_changeset(commit_id)
         root = chset.root
         docs = root.get_node('docs')
         assert docs is chset.get_node('docs')
         api = docs.get_node('api')
         assert api is chset.get_node('docs/api')
         index = api.get_node('index.rst')
         assert index is chset.get_node('docs/api/index.rst')
         assert index is chset.get_node('docs') \
                              .get_node('api') \
                              .get_node('index.rst')
     def test_branch_and_tags(self):
         # Those tests seem to show wrong results:
         # in Git, only heads have a branch - most changesets don't
         rev0 = self.repo.revisions[0]
         chset0 = self.repo.get_changeset(rev0)
         assert chset0.branch is None # should be 'master'?
         assert chset0.branches == [] # should be 'master'?
         assert chset0.tags == []
         rev10 = self.repo.revisions[10]
         chset10 = self.repo.get_changeset(rev10)
         assert chset10.branch is None # should be 'master'?
         assert chset10.branches == [] # should be 'master'?
         assert chset10.tags == []
         rev44 = self.repo.revisions[44]
         chset44 = self.repo.get_changeset(rev44)
         assert chset44.branch is None # should be 'web-branch'?
         assert chset44.branches == [] # should be 'web-branch'?
         tip = self.repo.get_changeset('tip')
         assert 'tip' not in tip.tags # it should be?
         assert not tip.tags # how it is!
     def _test_slices(self, limit, offset):
         count = self.repo.count()
         changesets = self.repo.get_changesets(limit=limit, offset=offset)
         idx = 0
         for changeset in changesets:
             rev = offset + idx
             idx += 1
             rev_id = self.repo.revisions[rev]
             if idx > limit:
                 pytest.fail("Exceeded limit already (getting revision %s, "
                     "there are %s total revisions, offset=%s, limit=%s)"
                     % (rev_id, count, offset, limit))
             assert changeset == self.repo.get_changeset(rev_id)
         result = list(self.repo.get_changesets(limit=limit, offset=offset))
         start = offset
         end = limit and offset + limit or None
         sliced = list(self.repo[start:end])
         pytest.assertEqual(result, sliced,
             msg="Comparison failed for limit=%s, offset=%s"
             "(get_changeset returned: %s and sliced: %s"
             % (limit, offset, result, sliced))
     def _test_file_size(self, revision, path, size):
         node = self.repo.get_changeset(revision).get_node(path)
         assert node.is_file()
         assert node.size == size
     def test_file_size(self):
         to_check = (
             ('c1214f7e79e02fc37156ff215cd71275450cffc3',
                 'vcs/backends/BaseRepository.py', 502),
             ('d7e0d30fbcae12c90680eb095a4f5f02505ce501',
                 'vcs/backends/hg.py', 854),
             ('6e125e7c890379446e98980d8ed60fba87d0f6d1',
                 'setup.py', 1068),
             ('d955cd312c17b02143c04fa1099a352b04368118',
                 'vcs/backends/base.py', 2921),
             ('ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
                 'vcs/backends/base.py', 3936),
             ('f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'vcs/backends/base.py', 6189),
+        )
         for revision, path, size in to_check:
             self._test_file_size(revision, path, size)
     def _test_dir_size(self, revision, path, size):
         node = self.repo.get_changeset(revision).get_node(path)
         assert node.size == size
     def test_dir_size(self):
         to_check = (
             ('5f2c6ee195929b0be80749243c18121c9864a3b3', '/', 674076),
             ('7ab37bc680b4aa72c34d07b230c866c28e9fc204', '/', 674049),
             ('6892503fb8f2a552cef5f4d4cc2cdbd13ae1cd2f', '/', 671830),
+        )
         for revision, path, size in to_check:
             self._test_dir_size(revision, path, size)
     def test_repo_size(self):
         assert self.repo.size == 674076
     def test_file_history(self):
         # we can only check if those revisions are present in the history
         # as we cannot update this test every time file is changed
         files = {
             'setup.py': [
                 '54386793436c938cff89326944d4c2702340037d',
                 '51d254f0ecf5df2ce50c0b115741f4cf13985dab',
                 '998ed409c795fec2012b1c0ca054d99888b22090',
                 '5e0eb4c47f56564395f76333f319d26c79e2fb09',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '7cb3fd1b6d8c20ba89e2264f1c8baebc8a52d36e',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/nodes.py': [
                 '33fa3223355104431402a888fa77a4e9956feb3e',
                 'fa014c12c26d10ba682fadb78f2a11c24c8118e1',
                 'e686b958768ee96af8029fe19c6050b1a8dd3b2b',
                 'ab5721ca0a081f26bf43d9051e615af2cc99952f',
                 'c877b68d18e792a66b7f4c529ea02c8f80801542',
                 '4313566d2e417cb382948f8d9d7c765330356054',
                 '6c2303a793671e807d1cfc70134c9ca0767d98c2',
                 '54386793436c938cff89326944d4c2702340037d',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
                 '1c6b3677b37ea064cb4b51714d8f7498f93f4b2b',
                 '2d03ca750a44440fb5ea8b751176d1f36f8e8f46',
                 '2a08b128c206db48c2f0b8f70df060e6db0ae4f8',
                 '30c26513ff1eb8e5ce0e1c6b477ee5dc50e2f34b',
                 'ac71e9503c2ca95542839af0ce7b64011b72ea7c',
                 '12669288fd13adba2a9b7dd5b870cc23ffab92d2',
                 '5a0c84f3e6fe3473e4c8427199d5a6fc71a9b382',
                 '12f2f5e2b38e6ff3fbdb5d722efed9aa72ecb0d5',
                 '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                 'f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'd7e390a45f6aa96f04f5e7f583ad4f867431aa25',
                 'f15c21f97864b4f071cddfbf2750ec2e23859414',
                 'e906ef056cf539a4e4e5fc8003eaf7cf14dd8ade',
                 'ea2b108b48aa8f8c9c4a941f66c1a03315ca1c3b',
                 '84dec09632a4458f79f50ddbbd155506c460b4f9',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '3bf1c5868e570e39569d094f922d33ced2fa3b2b',
                 'b8d04012574729d2c29886e53b1a43ef16dd00a1',
                 '6970b057cffe4aab0a792aa634c89f4bebf01441',
                 'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/backends/git.py': [
                 '4cf116ad5a457530381135e2f4c453e68a1b0105',
                 '9a751d84d8e9408e736329767387f41b36935153',
                 'cb681fb539c3faaedbcdf5ca71ca413425c18f01',
                 '428f81bb652bcba8d631bce926e8834ff49bdcc6',
                 '180ab15aebf26f98f714d8c68715e0f05fa6e1c7',
                 '2b8e07312a2e89e92b90426ab97f349f4bce2a3a',
                 '50e08c506174d8645a4bb517dd122ac946a0f3bf',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
             ],
+        }
         for path, revs in files.items():
             node = self.repo.get_changeset(revs[0]).get_node(path)
             node_revs = [chset.raw_id for chset in node.history]
             assert set(revs).issubset(set(node_revs)), "We assumed that %s is subset of revisions for which file %s " \
                 "has been changed, and history of that node returned: %s" \
                 % (revs, path, node_revs)
     def test_file_annotate(self):
         files = {
             'vcs/backends/__init__.py': {
                 'c1214f7e79e02fc37156ff215cd71275450cffc3': {
                     'lines_no': 1,
                     'changesets': [
                         'c1214f7e79e02fc37156ff215cd71275450cffc3',
                     ],
                 },
                 '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647': {
                     'lines_no': 21,
                     'changesets': [
                         '49d3fd156b6f7db46313fac355dca1a0b94a0017',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
                 'e29b67bd158580fc90fc5e9111240b90e6e86064': {
                     'lines_no': 32,
                     'changesets': [
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '54000345d2e78b03a99d561399e8e548de3f3203',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
             },
+        }
         for fname, revision_dict in files.items():
             for rev, data in revision_dict.items():
                 cs = self.repo.get_changeset(rev)
                 l1_1 = [x[1] for x in cs.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in cs.get_file_annotate(fname)]
                 assert l1_1 == l1_2
                 l1 = l1_1
                 l2 = files[fname][rev]['changesets']
                 assert l1 == l2, "The lists of revision for %s@rev %s" \
                     "from annotation list should match each other, " \
                     "got \n%s \nvs \n%s " % (fname, rev, l1, l2)
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         node = self.repo \
             .get_changeset('e6ea6d16e2f26250124a1f4b4fe37a912f9d86a0') \
             .get_node('vcs/utils/diffs.py')
         assert node.state, NodeState.ADDED
         assert node.added
         assert not node.changed
         assert not node.not_changed
         assert not node.removed
         node = self.repo \
             .get_changeset('33fa3223355104431402a888fa77a4e9956feb3e') \
             .get_node('.hgignore')
         assert node.state, NodeState.CHANGED
         assert not node.added
         assert node.changed
         assert not node.not_changed
         assert not node.removed
         node = self.repo \
             .get_changeset('e29b67bd158580fc90fc5e9111240b90e6e86064') \
             .get_node('setup.py')
         assert node.state, NodeState.NOT_CHANGED
         assert not node.added
         assert not node.changed
         assert node.not_changed
         assert not node.removed
         # If node has REMOVED state then trying to fetch it would raise
         # ChangesetError exception
         chset = self.repo.get_changeset(
             'fa6600f6848800641328adbf7811fd2372c02ab2')
         path = 'vcs/backends/BaseRepository.py'
         with pytest.raises(NodeDoesNotExistError):
             chset.get_node(path)
         # but it would be one of ``removed`` (changeset's attribute)
         assert path in [rf.path for rf in chset.removed]
         chset = self.repo.get_changeset(
             '54386793436c938cff89326944d4c2702340037d')
         changed = ['setup.py', 'tests/test_nodes.py', 'vcs/backends/hg.py',
             'vcs/nodes.py']
         assert set(changed) == set([f.path for f in chset.changed])
     def test_commit_message_is_str(self):
         for cs in self.repo:
             assert isinstance(cs.message, str)
     def test_changeset_author_is_str(self):
         for cs in self.repo:
             assert isinstance(cs.author, str)
     def test_repo_files_content_is_bytes(self):
         changeset = self.repo.get_changeset()
         for node in changeset.get_node('/'):
             if node.is_file():
                 assert isinstance(node.content, bytes)
     def test_wrong_path(self):
         # There is 'setup.py' in the root dir but not there:
         path = 'foo/bar/setup.py'
         tip = self.repo.get_changeset()
         with pytest.raises(VCSError):
             tip.get_node(path)
     def test_author_email(self):
         assert 'marcin@python-blog.com' == self.repo.get_changeset('c1214f7e79e02fc37156ff215cd71275450cffc3').author_email
         assert 'lukasz.balcerzak@python-center.pl' == self.repo.get_changeset('ff7ca51e58c505fec0dd2491de52c622bb7a806b').author_email
         assert '' == self.repo.get_changeset('8430a588b43b5d6da365400117c89400326e7992').author_email
     def test_author_username(self):
         assert 'Marcin Kuzminski' == self.repo.get_changeset('c1214f7e79e02fc37156ff215cd71275450cffc3').author_name
         assert 'Lukasz Balcerzak' == self.repo.get_changeset('ff7ca51e58c505fec0dd2491de52c622bb7a806b').author_name
         assert 'marcink none@none' == self.repo.get_changeset('8430a588b43b5d6da365400117c89400326e7992').author_name
 class TestGitSpecificWithRepo(_BackendTestMixin):
     backend_alias = 'git'
     @classmethod
     def _get_commits(cls):
         return [
+            {
                 'message': 'Initial',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar/static/js/admin/base.js', content='base'),
                     FileNode('foobar/static/admin', content='admin',
                         mode=0o120000), # this is a link
                     FileNode('foo', content='foo'),
                 ],
             },
+            {
                 'message': 'Second',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('foo2', content='foo2'),
                 ],
             },
+        ]
     def test_paths_slow_traversing(self):
         cs = self.repo.get_changeset()
         assert cs.get_node('foobar').get_node('static').get_node('js').get_node('admin').get_node('base.js').content == b'base'
     def test_paths_fast_traversing(self):
         cs = self.repo.get_changeset()
         assert cs.get_node('foobar/static/js/admin/base.js').content == b'base'
     def test_workdir_get_branch(self):
         self.repo.run_git_command(['checkout', '-b', 'production'])
         # Regression test: one of following would fail if we don't check
         # .git/HEAD file
         self.repo.run_git_command(['checkout', 'production'])
         assert self.repo.workdir.get_branch() == 'production'
         self.repo.run_git_command(['checkout', 'master'])
         assert self.repo.workdir.get_branch() == 'master'
     def test_get_diff_runs_git_command_with_hashes(self):
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1)
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1)], cwd=self.repo.path)
     def test_get_diff_runs_git_command_with_str_hashes(self):
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(self.repo.EMPTY_CHANGESET, 1)
         self.repo._run_git_command.assert_called_once_with(
             ['show', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(1)], cwd=self.repo.path)
     def test_get_diff_runs_git_command_with_path_if_its_given(self):
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1, 'foo')
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)
     def test_get_diff_does_not_sanitize_valid_context(self):
         almost_overflowed_long_int = 2**31-1
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1, 'foo', context=almost_overflowed_long_int)
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U' + str(almost_overflowed_long_int), '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)
     def test_get_diff_sanitizes_overflowing_context(self):
         overflowed_long_int = 2**31
         sanitized_overflowed_long_int = overflowed_long_int-1
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1, 'foo', context=overflowed_long_int)
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U' + str(sanitized_overflowed_long_int), '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)
     def test_get_diff_does_not_sanitize_zero_context(self):
         zero_context = 0
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1, 'foo', context=zero_context)
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U' + str(zero_context), '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)
     def test_get_diff_sanitizes_negative_context(self):
         negative_context = -10
         self.repo._run_git_command = mock.Mock(return_value=(b'', b''))
         self.repo.get_diff(0, 1, 'foo', context=negative_context)
         self.repo._run_git_command.assert_called_once_with(
             ['diff', '-U0', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'], cwd=self.repo.path)
 class TestGitRegression(_BackendTestMixin):
     backend_alias = 'git'
     @classmethod
     def _get_commits(cls):
         return [
+            {
                 'message': 'Initial',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('bot/__init__.py', content='base'),
                     FileNode('bot/templates/404.html', content='base'),
                     FileNode('bot/templates/500.html', content='base'),
                 ],
             },
+            {
                 'message': 'Second',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('bot/build/migrations/1.py', content='foo2'),
                     FileNode('bot/build/migrations/2.py', content='foo2'),
                     FileNode('bot/build/static/templates/f.html', content='foo2'),
                     FileNode('bot/build/static/templates/f1.html', content='foo2'),
                     FileNode('bot/build/templates/err.html', content='foo2'),
                     FileNode('bot/build/templates/err2.html', content='foo2'),
                 ],
             },
+        ]
     def test_similar_paths(self):
         cs = self.repo.get_changeset()
         paths = lambda *n: [x.path for x in n]
         def paths(*n):
             return [x.path for x in n]
         assert paths(*cs.get_nodes('bot')) == ['bot/build', 'bot/templates', 'bot/__init__.py']
         assert paths(*cs.get_nodes('bot/build')) == ['bot/build/migrations', 'bot/build/static', 'bot/build/templates']
         assert paths(*cs.get_nodes('bot/build/static')) == ['bot/build/static/templates']
         # this get_nodes below causes troubles !
         assert paths(*cs.get_nodes('bot/build/static/templates')) == ['bot/build/static/templates/f.html', 'bot/build/static/templates/f1.html']
         assert paths(*cs.get_nodes('bot/build/templates')) == ['bot/build/templates/err.html', 'bot/build/templates/err2.html']
         assert paths(*cs.get_nodes('bot/templates/')) == ['bot/templates/404.html', 'bot/templates/500.html']
 class TestGitHooks(object):
     """
     Tests related to hook functionality of Git repositories.
     """
     def setup_method(self):
         # For each run we want a fresh repo.
         self.repo_directory = get_new_dir("githookrepo")
         self.repo = GitRepository(self.repo_directory, create=True)
         # Create a dictionary where keys are hook names, and values are paths to
         # them in the non-bare repo. Deduplicates code in tests a bit.
         self.pre_receive = os.path.join(self.repo.path, '.git', 'hooks', "pre-receive")
         self.post_receive = os.path.join(self.repo.path, '.git', 'hooks', "post-receive")
         self.kallithea_hooks = {
             "pre-receive": self.pre_receive,
             "post-receive": self.post_receive,
+        }
     def test_hooks_created_if_missing(self):
         """
         Tests if hooks are installed in repository if they are missing.
         """
         for hook, hook_path in self.kallithea_hooks.items():
             if os.path.exists(hook_path):
                 os.remove(hook_path)
         ScmModel().install_git_hooks(self.repo)
         assert not os.path.exists(self.pre_receive)
         assert os.path.exists(self.post_receive)
     def test_kallithea_hooks_updated(self):
         """
         Tests if hooks are updated if they are Kallithea hooks already.
         """
         for hook, hook_path in self.kallithea_hooks.items():
             with open(hook_path, "w") as f:
                 f.write("KALLITHEA_HOOK_VER=0.0.0\nJUST_BOGUS")
         ScmModel().install_git_hooks(self.repo)
         assert not os.path.exists(self.pre_receive)
         with open(self.post_receive) as f:
             assert "JUST_BOGUS" not in f.read()
     def test_custom_hooks_untouched(self):
         """
         Tests if hooks are left untouched if they are not Kallithea hooks.
         """
         for hook, hook_path in self.kallithea_hooks.items():
             with open(hook_path, "w") as f:
                 f.write("#!/bin/bash\n#CUSTOM_HOOK")
         ScmModel().install_git_hooks(self.repo)
         for hook, hook_path in self.kallithea_hooks.items():
             with open(hook_path) as f:
                 assert "CUSTOM_HOOK" in f.read()
     def test_custom_hooks_forced_update(self):
         """
         Tests if hooks are forcefully updated even though they are custom hooks.
         """
         for hook, hook_path in self.kallithea_hooks.items():
             with open(hook_path, "w") as f:
                 f.write("#!/bin/bash\n#CUSTOM_HOOK")
         ScmModel().install_git_hooks(self.repo, force=True)
         with open(self.pre_receive) as f:
             assert "KALLITHEA_HOOK_VER" not in f.read()
         with open(self.post_receive) as f:
             assert "KALLITHEA_HOOK_VER" in f.read()

0 comments (0 inline, 0 general)