test_vcs_operations.py
641 lines
| 28.9 KiB
| text/x-python
|
PythonLexer
domruf
|
r6574 | # -*- 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. | ||||
""" | ||||
Mads Kiilerich
|
r7718 | import json | ||
domruf
|
r6574 | import os | ||
import re | ||||
import tempfile | ||||
import time | ||||
Mads Kiilerich
|
r8068 | import urllib.request | ||
Mads Kiilerich
|
r7718 | from subprocess import PIPE, Popen | ||
domruf
|
r6574 | from tempfile import _RandomNameSequence | ||
Mads Kiilerich
|
r7272 | import pytest | ||
Mads Kiilerich
|
r8433 | import kallithea | ||
Mads Kiilerich
|
r8086 | from kallithea.lib.utils2 import ascii_bytes, safe_str | ||
Mads Kiilerich
|
r8453 | from kallithea.model import db, meta | ||
Mads Kiilerich
|
r7718 | from kallithea.model.ssh_key import SshKeyModel | ||
from kallithea.model.user import UserModel | ||||
Mads Kiilerich
|
r7966 | from kallithea.tests import base | ||
domruf
|
r6574 | from kallithea.tests.fixture import Fixture | ||
Mads Kiilerich
|
r7718 | |||
domruf
|
r6574 | |||
DEBUG = True | ||||
HOST = '127.0.0.1:4999' # test host | ||||
fixture = Fixture() | ||||
domruf
|
r7582 | # Parameterize different kinds of VCS testing - both the kind of VCS and the | ||
# access method (HTTP/SSH) | ||||
Mads Kiilerich
|
r7683 | # Mixin for using HTTP and SSH URLs | ||
domruf
|
r7582 | class HttpVcsTest(object): | ||
@staticmethod | ||||
def repo_url_param(webserver, repo_name, **kwargs): | ||||
return webserver.repo_url(repo_name, **kwargs) | ||||
Mads Kiilerich
|
r7683 | class SshVcsTest(object): | ||
public_keys = { | ||||
Mads Kiilerich
|
r8093 | base.TEST_USER_REGULAR_LOGIN: 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== kallithea@localhost', | ||
base.TEST_USER_ADMIN_LOGIN: 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUq== kallithea@localhost', | ||||
Mads Kiilerich
|
r7683 | } | ||
@classmethod | ||||
Mads Kiilerich
|
r7966 | def repo_url_param(cls, webserver, repo_name, username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS, client_ip=base.IP_ADDR): | ||
Mads Kiilerich
|
r8453 | user = db.User.get_by_username(username) | ||
Mads Kiilerich
|
r7683 | if user.ssh_keys: | ||
ssh_key = user.ssh_keys[0] | ||||
else: | ||||
sshkeymodel = SshKeyModel() | ||||
Mads Kiilerich
|
r8093 | ssh_key = sshkeymodel.create(user, 'test key', cls.public_keys[user.username]) | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
Mads Kiilerich
|
r7683 | |||
return cls._ssh_param(repo_name, user, ssh_key, client_ip) | ||||
Mads Kiilerich
|
r7639 | # Mixins for using Mercurial and Git | ||
domruf
|
r7582 | class HgVcsTest(object): | ||
repo_type = 'hg' | ||||
Mads Kiilerich
|
r7966 | repo_name = base.HG_REPO | ||
domruf
|
r7582 | |||
class GitVcsTest(object): | ||||
repo_type = 'git' | ||||
Mads Kiilerich
|
r7966 | repo_name = base.GIT_REPO | ||
domruf
|
r7582 | |||
Mads Kiilerich
|
r7639 | # Combine mixins to give the combinations we want to parameterize tests with | ||
domruf
|
r7582 | class HgHttpVcsTest(HgVcsTest, HttpVcsTest): | ||
pass | ||||
class GitHttpVcsTest(GitVcsTest, HttpVcsTest): | ||||
pass | ||||
Mads Kiilerich
|
r7683 | class HgSshVcsTest(HgVcsTest, SshVcsTest): | ||
@staticmethod | ||||
def _ssh_param(repo_name, user, ssh_key, client_ip): | ||||
# Specify a custom ssh command on the command line | ||||
return r"""--config ui.ssh="bash -c 'SSH_ORIGINAL_COMMAND=\"\$2\" SSH_CONNECTION=\"%s 1024 127.0.0.1 22\" kallithea-cli ssh-serve -c %s %s %s' --" ssh://someuser@somehost/%s""" % ( | ||||
client_ip, | ||||
Mads Kiilerich
|
r8433 | kallithea.CONFIG['__file__'], | ||
Mads Kiilerich
|
r7683 | user.user_id, | ||
ssh_key.user_ssh_key_id, | ||||
repo_name) | ||||
class GitSshVcsTest(GitVcsTest, SshVcsTest): | ||||
@staticmethod | ||||
def _ssh_param(repo_name, user, ssh_key, client_ip): | ||||
# Set a custom ssh command in the global environment | ||||
os.environ['GIT_SSH_COMMAND'] = r"""bash -c 'SSH_ORIGINAL_COMMAND="$2" SSH_CONNECTION="%s 1024 127.0.0.1 22" kallithea-cli ssh-serve -c %s %s %s' --""" % ( | ||||
client_ip, | ||||
Mads Kiilerich
|
r8433 | kallithea.CONFIG['__file__'], | ||
Mads Kiilerich
|
r7683 | user.user_id, | ||
ssh_key.user_ssh_key_id) | ||||
return "ssh://someuser@somehost/%s""" % repo_name | ||||
Mads Kiilerich
|
r7966 | parametrize_vcs_test = base.parametrize('vt', [ | ||
domruf
|
r7582 | HgHttpVcsTest, | ||
GitHttpVcsTest, | ||||
Mads Kiilerich
|
r7683 | HgSshVcsTest, | ||
GitSshVcsTest, | ||||
domruf
|
r7582 | ]) | ||
Mads Kiilerich
|
r7966 | parametrize_vcs_test_hg = base.parametrize('vt', [ | ||
domruf
|
r7582 | HgHttpVcsTest, | ||
Mads Kiilerich
|
r7683 | HgSshVcsTest, | ||
domruf
|
r7582 | ]) | ||
Mads Kiilerich
|
r7966 | parametrize_vcs_test_http = base.parametrize('vt', [ | ||
domruf
|
r7582 | HgHttpVcsTest, | ||
GitHttpVcsTest, | ||||
]) | ||||
domruf
|
r6574 | class Command(object): | ||
def __init__(self, cwd): | ||||
self.cwd = cwd | ||||
domruf
|
r7582 | def execute(self, *args, **environ): | ||
domruf
|
r6574 | """ | ||
domruf
|
r7582 | Runs command on the system with given ``args`` using simple space | ||
join without safe quoting. | ||||
domruf
|
r6574 | """ | ||
domruf
|
r7582 | command = ' '.join(args) | ||
domruf
|
r6574 | ignoreReturnCode = environ.pop('ignoreReturnCode', False) | ||
if DEBUG: | ||||
Mads Kiilerich
|
r7750 | print('*** CMD %s ***' % command) | ||
domruf
|
r6574 | testenv = dict(os.environ) | ||
testenv['LANG'] = 'en_US.UTF-8' | ||||
testenv['LANGUAGE'] = 'en_US:en' | ||||
Thomas De Schampheleire
|
r6695 | testenv['HGPLAIN'] = '' | ||
testenv['HGRCPATH'] = '' | ||||
domruf
|
r6574 | 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: | ||||
Mads Kiilerich
|
r7750 | print('stdout:', stdout) | ||
domruf
|
r6574 | if stderr: | ||
Mads Kiilerich
|
r7750 | print('stderr:', stderr) | ||
domruf
|
r6574 | if not ignoreReturnCode: | ||
assert p.returncode == 0 | ||||
Mads Kiilerich
|
r8086 | return safe_str(stdout), safe_str(stderr) | ||
domruf
|
r6574 | |||
Mads Kiilerich
|
r6598 | def _get_tmp_dir(prefix='vcs_operations-', suffix=''): | ||
Mads Kiilerich
|
r7966 | return tempfile.mkdtemp(dir=base.TESTS_TMP_PATH, prefix=prefix, suffix=suffix) | ||
domruf
|
r6574 | |||
domruf
|
r7107 | def _add_files(vcs, dest_dir, files_no=3): | ||
domruf
|
r6574 | """ | ||
Lars Kruse
|
r6790 | Generate some files, add it to dest_dir repo and push back | ||
domruf
|
r6574 | vcs is git or hg and defines what VCS we want to make those files for | ||
:param vcs: | ||||
Lars Kruse
|
r6790 | :param dest_dir: | ||
domruf
|
r6574 | """ | ||
Mads Kiilerich
|
r8057 | added_file = '%ssetup.py' % next(_RandomNameSequence()) | ||
Mads Kiilerich
|
r7106 | open(os.path.join(dest_dir, added_file), 'a').close() | ||
domruf
|
r7582 | Command(dest_dir).execute(vcs, 'add', added_file) | ||
domruf
|
r6574 | |||
email = 'me@example.com' | ||||
if os.name == 'nt': | ||||
author_str = 'User <%s>' % email | ||||
else: | ||||
author_str = 'User ǝɯɐᴎ <%s>' % email | ||||
Mads Kiilerich
|
r8061 | for i in range(files_no): | ||
domruf
|
r6574 | cmd = """echo "added_line%s" >> %s""" % (i, added_file) | ||
Mads Kiilerich
|
r7106 | Command(dest_dir).execute(cmd) | ||
domruf
|
r6574 | 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 | ||||
Mads Kiilerich
|
r7106 | Command(dest_dir).execute(cmd, EMAIL=email) | ||
domruf
|
r6574 | |||
Mads Kiilerich
|
r7584 | def _add_files_and_push(webserver, vt, dest_dir, clone_url, ignoreReturnCode=False, files_no=3): | ||
_add_files(vt.repo_type, dest_dir, files_no=files_no) | ||||
domruf
|
r6574 | # PUSH it back | ||
stdout = stderr = None | ||||
Mads Kiilerich
|
r7584 | if vt.repo_type == 'hg': | ||
Mads Kiilerich
|
r7660 | stdout, stderr = Command(dest_dir).execute('hg push -f --verbose', clone_url, ignoreReturnCode=ignoreReturnCode) | ||
Mads Kiilerich
|
r7584 | elif vt.repo_type == 'git': | ||
Mads Kiilerich
|
r7660 | stdout, stderr = Command(dest_dir).execute('git push -f --verbose', clone_url, "master", ignoreReturnCode=ignoreReturnCode) | ||
domruf
|
r6574 | |||
return stdout, stderr | ||||
Mads Kiilerich
|
r7583 | def _check_outgoing(vcs, cwd, clone_url): | ||
domruf
|
r7112 | 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') | ||||
domruf
|
r6574 | def set_anonymous_access(enable=True): | ||
Mads Kiilerich
|
r8453 | user = db.User.get_default_user() | ||
domruf
|
r6574 | user.active = enable | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
Mads Kiilerich
|
r8453 | if enable != db.User.get_default_user().active: | ||
domruf
|
r6574 | raise Exception('Cannot set anonymous access') | ||
#============================================================================== | ||||
# TESTS | ||||
#============================================================================== | ||||
def _check_proper_git_push(stdout, stderr): | ||||
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") | ||||
Mads Kiilerich
|
r7966 | class TestVCSOperations(base.TestController): | ||
domruf
|
r6574 | |||
@classmethod | ||||
def setup_class(cls): | ||||
Lars Kruse
|
r6789 | # DISABLE ANONYMOUS ACCESS | ||
domruf
|
r6574 | set_anonymous_access(False) | ||
domruf
|
r7112 | @pytest.fixture() | ||
def testhook_cleanup(self): | ||||
yield | ||||
# remove hook | ||||
for hook in ['prechangegroup', 'pretxnchangegroup', 'preoutgoing', 'changegroup', 'outgoing', 'incoming']: | ||||
Mads Kiilerich
|
r8453 | entry = db.Ui.get_by_key('hooks', '%s.testhook' % hook) | ||
domruf
|
r7112 | if entry: | ||
Mads Kiilerich
|
r8452 | meta.Session().delete(entry) | ||
meta.Session().commit() | ||||
domruf
|
r7112 | |||
@pytest.fixture(scope="module") | ||||
def testfork(self): | ||||
# create fork so the repo stays untouched | ||||
Mads Kiilerich
|
r8093 | git_fork_name = '%s_fork%s' % (base.GIT_REPO, next(_RandomNameSequence())) | ||
Mads Kiilerich
|
r7966 | fixture.create_fork(base.GIT_REPO, git_fork_name) | ||
Mads Kiilerich
|
r8093 | hg_fork_name = '%s_fork%s' % (base.HG_REPO, next(_RandomNameSequence())) | ||
Mads Kiilerich
|
r7966 | fixture.create_fork(base.HG_REPO, hg_fork_name) | ||
domruf
|
r7112 | return {'git': git_fork_name, 'hg': hg_fork_name} | ||
domruf
|
r7582 | @parametrize_vcs_test | ||
def test_clone_repo_by_admin(self, webserver, vt): | ||||
Mads Kiilerich
|
r7585 | clone_url = vt.repo_url_param(webserver, vt.repo_name) | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir()) | ||
domruf
|
r6574 | |||
domruf
|
r7582 | if vt.repo_type == 'git': | ||
assert 'Cloning into' in stdout + stderr | ||||
assert stderr == '' or stdout == '' | ||||
elif vt.repo_type == 'hg': | ||||
assert 'requesting all changes' in stdout | ||||
assert 'adding changesets' in stdout | ||||
assert 'adding manifests' in stdout | ||||
assert 'adding file changes' in stdout | ||||
assert stderr == '' | ||||
domruf
|
r6574 | |||
domruf
|
r7582 | @parametrize_vcs_test_http | ||
def test_clone_wrong_credentials(self, webserver, vt): | ||||
Mads Kiilerich
|
r7585 | clone_url = vt.repo_url_param(webserver, vt.repo_name, password='bad!') | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) | ||
domruf
|
r7582 | if vt.repo_type == 'git': | ||
assert 'fatal: Authentication failed' in stderr | ||||
elif vt.repo_type == 'hg': | ||||
assert 'abort: authorization failed' in stderr | ||||
domruf
|
r6574 | |||
def test_clone_git_dir_as_hg(self, webserver): | ||||
Mads Kiilerich
|
r7966 | clone_url = HgHttpVcsTest.repo_url_param(webserver, base.GIT_REPO) | ||
stdout, stderr = Command(base.TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) | ||||
Mads Kiilerich
|
r7683 | assert 'HTTP Error 404: Not Found' in stderr or "not a valid repository" in stdout and 'abort:' in stderr | ||
domruf
|
r6574 | |||
def test_clone_hg_repo_as_git(self, webserver): | ||||
Mads Kiilerich
|
r7966 | clone_url = GitHttpVcsTest.repo_url_param(webserver, base.HG_REPO) | ||
stdout, stderr = Command(base.TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) | ||||
domruf
|
r6574 | assert 'not found' in stderr | ||
domruf
|
r7582 | @parametrize_vcs_test | ||
def test_clone_non_existing_path(self, webserver, vt): | ||||
Mads Kiilerich
|
r7585 | clone_url = vt.repo_url_param(webserver, 'trololo') | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) | ||
domruf
|
r7582 | if vt.repo_type == 'git': | ||
Mads Kiilerich
|
r7683 | assert 'not found' in stderr or 'abort: Access to %r denied' % 'trololo' in stderr | ||
domruf
|
r7582 | elif vt.repo_type == 'hg': | ||
Mads Kiilerich
|
r8348 | assert 'HTTP Error 404: Not Found' in stderr or 'abort: no suitable response from remote hg' in stderr and 'remote: abort: Access to %r denied' % 'trololo' in stdout + stderr | ||
domruf
|
r6574 | |||
Mads Kiilerich
|
r7586 | @parametrize_vcs_test | ||
def test_push_new_repo(self, webserver, vt): | ||||
Mads Kiilerich
|
r7272 | # Clear the log so we know what is added | ||
Mads Kiilerich
|
r8453 | db.UserLog.query().delete() | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
Mads Kiilerich
|
r7272 | |||
# Create an empty server repo using the API | ||||
Mads Kiilerich
|
r8093 | repo_name = 'new_%s_%s' % (vt.repo_type, next(_RandomNameSequence())) | ||
Mads Kiilerich
|
r8453 | usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN) | ||
Mads Kiilerich
|
r7272 | params = { | ||
"id": 7, | ||||
"api_key": usr.api_key, | ||||
"method": 'create_repo', | ||||
"args": dict(repo_name=repo_name, | ||||
Mads Kiilerich
|
r7966 | owner=base.TEST_USER_ADMIN_LOGIN, | ||
Mads Kiilerich
|
r7586 | repo_type=vt.repo_type), | ||
Mads Kiilerich
|
r7272 | } | ||
Mads Kiilerich
|
r8068 | req = urllib.request.Request( | ||
Mads Kiilerich
|
r7272 | 'http://%s:%s/_admin/api' % webserver.server_address, | ||
Mads Kiilerich
|
r7960 | data=ascii_bytes(json.dumps(params)), | ||
Mads Kiilerich
|
r7272 | headers={'content-type': 'application/json'}) | ||
Mads Kiilerich
|
r8068 | response = urllib.request.urlopen(req) | ||
Mads Kiilerich
|
r7272 | result = json.loads(response.read()) | ||
# Expect something like: | ||||
Mads Kiilerich
|
r7586 | # {u'result': {u'msg': u'Created new repository `new_XXX`', u'task': None, u'success': True}, u'id': 7, u'error': None} | ||
Mads Kiilerich
|
r8093 | assert result['result']['success'] | ||
Mads Kiilerich
|
r7272 | |||
# Create local clone of the empty server repo | ||||
local_clone_dir = _get_tmp_dir() | ||||
Mads Kiilerich
|
r7586 | clone_url = vt.repo_url_param(webserver, repo_name) | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, local_clone_dir) | ||
Mads Kiilerich
|
r7272 | |||
# 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. | ||||
Mads Kiilerich
|
r7586 | stdout, stderr = _add_files_and_push(webserver, vt, local_clone_dir, clone_url=clone_url) | ||
if vt.repo_type == 'git': | ||||
_check_proper_git_push(stdout, stderr) | ||||
elif vt.repo_type == 'hg': | ||||
assert 'pushing to ' in stdout | ||||
assert 'remote: added ' in stdout | ||||
Mads Kiilerich
|
r7272 | |||
# 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')> | ||||
Mads Kiilerich
|
r7285 | # <UserLog('id:new_git_XXX:push:aed9d4c1732a1927da3be42c47eb9afdc200d427,d38b083a07af10a9f44193486959a96a23db78da,4841ff9a2b385bec995f4679ef649adb3f437622')> | ||
Mads Kiilerich
|
r8452 | meta.Session.close() # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...) | ||
Mads Kiilerich
|
r8453 | action_parts = [ul.action.split(':', 1) for ul in db.UserLog.query().order_by(db.UserLog.user_log_id)] | ||
Mads Kiilerich
|
r7586 | assert [(t[0], (t[1].count(',') + 1) if len(t) == 2 else 0) for t in action_parts] == ([ | ||
Mads Kiilerich
|
r8093 | ('started_following_repo', 0), | ||
('user_created_repo', 0), | ||||
('pull', 0), | ||||
('push', 3)] | ||||
Mads Kiilerich
|
r7586 | if vt.repo_type == 'git' else [ | ||
Mads Kiilerich
|
r8093 | ('started_following_repo', 0), | ||
('user_created_repo', 0), | ||||
Mads Kiilerich
|
r7586 | # (u'pull', 0), # Mercurial outgoing hook is not called for empty clones | ||
Mads Kiilerich
|
r8093 | ('push', 3)]) | ||
Mads Kiilerich
|
r7272 | |||
domruf
|
r7582 | @parametrize_vcs_test | ||
def test_push_new_file(self, webserver, testfork, vt): | ||||
Mads Kiilerich
|
r8453 | db.UserLog.query().delete() | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
Mads Kiilerich
|
r7587 | |||
Lars Kruse
|
r6790 | dest_dir = _get_tmp_dir() | ||
Mads Kiilerich
|
r7585 | clone_url = vt.repo_url_param(webserver, vt.repo_name) | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) | ||
domruf
|
r6574 | |||
Mads Kiilerich
|
r7585 | clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type]) | ||
Mads Kiilerich
|
r7584 | stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, clone_url=clone_url) | ||
domruf
|
r6574 | |||
domruf
|
r7582 | if vt.repo_type == 'git': | ||
_check_proper_git_push(stdout, stderr) | ||||
elif vt.repo_type == 'hg': | ||||
assert 'pushing to' in stdout | ||||
assert 'Repository size' in stdout | ||||
assert 'Last revision is now' in stdout | ||||
domruf
|
r6574 | |||
Mads Kiilerich
|
r8452 | meta.Session.close() # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...) | ||
Mads Kiilerich
|
r8453 | action_parts = [ul.action.split(':', 1) for ul in db.UserLog.query().order_by(db.UserLog.user_log_id)] | ||
Mads Kiilerich
|
r7587 | assert [(t[0], (t[1].count(',') + 1) if len(t) == 2 else 0) for t in action_parts] == \ | ||
Mads Kiilerich
|
r8093 | [('pull', 0), ('push', 3)] | ||
Mads Kiilerich
|
r7587 | |||
domruf
|
r7582 | @parametrize_vcs_test | ||
Mads Kiilerich
|
r7650 | def test_pull(self, webserver, testfork, vt): | ||
Mads Kiilerich
|
r8453 | db.UserLog.query().delete() | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
Mads Kiilerich
|
r7650 | |||
dest_dir = _get_tmp_dir() | ||||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'init', dest_dir) | ||
Mads Kiilerich
|
r7650 | |||
clone_url = vt.repo_url_param(webserver, vt.repo_name) | ||||
stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url) | ||||
Mads Kiilerich
|
r8452 | meta.Session.close() # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...) | ||
Mads Kiilerich
|
r7650 | |||
if vt.repo_type == 'git': | ||||
assert 'FETCH_HEAD' in stderr | ||||
elif vt.repo_type == 'hg': | ||||
assert 'new changesets' in stdout | ||||
Mads Kiilerich
|
r8453 | action_parts = [ul.action for ul in db.UserLog.query().order_by(db.UserLog.user_log_id)] | ||
Mads Kiilerich
|
r8093 | assert action_parts == ['pull'] | ||
Mads Kiilerich
|
r7650 | |||
Mads Kiilerich
|
r8026 | # Test handling of URLs with extra '/' around repo_name | ||
stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url.replace('/' + vt.repo_name, '/./%s/' % vt.repo_name), ignoreReturnCode=True) | ||||
if issubclass(vt, HttpVcsTest): | ||||
if vt.repo_type == 'git': | ||||
# NOTE: when pulling from http://hostname/./vcs_test_git/ , the git client will normalize that and issue an HTTP request to /vcs_test_git/info/refs | ||||
assert 'Already up to date.' in stdout | ||||
else: | ||||
assert vt.repo_type == 'hg' | ||||
assert "abort: HTTP Error 404: Not Found" in stderr | ||||
else: | ||||
assert issubclass(vt, SshVcsTest) | ||||
if vt.repo_type == 'git': | ||||
Mads Kiilerich
|
r8028 | assert "abort: Access to './%s' denied" % vt.repo_name in stderr | ||
Mads Kiilerich
|
r8026 | else: | ||
Mads Kiilerich
|
r8348 | assert "abort: Access to './%s' denied" % vt.repo_name in stdout + stderr | ||
Mads Kiilerich
|
r8026 | |||
stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url.replace('/' + vt.repo_name, '/%s/' % vt.repo_name), ignoreReturnCode=True) | ||||
Mads Kiilerich
|
r8028 | if vt.repo_type == 'git': | ||
assert 'Already up to date.' in stdout | ||||
Mads Kiilerich
|
r8026 | else: | ||
Mads Kiilerich
|
r8028 | assert vt.repo_type == 'hg' | ||
assert "no changes found" in stdout | ||||
assert "denied" not in stderr | ||||
assert "denied" not in stdout | ||||
assert "404" not in stdout | ||||
Mads Kiilerich
|
r8026 | |||
Mads Kiilerich
|
r7650 | @parametrize_vcs_test | ||
domruf
|
r7582 | def test_push_invalidates_cache(self, webserver, testfork, vt): | ||
Mads Kiilerich
|
r8453 | pre_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in db.Repository.query().filter(db.Repository.repo_name == testfork[vt.repo_type])] | ||
Mads Kiilerich
|
r7589 | |||
Lars Kruse
|
r6790 | dest_dir = _get_tmp_dir() | ||
Mads Kiilerich
|
r7585 | clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type]) | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) | ||
domruf
|
r7582 | |||
Mads Kiilerich
|
r7584 | stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, files_no=1, clone_url=clone_url) | ||
domruf
|
r6574 | |||
domruf
|
r7582 | if vt.repo_type == 'git': | ||
_check_proper_git_push(stdout, stderr) | ||||
Mads Kiilerich
|
r7589 | |||
Mads Kiilerich
|
r8452 | meta.Session.close() # expire session to make sure SA fetches new Repository instances after last_changeset has been updated by server side hook in another process | ||
Mads Kiilerich
|
r8453 | post_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in db.Repository.query().filter(db.Repository.repo_name == testfork[vt.repo_type])] | ||
Mads Kiilerich
|
r7589 | assert pre_cached_tip != post_cached_tip | ||
domruf
|
r7582 | @parametrize_vcs_test_http | ||
def test_push_wrong_credentials(self, webserver, vt): | ||||
Lars Kruse
|
r6790 | dest_dir = _get_tmp_dir() | ||
Mads Kiilerich
|
r7585 | clone_url = vt.repo_url_param(webserver, vt.repo_name) | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) | ||
domruf
|
r6574 | |||
Mads Kiilerich
|
r7583 | clone_url = webserver.repo_url(vt.repo_name, username='bad', password='name') | ||
Mads Kiilerich
|
r7584 | stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, | ||
Mads Kiilerich
|
r7583 | clone_url=clone_url, ignoreReturnCode=True) | ||
domruf
|
r6574 | |||
domruf
|
r7582 | if vt.repo_type == 'git': | ||
assert 'fatal: Authentication failed' in stderr | ||||
elif vt.repo_type == 'hg': | ||||
assert 'abort: authorization failed' in stderr | ||||
domruf
|
r6574 | |||
domruf
|
r7582 | @parametrize_vcs_test | ||
def test_push_with_readonly_credentials(self, webserver, vt): | ||||
Mads Kiilerich
|
r8453 | db.UserLog.query().delete() | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
Mads Kiilerich
|
r7587 | |||
domruf
|
r7103 | dest_dir = _get_tmp_dir() | ||
Mads Kiilerich
|
r7966 | clone_url = vt.repo_url_param(webserver, vt.repo_name, username=base.TEST_USER_REGULAR_LOGIN, password=base.TEST_USER_REGULAR_PASS) | ||
stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) | ||||
domruf
|
r7103 | |||
Mads Kiilerich
|
r7584 | stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, ignoreReturnCode=True, clone_url=clone_url) | ||
domruf
|
r7103 | |||
domruf
|
r7582 | if vt.repo_type == 'git': | ||
Mads Kiilerich
|
r7683 | assert 'The requested URL returned error: 403' in stderr or 'abort: Push access to %r denied' % str(vt.repo_name) in stderr | ||
domruf
|
r7582 | elif vt.repo_type == 'hg': | ||
Mads Kiilerich
|
r7683 | assert 'abort: HTTP Error 403: Forbidden' in stderr or 'abort: push failed on remote' in stderr and 'remote: Push access to %r denied' % str(vt.repo_name) in stdout | ||
domruf
|
r7103 | |||
Mads Kiilerich
|
r8452 | meta.Session.close() # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...) | ||
Mads Kiilerich
|
r8453 | action_parts = [ul.action.split(':', 1) for ul in db.UserLog.query().order_by(db.UserLog.user_log_id)] | ||
Mads Kiilerich
|
r7587 | assert [(t[0], (t[1].count(',') + 1) if len(t) == 2 else 0) for t in action_parts] == \ | ||
Mads Kiilerich
|
r8093 | [('pull', 0)] | ||
Mads Kiilerich
|
r7587 | |||
domruf
|
r7582 | @parametrize_vcs_test | ||
def test_push_back_to_wrong_url(self, webserver, vt): | ||||
Lars Kruse
|
r6790 | dest_dir = _get_tmp_dir() | ||
Mads Kiilerich
|
r7585 | clone_url = vt.repo_url_param(webserver, vt.repo_name) | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) | ||
domruf
|
r6574 | |||
Lars Kruse
|
r6790 | stdout, stderr = _add_files_and_push( | ||
Mads Kiilerich
|
r7584 | webserver, vt, dest_dir, clone_url='http://%s:%s/tmp' % ( | ||
Lars Kruse
|
r6790 | webserver.server_address[0], webserver.server_address[1]), | ||
ignoreReturnCode=True) | ||||
domruf
|
r6574 | |||
domruf
|
r7582 | if vt.repo_type == 'git': | ||
assert 'not found' in stderr | ||||
elif vt.repo_type == 'hg': | ||||
assert 'HTTP Error 404: Not Found' in stderr | ||||
domruf
|
r6574 | |||
domruf
|
r7582 | @parametrize_vcs_test | ||
def test_ip_restriction(self, webserver, vt): | ||||
domruf
|
r6574 | user_model = UserModel() | ||
try: | ||||
domruf
|
r7582 | # Add IP constraint that excludes the test context: | ||
Mads Kiilerich
|
r7966 | user_model.add_extra_ip(base.TEST_USER_ADMIN_LOGIN, '10.10.10.10/32') | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
Thomas De Schampheleire
|
r7206 | # IP permissions are cached, need to wait for the cache in the server process to expire | ||
time.sleep(1.5) | ||||
Mads Kiilerich
|
r7585 | clone_url = vt.repo_url_param(webserver, vt.repo_name) | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True) | ||
domruf
|
r7582 | if vt.repo_type == 'git': | ||
# The message apparently changed in Git 1.8.3, so match it loosely. | ||||
Mads Kiilerich
|
r7683 | assert re.search(r'\b403\b', stderr) or 'abort: User test_admin from 127.0.0.127 cannot be authorized' in stderr | ||
domruf
|
r7582 | elif vt.repo_type == 'hg': | ||
Mads Kiilerich
|
r8348 | assert 'abort: HTTP Error 403: Forbidden' in stderr or 'remote: abort: User test_admin from 127.0.0.127 cannot be authorized' in stdout + stderr | ||
domruf
|
r6574 | finally: | ||
Lars Kruse
|
r6789 | # release IP restrictions | ||
Mads Kiilerich
|
r8453 | for ip in db.UserIpMap.query(): | ||
db.UserIpMap.delete(ip.ip_id) | ||||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
Thomas De Schampheleire
|
r7206 | # IP permissions are cached, need to wait for the cache in the server process to expire | ||
time.sleep(1.5) | ||||
domruf
|
r6574 | |||
Mads Kiilerich
|
r7585 | clone_url = vt.repo_url_param(webserver, vt.repo_name) | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir()) | ||
domruf
|
r6574 | |||
domruf
|
r7582 | if vt.repo_type == 'git': | ||
assert 'Cloning into' in stdout + stderr | ||||
assert stderr == '' or stdout == '' | ||||
elif vt.repo_type == 'hg': | ||||
assert 'requesting all changes' in stdout | ||||
assert 'adding changesets' in stdout | ||||
assert 'adding manifests' in stdout | ||||
assert 'adding file changes' in stdout | ||||
domruf
|
r6574 | |||
domruf
|
r7582 | assert stderr == '' | ||
domruf
|
r7112 | |||
domruf
|
r7582 | @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks | ||
def test_custom_hooks_preoutgoing(self, testhook_cleanup, webserver, testfork, vt): | ||||
domruf
|
r7112 | # set prechangegroup to failing hook (returns True) | ||
Mads Kiilerich
|
r8453 | db.Ui.create_or_update_hook('preoutgoing.testhook', 'python:kallithea.tests.fixture.failing_test_hook') | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
domruf
|
r7112 | # clone repo | ||
Mads Kiilerich
|
r7966 | clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type], username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS) | ||
domruf
|
r7112 | dest_dir = _get_tmp_dir() | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH) \ | ||
domruf
|
r7582 | .execute(vt.repo_type, 'clone', clone_url, dest_dir, ignoreReturnCode=True) | ||
if vt.repo_type == 'hg': | ||||
Mads Kiilerich
|
r8654 | assert 'preoutgoing.testhook hook failed' in stdout + stderr | ||
domruf
|
r7582 | elif vt.repo_type == 'git': | ||
domruf
|
r7112 | assert 'error: 406' in stderr | ||
domruf
|
r7582 | @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks | ||
def test_custom_hooks_prechangegroup(self, testhook_cleanup, webserver, testfork, vt): | ||||
Mads Kiilerich
|
r7649 | # set prechangegroup to failing hook (returns exit code 1) | ||
Mads Kiilerich
|
r8453 | db.Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.failing_test_hook') | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
domruf
|
r7112 | # clone repo | ||
Mads Kiilerich
|
r7966 | clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type], username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS) | ||
domruf
|
r7112 | dest_dir = _get_tmp_dir() | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir) | ||
domruf
|
r7112 | |||
Mads Kiilerich
|
r7584 | stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, clone_url, | ||
domruf
|
r7112 | 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 | ||||
domruf
|
r7582 | stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url) | ||
domruf
|
r7112 | assert stdout != '' | ||
# set prechangegroup hook to exception throwing method | ||||
Mads Kiilerich
|
r8453 | db.Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.exception_test_hook') | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
domruf
|
r7112 | # re-try to push | ||
domruf
|
r7582 | stdout, stderr = Command(dest_dir).execute('%s push' % vt.repo_type, clone_url, ignoreReturnCode=True) | ||
Mads Kiilerich
|
r7649 | if vt is HgHttpVcsTest: | ||
domruf
|
r7112 | # like with 'hg serve...' 'HTTP Error 500: INTERNAL SERVER ERROR' should be returned | ||
assert 'HTTP Error 500: INTERNAL SERVER ERROR' in stderr | ||||
Mads Kiilerich
|
r7683 | elif vt is HgSshVcsTest: | ||
assert 'remote: Exception: exception_test_hook threw an exception' in stdout | ||||
Mads Kiilerich
|
r7649 | else: assert False | ||
domruf
|
r7112 | # there are still outgoing changesets | ||
domruf
|
r7582 | stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url) | ||
domruf
|
r7112 | assert stdout != '' | ||
# set prechangegroup hook to method that returns False | ||||
Mads Kiilerich
|
r8453 | db.Ui.create_or_update_hook('prechangegroup.testhook', 'python:kallithea.tests.fixture.passing_test_hook') | ||
Mads Kiilerich
|
r8452 | meta.Session().commit() | ||
domruf
|
r7112 | # re-try to push | ||
domruf
|
r7582 | stdout, stderr = Command(dest_dir).execute('%s push' % vt.repo_type, clone_url, ignoreReturnCode=True) | ||
domruf
|
r7112 | assert 'passing_test_hook succeeded' in stdout + stderr | ||
assert 'Traceback' not in stdout + stderr | ||||
assert 'prechangegroup.testhook hook failed' not in stdout + stderr | ||||
# no more outgoing changesets | ||||
domruf
|
r7582 | stdout, stderr = _check_outgoing(vt.repo_type, dest_dir, clone_url) | ||
domruf
|
r7112 | assert stdout == '' | ||
assert stderr == '' | ||||
domruf
|
r7122 | |||
domruf
|
r7582 | def test_add_submodule_git(self, webserver, testfork): | ||
domruf
|
r7122 | dest_dir = _get_tmp_dir() | ||
Mads Kiilerich
|
r7966 | clone_url = GitHttpVcsTest.repo_url_param(webserver, base.GIT_REPO) | ||
domruf
|
r7122 | |||
Mads Kiilerich
|
r7585 | fork_url = GitHttpVcsTest.repo_url_param(webserver, testfork['git']) | ||
domruf
|
r7122 | |||
# add submodule | ||||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(base.TESTS_TMP_PATH).execute('git clone', fork_url, dest_dir) | ||
domruf
|
r7122 | stdout, stderr = Command(dest_dir).execute('git submodule add', clone_url, 'testsubmodule') | ||
Mads Kiilerich
|
r7966 | stdout, stderr = Command(dest_dir).execute('git commit -am "added testsubmodule pointing to', clone_url, '"', EMAIL=base.TEST_USER_ADMIN_EMAIL) | ||
domruf
|
r7122 | stdout, stderr = Command(dest_dir).execute('git push', fork_url, 'master') | ||
# check for testsubmodule link in files page | ||||
self.log_user() | ||||
Mads Kiilerich
|
r7966 | response = self.app.get(base.url(controller='files', action='index', | ||
domruf
|
r7193 | repo_name=testfork['git'], | ||
domruf
|
r7122 | revision='tip', | ||
f_path='/')) | ||||
Mads Kiilerich
|
r7654 | # check _repo_files_url that will be used to reload as AJAX | ||
response.mustcontain('var _repo_files_url = ("/%s/files/");' % testfork['git']) | ||||
domruf
|
r7128 | response.mustcontain('<a class="submodule-dir" href="%s" target="_blank"><i class="icon-file-submodule"></i><span>testsubmodule @ ' % clone_url) | ||
Mads Kiilerich
|
r7123 | |||
# check that following a submodule link actually works - and redirects | ||||
Mads Kiilerich
|
r7966 | response = self.app.get(base.url(controller='files', action='index', | ||
domruf
|
r7193 | repo_name=testfork['git'], | ||
Mads Kiilerich
|
r7123 | revision='tip', | ||
f_path='/testsubmodule'), | ||||
domruf
|
r7127 | status=302) | ||
assert response.location == clone_url | ||||