kallithea Changeset - 6af08d44daa8

Changeset - 6af08d44daa8

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

Mads Kiilerich - 7 years ago 2018-08-08 02:23:11
mads@kiilerich.com

git: fix push to empty repo (Issue 323)

Git would fail to log revisions when the list of heads to exclude included an
empty string (in place of the pushed ref).

To avoid that, skip the skipped revision instead of making it an empty string.
`git log --not` works fine without providing any revisions to "not".

Verify in test_push_new_repo_git that it actually logged the push.

2 files changed with 7 insertions and 7 deletions:

kallithea/lib/hooks.py

kallithea/tests/other/test_vcs_operations.py

0 comments (0 inline, 0 general)

kallithea/lib/hooks.py

➞

Show inline comments

@@ @@ -66,406 +66,406 @@ def _get_scm_size(alias, root_path): @@
 def repo_size(ui, repo, hooktype=None, **kwargs):
     """Presents size of repository after push"""
     size_hg_f, size_root_f, size_total_f = _get_scm_size('.hg', repo.root)
     last_cs = repo[len(repo) - 1]
     msg = ('Repository size .hg: %s Checkout: %s Total: %s\n'
            'Last revision is now r%s:%s\n') % (
         size_hg_f, size_root_f, size_total_f, last_cs.rev(), last_cs.hex()[:12]
+    )
     ui.status(msg)
 def push_lock_handling(ui, repo, **kwargs):
     """Pre push function, currently used to ban pushing when repository is locked.
     Called as Mercurial hook prechangegroup.push_lock_handling or from the Git pre-receive hook calling handle_git_pre_receive.
     """
     ex = _extract_extras()
     usr = User.get_by_username(ex.username)
     if ex.locked_by[0] and usr.user_id != int(ex.locked_by[0]):
         locked_by = User.get(ex.locked_by[0]).username
         # this exception is interpreted in git/hg middlewares and based
         # on that proper return code is server to client
         _http_ret = HTTPLockedRC(ex.repository, locked_by)
         if str(_http_ret.code).startswith('2'):
             # 2xx Codes don't raise exceptions
             ui.status(safe_str(_http_ret.title))
         else:
             raise _http_ret
 def pull_lock_handling(ui, repo, **kwargs):
     """Called as Mercurial hook preoutgoing.pull_lock_handling or from Kallithea before invoking Git"""
     # pre pull function ...
     ex = _extract_extras()
     if ex.locked_by[0]:
         locked_by = User.get(ex.locked_by[0]).username
         # this exception is interpreted in git/hg middlewares and based
         # on that proper return code is server to client
         _http_ret = HTTPLockedRC(ex.repository, locked_by)
         if str(_http_ret.code).startswith('2'):
             # 2xx Codes don't raise exceptions
             ui.status(safe_str(_http_ret.title))
         else:
             raise _http_ret
 def log_pull_action(ui, repo, **kwargs):
     """Logs user last pull action, and also handle locking
     Called as Mercurial hook outgoing.pull_logger or from Kallithea before invoking Git.
     """
     ex = _extract_extras()
     user = User.get_by_username(ex.username)
     action = 'pull'
     action_logger(user, action, ex.repository, ex.ip, commit=True)
     # extension hook call
     from kallithea import EXTENSIONS
     callback = getattr(EXTENSIONS, 'PULL_HOOK', None)
     if callable(callback):
         kw = {}
         kw.update(ex)
         callback(**kw)
     if ex.make_lock is not None and ex.make_lock:
         Repository.lock(Repository.get_by_repo_name(ex.repository), user.user_id)
         #msg = 'Made lock on repo `%s`' % repository
         #ui.status(msg)
     if ex.locked_by[0]:
         locked_by = User.get(ex.locked_by[0]).username
         _http_ret = HTTPLockedRC(ex.repository, locked_by)
         if str(_http_ret.code).startswith('2'):
             # 2xx Codes don't raise exceptions
             ui.status(safe_str(_http_ret.title))
     return 0
 def log_push_action(ui, repo, **kwargs):
     """
     Register that changes have been pushed - log it *and* invalidate caches.
     Note: It is not only logging, but also the side effect invalidating cahes!
     The function should perhaps be renamed.
     Called as Mercurial hook changegroup.push_logger or from the Git
     post-receive hook calling handle_git_post_receive ... or from scm _handle_push.
     Revisions are passed in different hack-ish ways.
     """
     ex = _extract_extras()
     action_tmpl = ex.action + ':%s'
     revs = []
     if ex.scm == 'hg':
         node = kwargs['node']
         def get_revs(repo, rev_opt):
             if rev_opt:
                 revs = revrange(repo, rev_opt)
                 if len(revs) == 0:
                     return (nullrev, nullrev)
                 return max(revs), min(revs)
             else:
                 return len(repo) - 1, 0
         stop, start = get_revs(repo, [node + ':'])
         _h = binascii.hexlify
         revs = [_h(repo[r].node()) for r in xrange(start, stop + 1)]
     elif ex.scm == 'git':
         revs = kwargs.get('_git_revs', [])
         if '_git_revs' in kwargs:
             kwargs.pop('_git_revs')
     action = action_tmpl % ','.join(revs)
     action_logger(ex.username, action, ex.repository, ex.ip, commit=True)
     from kallithea.model.scm import ScmModel
     ScmModel().mark_for_invalidation(ex.repository)
     # extension hook call
     from kallithea import EXTENSIONS
     callback = getattr(EXTENSIONS, 'PUSH_HOOK', None)
     if callable(callback):
         kw = {'pushed_revs': revs}
         kw.update(ex)
         callback(**kw)
     if ex.make_lock is not None and not ex.make_lock:
         Repository.unlock(Repository.get_by_repo_name(ex.repository))
         ui.status(safe_str('Released lock on repo `%s`\n' % ex.repository))
     if ex.locked_by[0]:
         locked_by = User.get(ex.locked_by[0]).username
         _http_ret = HTTPLockedRC(ex.repository, locked_by)
         if str(_http_ret.code).startswith('2'):
             # 2xx Codes don't raise exceptions
             ui.status(safe_str(_http_ret.title))
     return 0
 def log_create_repository(repository_dict, created_by, **kwargs):
     """
     Post create repository Hook.
     :param repository: dict dump of repository object
     :param created_by: username who created repository
     available keys of repository_dict:
      'repo_type',
      'description',
      'private',
      'created_on',
      'enable_downloads',
      'repo_id',
      'owner_id',
      'enable_statistics',
      'clone_uri',
      'fork_id',
      'group_id',
      'repo_name'
     """
     from kallithea import EXTENSIONS
     callback = getattr(EXTENSIONS, 'CREATE_REPO_HOOK', None)
     if callable(callback):
         kw = {}
         kw.update(repository_dict)
         kw.update({'created_by': created_by})
         kw.update(kwargs)
         return callback(**kw)
     return 0
 def check_allowed_create_user(user_dict, created_by, **kwargs):
     # pre create hooks
     from kallithea import EXTENSIONS
     callback = getattr(EXTENSIONS, 'PRE_CREATE_USER_HOOK', None)
     if callable(callback):
         allowed, reason = callback(created_by=created_by, **user_dict)
         if not allowed:
             raise UserCreationError(reason)
 def log_create_user(user_dict, created_by, **kwargs):
     """
     Post create user Hook.
     :param user_dict: dict dump of user object
     available keys for user_dict:
      'username',
      'full_name_or_username',
      'full_contact',
      'user_id',
      'name',
      'firstname',
      'short_contact',
      'admin',
      'lastname',
      'ip_addresses',
      'ldap_dn',
      'email',
      'api_key',
      'last_login',
      'full_name',
      'active',
      'password',
      'emails',
      'inherit_default_permissions'
     """
     from kallithea import EXTENSIONS
     callback = getattr(EXTENSIONS, 'CREATE_USER_HOOK', None)
     if callable(callback):
         return callback(created_by=created_by, **user_dict)
     return 0
 def log_delete_repository(repository_dict, deleted_by, **kwargs):
     """
     Post delete repository Hook.
     :param repository: dict dump of repository object
     :param deleted_by: username who deleted the repository
     available keys of repository_dict:
      'repo_type',
      'description',
      'private',
      'created_on',
      'enable_downloads',
      'repo_id',
      'owner_id',
      'enable_statistics',
      'clone_uri',
      'fork_id',
      'group_id',
      'repo_name'
     """
     from kallithea import EXTENSIONS
     callback = getattr(EXTENSIONS, 'DELETE_REPO_HOOK', None)
     if callable(callback):
         kw = {}
         kw.update(repository_dict)
         kw.update({'deleted_by': deleted_by,
                    'deleted_on': time.time()})
         kw.update(kwargs)
         return callback(**kw)
     return 0
 def log_delete_user(user_dict, deleted_by, **kwargs):
     """
     Post delete user Hook.
     :param user_dict: dict dump of user object
     available keys for user_dict:
      'username',
      'full_name_or_username',
      'full_contact',
      'user_id',
      'name',
      'firstname',
      'short_contact',
      'admin',
      'lastname',
      'ip_addresses',
      'ldap_dn',
      'email',
      'api_key',
      'last_login',
      'full_name',
      'active',
      'password',
      'emails',
      'inherit_default_permissions'
     """
     from kallithea import EXTENSIONS
     callback = getattr(EXTENSIONS, 'DELETE_USER_HOOK', None)
     if callable(callback):
         return callback(deleted_by=deleted_by, **user_dict)
     return 0
 def _hook_environment(repo_path):
     """
     Create a light-weight environment for stand-alone scripts and return an UI and the
     db repository.
     Git hooks are executed as subprocess of Git while Kallithea is waiting, and
     they thus need enough info to be able to create an app environment and
     connect to the database.
     """
     from paste.deploy import appconfig
     from sqlalchemy import engine_from_config
     from kallithea.config.environment import load_environment
     from kallithea.model.base import init_model
     extras = _extract_extras()
     ini_file_path = extras['config']
     #logging.config.fileConfig(ini_file_path) # Note: we are in a different process - don't use configured logging
     app_conf = appconfig('config:%s' % ini_file_path)
     conf = load_environment(app_conf.global_conf, app_conf.local_conf)
     setup_cache_regions(conf)
     engine = engine_from_config(conf, 'sqlalchemy.')
     init_model(engine)
     repo_path = safe_unicode(repo_path)
     # fix if it's not a bare repo
     if repo_path.endswith(os.sep + '.git'):
         repo_path = repo_path[:-5]
     repo = Repository.get_by_full_path(repo_path)
     if not repo:
         raise OSError('Repository %s not found in database'
                       % (safe_str(repo_path)))
     baseui = make_ui('db')
     return baseui, repo
 def handle_git_pre_receive(repo_path, git_stdin_lines):
     """Called from Git pre-receive hook"""
     baseui, repo = _hook_environment(repo_path)
     scm_repo = repo.scm_instance
     push_lock_handling(baseui, scm_repo)
     return 0
 def handle_git_post_receive(repo_path, git_stdin_lines):
     """Called from Git post-receive hook"""
     baseui, repo = _hook_environment(repo_path)
     # the post push hook should never use the cached instance
     scm_repo = repo.scm_instance_no_cache()
     _hooks = dict(baseui.configitems('hooks')) or {}
     # if push hook is enabled via web interface
     if _hooks.get(Ui.HOOK_PUSH_LOG):
         rev_data = []
         for l in git_stdin_lines:
             old_rev, new_rev, ref = l.strip().split(' ')
             _ref_data = ref.split('/')
             if _ref_data[1] in ['tags', 'heads']:
                 rev_data.append({'old_rev': old_rev,
                                  'new_rev': new_rev,
                                  'ref': ref,
                                  'type': _ref_data[1],
                                  'name': '/'.join(_ref_data[2:])})
         git_revs = []
         for push_ref in rev_data:
             _type = push_ref['type']
             if _type == 'heads':
                 if push_ref['old_rev'] == EmptyChangeset().raw_id:
                     # update the symbolic ref if we push new repo
                     if scm_repo.is_empty():
                         scm_repo._repo.refs.set_symbolic_ref('HEAD',
                                             'refs/heads/%s' % push_ref['name'])
                     # build exclude list without the ref
                     cmd = ['for-each-ref', '--format=%(refname)', 'refs/heads/*']
                     stdout, stderr = scm_repo.run_git_command(cmd)
                     ref = push_ref['ref']
-                    heads = [head if head != ref else '' for head in stdout.splitlines()]
+                    heads = [head for head in stdout.splitlines() if head != ref]
                     # now list the git revs while excluding from the list
                     cmd = ['log', push_ref['new_rev'], '--reverse', '--pretty=format:%H']
                     cmd.append('--not')
                     cmd.extend(heads)
+                    cmd.extend(heads) # empty list is ok
                     stdout, stderr = scm_repo.run_git_command(cmd)
                     git_revs += stdout.splitlines()
                 elif push_ref['new_rev'] == EmptyChangeset().raw_id:
                     # delete branch case
                     git_revs += ['delete_branch=>%s' % push_ref['name']]
                 else:
                     cmd = ['log', '%(old_rev)s..%(new_rev)s' % push_ref,
                            '--reverse', '--pretty=format:%H']
                     stdout, stderr = scm_repo.run_git_command(cmd)
                     git_revs += stdout.splitlines()
             elif _type == 'tags':
                 git_revs += ['tag=>%s' % push_ref['name']]
         log_push_action(baseui, scm_repo, _git_revs=git_revs)
     return 0

kallithea/tests/other/test_vcs_operations.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/>.
 """
 Test suite for vcs push/pull operations.
 The tests need Git > 1.8.1.
 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: Dec 30, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import re
 import tempfile
 import time
 import urllib2
 import json
 from tempfile import _RandomNameSequence
 from subprocess import Popen, PIPE
 import pytest
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.model.db import User, Repository, UserIpMap, CacheInvalidation, Ui, UserLog
 from kallithea.model.meta import Session
 from kallithea.model.repo import RepoModel
 from kallithea.model.user import UserModel
 DEBUG = True
 HOST = '127.0.0.1:4999'  # test host
 fixture = Fixture()
 class Command(object):
     def __init__(self, cwd):
         self.cwd = cwd
     def execute(self, cmd, *args, **environ):
         """
         Runs command on the system with given ``args``.
         """
         command = cmd + ' ' + ' '.join(args)
         ignoreReturnCode = environ.pop('ignoreReturnCode', False)
         if DEBUG:
             print '*** CMD %s ***' % command
         testenv = dict(os.environ)
         testenv['LANG'] = 'en_US.UTF-8'
         testenv['LANGUAGE'] = 'en_US:en'
         testenv['HGPLAIN'] = ''
         testenv['HGRCPATH'] = ''
         testenv.update(environ)
         p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd, env=testenv)
         stdout, stderr = p.communicate()
         if DEBUG:
             if stdout:
                 print 'stdout:', repr(stdout)
             if stderr:
                 print 'stderr:', repr(stderr)
         if not ignoreReturnCode:
             assert p.returncode == 0
         return stdout, stderr
 def _get_tmp_dir(prefix='vcs_operations-', suffix=''):
     return tempfile.mkdtemp(dir=TESTS_TMP_PATH, prefix=prefix, suffix=suffix)
 def _add_files(vcs, dest_dir, files_no=3):
     """
     Generate some files, add it to dest_dir repo and push back
     vcs is git or hg and defines what VCS we want to make those files for
     :param vcs:
     :param dest_dir:
     """
     added_file = '%ssetup.py' % _RandomNameSequence().next()
     open(os.path.join(dest_dir, added_file), 'a').close()
     Command(dest_dir).execute('%s add %s' % (vcs, added_file))
     email = 'me@example.com'
     if os.name == 'nt':
         author_str = 'User <%s>' % email
     else:
         author_str = 'User ǝɯɐᴎ <%s>' % email
     for i in xrange(files_no):
         cmd = """echo "added_line%s" >> %s""" % (i, added_file)
         Command(dest_dir).execute(cmd)
         if vcs == 'hg':
             cmd = """hg commit -m "committed new %s" -u "%s" "%s" """ % (
                 i, author_str, added_file
+            )
         elif vcs == 'git':
             cmd = """git commit -m "committed new %s" --author "%s" "%s" """ % (
                 i, author_str, added_file
+            )
         # git commit needs EMAIL on some machines
         Command(dest_dir).execute(cmd, EMAIL=email)
 def _add_files_and_push(webserver, vcs, dest_dir, ignoreReturnCode=False, files_no=3,
                             clone_url=None, username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS):
     _add_files(vcs, dest_dir, files_no=files_no)
     # PUSH it back
     _REPO = None
     if vcs == 'hg':
         _REPO = HG_REPO
     elif vcs == 'git':
         _REPO = GIT_REPO
     if clone_url is None:
         clone_url = webserver.repo_url(_REPO, username=username, password=password)
     stdout = stderr = None
     if vcs == 'hg':
         stdout, stderr = Command(dest_dir).execute('hg push --verbose', clone_url, ignoreReturnCode=ignoreReturnCode)
     elif vcs == 'git':
         stdout, stderr = Command(dest_dir).execute('git push --verbose', clone_url, "master", ignoreReturnCode=ignoreReturnCode)
     return stdout, stderr
 def _check_outgoing(vcs, cwd, clone_url=''):
     if vcs == 'hg':
         # hg removes the password from default URLs, so we have to provide it here via the clone_url
         return Command(cwd).execute('hg -q outgoing', clone_url, ignoreReturnCode=True)
     elif vcs == 'git':
         Command(cwd).execute('git remote update')
         return Command(cwd).execute('git log origin/master..master')
 def set_anonymous_access(enable=True):
     user = User.get_default_user()
     user.active = enable
     Session().commit()
     if enable != User.get_default_user().active:
         raise Exception('Cannot set anonymous access')
 #==============================================================================
 # TESTS
 #==============================================================================
 def _check_proper_git_push(stdout, stderr):
     # WTF Git stderr is output ?!
     assert 'fatal' not in stderr
     assert 'rejected' not in stderr
     assert 'Pushing to' in stderr
     assert 'master -> master' in stderr
 @pytest.mark.usefixtures("test_context_fixture")
 class TestVCSOperations(TestController):
     @classmethod
     def setup_class(cls):
         # DISABLE ANONYMOUS ACCESS
         set_anonymous_access(False)
     def setup_method(self, method):
         r = Repository.get_by_repo_name(GIT_REPO)
         Repository.unlock(r)
         r.enable_locking = False
         Session().commit()
         r = Repository.get_by_repo_name(HG_REPO)
         Repository.unlock(r)
         r.enable_locking = False
         Session().commit()
     @pytest.fixture()
     def testhook_cleanup(self):
         yield
         # remove hook
         for hook in ['prechangegroup', 'pretxnchangegroup', 'preoutgoing', 'changegroup', 'outgoing', 'incoming']:
             entry = Ui.get_by_key('hooks', '%s.testhook' % hook)
             if entry:
                 Session().delete(entry)
         Session().commit()
     @pytest.fixture(scope="module")
     def testfork(self):
         # create fork so the repo stays untouched
         git_fork_name = u'%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())
         fixture.create_fork(GIT_REPO, git_fork_name)
         hg_fork_name = u'%s_fork%s' % (HG_REPO, _RandomNameSequence().next())
         fixture.create_fork(HG_REPO, hg_fork_name)
         return {'git': git_fork_name, 'hg': hg_fork_name}
     def test_clone_hg_repo_by_admin(self, webserver):
         clone_url = webserver.repo_url(HG_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir())
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
         assert stderr == ''
     def test_clone_git_repo_by_admin(self, webserver):
         clone_url = webserver.repo_url(GIT_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir())
         assert 'Cloning into' in stdout + stderr
         assert stderr == '' or stdout == ''
     def test_clone_wrong_credentials_hg(self, webserver):
         clone_url = webserver.repo_url(HG_REPO, password='bad!')
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'abort: authorization failed' in stderr
     def test_clone_wrong_credentials_git(self, webserver):
         clone_url = webserver.repo_url(GIT_REPO, password='bad!')
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'fatal: Authentication failed' in stderr
     def test_clone_git_dir_as_hg(self, webserver):
         clone_url = webserver.repo_url(GIT_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_clone_hg_repo_as_git(self, webserver):
         clone_url = webserver.repo_url(HG_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'not found' in stderr
     def test_clone_non_existing_path_hg(self, webserver):
         clone_url = webserver.repo_url('trololo')
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_clone_non_existing_path_git(self, webserver):
         clone_url = webserver.repo_url('trololo')
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         assert 'not found' in stderr
     def test_push_new_repo_git(self, webserver):
         # Clear the log so we know what is added
         UserLog.query().delete()
         Session().commit()
         # Create an empty server repo using the API
         repo_name = u'new_git_%s' % _RandomNameSequence().next()
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         params = {
             "id": 7,
             "api_key": usr.api_key,
             "method": 'create_repo',
             "args": dict(repo_name=repo_name,
                          owner=TEST_USER_ADMIN_LOGIN,
                          repo_type='git'),
+        }
         req = urllib2.Request(
             'http://%s:%s/_admin/api' % webserver.server_address,
             data=json.dumps(params),
             headers={'content-type': 'application/json'})
         response = urllib2.urlopen(req)
         result = json.loads(response.read())
         # Expect something like:
         # {u'result': {u'msg': u'Created new repository `new_git_XXX`', u'task': None, u'success': True}, u'id': 7, u'error': None}
         assert result[u'result'][u'success']
         # Create local clone of the empty server repo
         local_clone_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(repo_name)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, local_clone_dir, ignoreReturnCode=True)
         # Make 3 commits and push to the empty server repo.
         # The server repo doesn't have any other heads than the
         # refs/heads/master we are pushing, but the `git log` in the push hook
         # should still list the 3 commits.
         stdout, stderr = _add_files_and_push(webserver, 'git', local_clone_dir, clone_url=clone_url)
         # FIXME: the push kind of failed with something like:
         # remote: fatal: ambiguous argument '': unknown revision or path not in the working tree.
         assert 'remote: fatal' in stderr
         _check_proper_git_push(stdout, stderr)
         # Verify that we got the right events in UserLog. Expect something like:
         # <UserLog('id:new_git_XXX:started_following_repo')>
         # <UserLog('id:new_git_XXX:user_created_repo')>
         # <UserLog('id:new_git_XXX:pull')>
-        # - but no push logging
+        # <UserLog('id:new_git_XXX:push:aed9d4c1732a1927da3be42c47eb9afdc200d427,d38b083a07af10a9f44193486959a96a23db78da,4841ff9a2b385bec995f4679ef649adb3f437622')>
         uls = list(UserLog.query().order_by(UserLog.user_log_id))
-        assert len(uls) == 3
+        assert len(uls) == 4
         assert uls[0].action == 'started_following_repo'
         assert uls[1].action == 'user_created_repo'
         assert uls[2].action == 'pull'
         assert uls[3].action.startswith(u'push:')
         assert uls[3].action.count(',') == 2 # expect 3 commits
     def test_push_new_file_hg(self, webserver, testfork):
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(HG_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)
         clone_url = webserver.repo_url(testfork['hg'])
         stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, clone_url=clone_url)
         assert 'pushing to' in stdout
         assert 'Repository size' in stdout
         assert 'Last revision is now' in stdout
     def test_push_new_file_git(self, webserver, testfork):
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(GIT_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)
         # commit some stuff into this repo
         clone_url = webserver.repo_url(testfork['git'])
         stdout, stderr = _add_files_and_push(webserver, 'git', dest_dir, clone_url=clone_url)
         print [(x.repo_full_path,x.repo_path) for x in Repository.query()] # TODO: what is this for
         _check_proper_git_push(stdout, stderr)
     def test_push_invalidates_cache_hg(self, webserver, testfork):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                == HG_REPO).scalar()
         if not key:
             key = CacheInvalidation(HG_REPO, HG_REPO)
             Session().add(key)
         key.cache_active = True
         Session().commit()
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(testfork['hg'])
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, files_no=1, clone_url=clone_url)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                == testfork['hg']).all()
         assert key == []
     def test_push_invalidates_cache_git(self, webserver, testfork):
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                == GIT_REPO).scalar()
         if not key:
             key = CacheInvalidation(GIT_REPO, GIT_REPO)
             Session().add(key)
         key.cache_active = True
         Session().commit()
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(testfork['git'])
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)
         # commit some stuff into this repo
         stdout, stderr = _add_files_and_push(webserver, 'git', dest_dir, files_no=1, clone_url=clone_url)
         _check_proper_git_push(stdout, stderr)
         key = CacheInvalidation.query().filter(CacheInvalidation.cache_key
                                                == testfork['git']).all()
         assert key == []
     def test_push_wrong_credentials_hg(self, webserver):
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(HG_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, username='bad',
                                              password='name', ignoreReturnCode=True)
         assert 'abort: authorization failed' in stderr
     def test_push_wrong_credentials_git(self, webserver):
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(GIT_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, 'git', dest_dir, username='bad',
                                              password='name', ignoreReturnCode=True)
         assert 'fatal: Authentication failed' in stderr
     def test_push_with_readonly_credentials_hg(self, webserver):
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(HG_REPO, username=TEST_USER_REGULAR_LOGIN, password=TEST_USER_REGULAR_PASS)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, username=TEST_USER_REGULAR_LOGIN,
                                              password=TEST_USER_REGULAR_PASS, ignoreReturnCode=True)
         assert 'abort: HTTP Error 403: Forbidden' in stderr
     def test_push_with_readonly_credentials_git(self, webserver):
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(GIT_REPO, username=TEST_USER_REGULAR_LOGIN, password=TEST_USER_REGULAR_PASS)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, 'git', dest_dir, username=TEST_USER_REGULAR_LOGIN,
                                              password=TEST_USER_REGULAR_PASS, ignoreReturnCode=True)
         assert 'The requested URL returned error: 403' in stderr
     def test_push_back_to_wrong_url_hg(self, webserver):
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(HG_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(
             webserver, 'hg', dest_dir, clone_url='http://%s:%s/tmp' % (
                 webserver.server_address[0], webserver.server_address[1]),
             ignoreReturnCode=True)
         assert 'HTTP Error 404: Not Found' in stderr
     def test_push_back_to_wrong_url_git(self, webserver):
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(GIT_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(
             webserver, 'git', dest_dir, clone_url='http://%s:%s/tmp' % (
                 webserver.server_address[0], webserver.server_address[1]),
             ignoreReturnCode=True)
         assert 'not found' in stderr
     def test_clone_and_create_lock_hg(self, webserver):
         # enable locking
         r = Repository.get_by_repo_name(HG_REPO)
         r.enable_locking = True
         Session().commit()
         # clone
         clone_url = webserver.repo_url(HG_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir())
         # check if lock was made
         r = Repository.get_by_repo_name(HG_REPO)
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
     def test_clone_and_create_lock_git(self, webserver):
         # enable locking
         r = Repository.get_by_repo_name(GIT_REPO)
         r.enable_locking = True
         Session().commit()
         # clone
         clone_url = webserver.repo_url(GIT_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir())
         # check if lock was made
         r = Repository.get_by_repo_name(GIT_REPO)
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
     def test_clone_after_repo_was_locked_hg(self, webserver):
         # lock repo
         r = Repository.get_by_repo_name(HG_REPO)
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         # pull fails since repo is locked
         clone_url = webserver.repo_url(HG_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
     def test_clone_after_repo_was_locked_git(self, webserver):
         # lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         # pull fails since repo is locked
         clone_url = webserver.repo_url(GIT_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
         msg = ("""The requested URL returned error: 423""")
         assert msg in stderr
     def test_push_on_locked_repo_by_other_user_hg(self, webserver):
         # clone some temp
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(HG_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)
         # lock repo
         r = Repository.get_by_repo_name(HG_REPO)
         # let this user actually push !
         RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,
                                           perm='repository.write')
         Session().commit()
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         # push fails repo is locked by other user !
         stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir,
                                              username=TEST_USER_REGULAR_LOGIN,
                                              password=TEST_USER_REGULAR_PASS,
                                              ignoreReturnCode=True)
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (HG_REPO, TEST_USER_ADMIN_LOGIN))
         assert msg in stderr
     def test_push_on_locked_repo_by_other_user_git(self, webserver):
         # Note: Git hooks must be executable on unix. This test will thus fail
         # for example on Linux if /tmp is mounted noexec.
         # clone some temp
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(GIT_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)
         # lock repo
         r = Repository.get_by_repo_name(GIT_REPO)
         # let this user actually push !
         RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,
                                           perm='repository.write')
         Session().commit()
         Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)
         # push fails repo is locked by other user !
         stdout, stderr = _add_files_and_push(webserver, 'git', dest_dir,
                                              username=TEST_USER_REGULAR_LOGIN,
                                              password=TEST_USER_REGULAR_PASS,
                                              ignoreReturnCode=True)
         err = 'Repository `%s` locked by user `%s`' % (GIT_REPO, TEST_USER_ADMIN_LOGIN)
         assert err in stderr
         # TODO: fix this somehow later on Git, Git is stupid and even if we throw
         # back 423 to it, it makes ANOTHER request and we fail there with 405 :/
         msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""
                 % (GIT_REPO, TEST_USER_ADMIN_LOGIN))
         #msg = "405 Method Not Allowed"
         #assert msg in stderr
     def test_push_unlocks_repository_hg(self, webserver, testfork):
         # enable locking
         r = Repository.get_by_repo_name(testfork['hg'])
         r.enable_locking = True
         Session().commit()
         # clone some temp
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(testfork['hg'])
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, dest_dir)
         # check for lock repo after clone
         r = Repository.get_by_repo_name(testfork['hg'])
         uid = User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         assert r.locked[0] == uid
         # push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push(webserver, 'hg', dest_dir, clone_url=clone_url)
         assert str('remote: Released lock on repo `%s`' % testfork['hg']) in stdout
         # we need to cleanup the Session Here !
         Session.remove()
         r = Repository.get_by_repo_name(testfork['hg'])
         assert r.locked == [None, None]
     # TODO: fix me ! somehow during tests hooks don't get called on Git
     def test_push_unlocks_repository_git(self, webserver, testfork):
         # enable locking
         r = Repository.get_by_repo_name(testfork['git'])
         r.enable_locking = True
         Session().commit()
         # clone some temp
         dest_dir = _get_tmp_dir()
         clone_url = webserver.repo_url(testfork['git'])
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, dest_dir)
         # check for lock repo after clone
         r = Repository.get_by_repo_name(testfork['git'])
         assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id
         # push is ok and repo is now unlocked
         stdout, stderr = _add_files_and_push(webserver, 'git', dest_dir, clone_url=clone_url)
         _check_proper_git_push(stdout, stderr)
         assert ('remote: Released lock on repo `%s`' % testfork['git']) in stderr
         # we need to cleanup the Session Here !
         Session.remove()
         r = Repository.get_by_repo_name(testfork['git'])
         assert r.locked == [None, None]
     def test_ip_restriction_hg(self, webserver):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             # IP permissions are cached, need to wait for the cache in the server process to expire
             time.sleep(1.5)
             clone_url = webserver.repo_url(HG_REPO)
             stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
             assert 'abort: HTTP Error 403: Forbidden' in stderr
         finally:
             # release IP restrictions
             for ip in UserIpMap.query():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
             # IP permissions are cached, need to wait for the cache in the server process to expire
             time.sleep(1.5)
         clone_url = webserver.repo_url(HG_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir())
         assert 'requesting all changes' in stdout
         assert 'adding changesets' in stdout
         assert 'adding manifests' in stdout
         assert 'adding file changes' in stdout
         assert stderr == ''
     def test_ip_restriction_git(self, webserver):
         user_model = UserModel()
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             # IP permissions are cached, need to wait for the cache in the server process to expire
             time.sleep(1.5)
             clone_url = webserver.repo_url(GIT_REPO)
             stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)
             # The message apparently changed in Git 1.8.3, so match it loosely.
             assert re.search(r'\b403\b', stderr)
         finally:
             # release IP restrictions
             for ip in UserIpMap.query():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
             # IP permissions are cached, need to wait for the cache in the server process to expire
             time.sleep(1.5)
         clone_url = webserver.repo_url(GIT_REPO)
         stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir())
         assert 'Cloning into' in stdout + stderr
         assert stderr == '' or stdout == ''
     @parametrize('repo_type, repo_name', [
                 #('git',      GIT_REPO), # git hooks doesn't work like hg hooks
                 ('hg',       HG_REPO),
     ])
     def test_custom_hooks_preoutgoing(self, testhook_cleanup, webserver, testfork, repo_type, repo_name):
         # set prechangegroup to failing hook (returns True)
         Ui.create_or_update_hook('preoutgoing.testhook', 'python:kallithea.tests.fixture.failing_test_hook')
         Session().commit()
         # clone repo
         clone_url = webserver.repo_url(testfork[repo_type], username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS)
         dest_dir = _get_tmp_dir()
         stdout, stderr = Command(TESTS_TMP_PATH) \
             .execute('%s clone' % repo_type, clone_url, dest_dir, ignoreReturnCode=True)
         if repo_type == 'hg':
             assert 'preoutgoing.testhook hook failed' in stdout
         elif repo_type == 'git':
             assert 'error: 406' in stderr
     @parametrize('repo_type, repo_name', [
                 #('git',      GIT_REPO), # git hooks doesn't work like hg hooks
                 ('hg',       HG_REPO),
     ])
     def test_custom_hooks_prechangegroup(self, testhook_cleanup, webserver, testfork, repo_type, repo_name):
         # set prechangegroup to failing hook (returns True)
         Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.failing_test_hook')
         Session().commit()
         # clone repo
         clone_url = webserver.repo_url(testfork[repo_type], username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS)
         dest_dir = _get_tmp_dir()
         stdout, stderr = Command(TESTS_TMP_PATH).execute('%s clone' % repo_type, clone_url, dest_dir)
         stdout, stderr = _add_files_and_push(webserver, repo_type, dest_dir,
                                              username=TEST_USER_ADMIN_LOGIN,
                                              password=TEST_USER_ADMIN_PASS,
                                              ignoreReturnCode=True)
         assert 'failing_test_hook failed' in stdout + stderr
         assert 'Traceback' not in stdout + stderr
         assert 'prechangegroup.testhook hook failed' in stdout + stderr
         # there are still outgoing changesets
         stdout, stderr = _check_outgoing(repo_type, dest_dir, clone_url)
         assert stdout != ''
         # set prechangegroup hook to exception throwing method
         Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.exception_test_hook')
         Session().commit()
         # re-try to push
         stdout, stderr = Command(dest_dir).execute('%s push' % repo_type, clone_url, ignoreReturnCode=True)
         if repo_type == 'hg':
             # like with 'hg serve...' 'HTTP Error 500: INTERNAL SERVER ERROR' should be returned

0 comments (0 inline, 0 general)