diff --git a/MANIFEST.in b/MANIFEST.in --- a/MANIFEST.in +++ b/MANIFEST.in @@ -8,6 +8,9 @@ include vcsserver/VERSION # all config files recursive-include configs * +# hook templates +recursive-include vcsserver/hook_utils/hook_templates * + # skip any tests files recursive-exclude vcsserver/tests * diff --git a/configs/development_http.ini b/configs/development_http.ini --- a/configs/development_http.ini +++ b/configs/development_http.ini @@ -20,6 +20,10 @@ beaker.cache.repo_object.max_items = 100 beaker.cache.repo_object.expire = 300 beaker.cache.repo_object.enabled = true +# path to binaries for vcsserver, it should be set by the installer +# at installation time, e.g /home/user/vcsserver-1/profile/bin +core.binary_dir = "" + [server:main] ## COMMON ## host = 0.0.0.0 diff --git a/configs/production_http.ini b/configs/production_http.ini --- a/configs/production_http.ini +++ b/configs/production_http.ini @@ -39,7 +39,7 @@ use = egg:rhodecode-vcsserver pyramid.default_locale_name = en pyramid.includes = -## default locale used by VCS systems +# default locale used by VCS systems locale = en_US.UTF-8 # cache regions, please don't change @@ -50,6 +50,10 @@ beaker.cache.repo_object.max_items = 100 beaker.cache.repo_object.expire = 300 beaker.cache.repo_object.enabled = true +# path to binaries for vcsserver, it should be set by the installer +# at installation time, e.g /home/user/vcsserver-1/profile/bin +core.binary_dir = "" + ################################ ### LOGGING CONFIGURATION #### diff --git a/vcsserver/git.py b/vcsserver/git.py --- a/vcsserver/git.py +++ b/vcsserver/git.py @@ -655,6 +655,12 @@ class GitRemote(object): else: raise exceptions.VcsException(tb_err) + @reraise_safe_exceptions + def install_hooks(self, wire, force=False): + from vcsserver.hook_utils import install_git_hooks + repo = self._factory.repo(wire) + return install_git_hooks(repo.path, repo.bare, force_create=force) + def str_to_dulwich(value): """ diff --git a/vcsserver/hg.py b/vcsserver/hg.py --- a/vcsserver/hg.py +++ b/vcsserver/hg.py @@ -769,3 +769,8 @@ class HgRemote(object): repo = self._factory.repo(wire) baseui = self._factory._create_config(wire['config']) commands.bookmark(baseui, repo, bookmark, rev=revision, force=True) + + @reraise_safe_exceptions + def install_hooks(self, wire, force=False): + # we don't need any special hooks for Mercurial + pass diff --git a/vcsserver/hook_utils/__init__.py b/vcsserver/hook_utils/__init__.py new file mode 100644 --- /dev/null +++ b/vcsserver/hook_utils/__init__.py @@ -0,0 +1,154 @@ +# -*- coding: utf-8 -*- + +# RhodeCode VCSServer provides access to different vcs backends via network. +# Copyright (C) 2014-2018 RhodeCode GmbH +# +# 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, write to the Free Software Foundation, +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + +import re +import os +import sys +import datetime +import logging +import pkg_resources + +import vcsserver + +log = logging.getLogger(__name__) + + +def install_git_hooks(repo_path, bare, executable=None, force_create=False): + """ + Creates a RhodeCode hook inside a git repository + + :param repo_path: path to repository + :param executable: binary executable to put in the hooks + :param force_create: Create even if same name hook exists + """ + executable = executable or sys.executable + hooks_path = os.path.join(repo_path, 'hooks') + if not bare: + hooks_path = os.path.join(repo_path, '.git', 'hooks') + if not os.path.isdir(hooks_path): + os.makedirs(hooks_path, mode=0777) + + tmpl_post = pkg_resources.resource_string( + 'vcsserver', '/'.join( + ('hook_utils', 'hook_templates', 'git_post_receive.py.tmpl'))) + tmpl_pre = pkg_resources.resource_string( + 'vcsserver', '/'.join( + ('hook_utils', 'hook_templates', 'git_pre_receive.py.tmpl'))) + + path = '' # not used for now + timestamp = datetime.datetime.utcnow().isoformat() + + for h_type, template in [('pre', tmpl_pre), ('post', tmpl_post)]: + log.debug('Installing git hook in repo %s', repo_path) + _hook_file = os.path.join(hooks_path, '%s-receive' % h_type) + _rhodecode_hook = check_rhodecode_hook(_hook_file) + + if _rhodecode_hook or force_create: + log.debug('writing git %s hook file at %s !', h_type, _hook_file) + try: + with open(_hook_file, 'wb') as f: + template = template.replace( + '_TMPL_', vcsserver.__version__) + template = template.replace('_DATE_', timestamp) + template = template.replace('_ENV_', executable) + template = template.replace('_PATH_', path) + f.write(template) + os.chmod(_hook_file, 0755) + except IOError: + log.exception('error writing hook file %s', _hook_file) + else: + log.debug('skipping writing hook file') + + return True + + +def install_svn_hooks(repo_path, executable=None, force_create=False): + """ + Creates RhodeCode hooks inside a svn repository + + :param repo_path: path to repository + :param executable: binary executable to put in the hooks + :param force_create: Create even if same name hook exists + """ + executable = executable or sys.executable + hooks_path = os.path.join(repo_path, 'hooks') + if not os.path.isdir(hooks_path): + os.makedirs(hooks_path, mode=0777) + + tmpl_post = pkg_resources.resource_string( + 'vcsserver', '/'.join( + ('hook_utils', 'hook_templates', 'svn_post_commit_hook.py.tmpl'))) + tmpl_pre = pkg_resources.resource_string( + 'vcsserver', '/'.join( + ('hook_utils', 'hook_templates', 'svn_pre_commit_hook.py.tmpl'))) + + path = '' # not used for now + timestamp = datetime.datetime.utcnow().isoformat() + + for h_type, template in [('pre', tmpl_pre), ('post', tmpl_post)]: + log.debug('Installing svn hook in repo %s', repo_path) + _hook_file = os.path.join(hooks_path, '%s-commit' % h_type) + _rhodecode_hook = check_rhodecode_hook(_hook_file) + + if _rhodecode_hook or force_create: + log.debug('writing svn %s hook file at %s !', h_type, _hook_file) + + try: + with open(_hook_file, 'wb') as f: + template = template.replace( + '_TMPL_', vcsserver.__version__) + template = template.replace('_DATE_', timestamp) + template = template.replace('_ENV_', executable) + template = template.replace('_PATH_', path) + + f.write(template) + os.chmod(_hook_file, 0755) + except IOError: + log.exception('error writing hook file %s', _hook_file) + else: + log.debug('skipping writing hook file') + + return True + + +def check_rhodecode_hook(hook_path): + """ + Check if the hook was created by RhodeCode + """ + if not os.path.exists(hook_path): + return True + + log.debug('hook exists, checking if it is from rhodecode') + hook_content = read_hook_content(hook_path) + matches = re.search(r'(?:RC_HOOK_VER)\s*=\s*(.*)', hook_content) + if matches: + try: + version = matches.groups()[0] + log.debug('got version %s from hooks.', version) + return True + except Exception: + log.exception("Exception while reading the hook version.") + + return False + + +def read_hook_content(hook_path): + with open(hook_path, 'rb') as f: + content = f.read() + return content diff --git a/vcsserver/hook_utils/hook_templates/git_post_receive.py.tmpl b/vcsserver/hook_utils/hook_templates/git_post_receive.py.tmpl new file mode 100644 --- /dev/null +++ b/vcsserver/hook_utils/hook_templates/git_post_receive.py.tmpl @@ -0,0 +1,51 @@ +#!_ENV_ +import os +import sys +path_adjust = [_PATH_] + +if path_adjust: + sys.path = path_adjust + +try: + from vcsserver import hooks +except ImportError: + if os.environ.get('RC_DEBUG_GIT_HOOK'): + import traceback + print traceback.format_exc() + hooks = None + + +# TIMESTAMP: _DATE_ +RC_HOOK_VER = '_TMPL_' + + +def main(): + if hooks is None: + # exit with success if we cannot import vcsserver.hooks !! + # this allows simply push to this repo even without rhodecode + sys.exit(0) + + if os.environ.get('RC_SKIP_HOOKS'): + sys.exit(0) + + repo_path = os.getcwd() + push_data = sys.stdin.readlines() + os.environ['RC_HOOK_VER'] = RC_HOOK_VER + # os.environ is modified here by a subprocess call that + # runs git and later git executes this hook. + # Environ gets some additional info from rhodecode system + # like IP or username from basic-auth + try: + result = hooks.git_post_receive(repo_path, push_data, os.environ) + sys.exit(result) + except Exception as error: + # TODO: johbo: Improve handling of this special case + if not getattr(error, '_vcs_kind', None) == 'repo_locked': + raise + print 'ERROR:', error + sys.exit(1) + sys.exit(0) + + +if __name__ == '__main__': + main() diff --git a/vcsserver/hook_utils/hook_templates/git_pre_receive.py.tmpl b/vcsserver/hook_utils/hook_templates/git_pre_receive.py.tmpl new file mode 100644 --- /dev/null +++ b/vcsserver/hook_utils/hook_templates/git_pre_receive.py.tmpl @@ -0,0 +1,51 @@ +#!_ENV_ +import os +import sys +path_adjust = [_PATH_] + +if path_adjust: + sys.path = path_adjust + +try: + from vcsserver import hooks +except ImportError: + if os.environ.get('RC_DEBUG_GIT_HOOK'): + import traceback + print traceback.format_exc() + hooks = None + + +# TIMESTAMP: _DATE_ +RC_HOOK_VER = '_TMPL_' + + +def main(): + if hooks is None: + # exit with success if we cannot import vcsserver.hooks !! + # this allows simply push to this repo even without rhodecode + sys.exit(0) + + if os.environ.get('RC_SKIP_HOOKS'): + sys.exit(0) + + repo_path = os.getcwd() + push_data = sys.stdin.readlines() + os.environ['RC_HOOK_VER'] = RC_HOOK_VER + # os.environ is modified here by a subprocess call that + # runs git and later git executes this hook. + # Environ gets some additional info from rhodecode system + # like IP or username from basic-auth + try: + result = hooks.git_pre_receive(repo_path, push_data, os.environ) + sys.exit(result) + except Exception as error: + # TODO: johbo: Improve handling of this special case + if not getattr(error, '_vcs_kind', None) == 'repo_locked': + raise + print 'ERROR:', error + sys.exit(1) + sys.exit(0) + + +if __name__ == '__main__': + main() diff --git a/vcsserver/hook_utils/hook_templates/svn_post_commit_hook.py.tmpl b/vcsserver/hook_utils/hook_templates/svn_post_commit_hook.py.tmpl new file mode 100644 --- /dev/null +++ b/vcsserver/hook_utils/hook_templates/svn_post_commit_hook.py.tmpl @@ -0,0 +1,50 @@ +#!_ENV_ + +import os +import sys +path_adjust = [_PATH_] + +if path_adjust: + sys.path = path_adjust + +try: + from vcsserver import hooks +except ImportError: + if os.environ.get('RC_DEBUG_SVN_HOOK'): + import traceback + print traceback.format_exc() + hooks = None + + +# TIMESTAMP: _DATE_ +RC_HOOK_VER = '_TMPL_' + + +def main(): + if hooks is None: + # exit with success if we cannot import vcsserver.hooks !! + # this allows simply push to this repo even without rhodecode + sys.exit(0) + + if os.environ.get('RC_SKIP_HOOKS'): + sys.exit(0) + repo_path = os.getcwd() + push_data = sys.argv[1:] + + os.environ['RC_HOOK_VER'] = RC_HOOK_VER + + try: + result = hooks.svn_post_commit(repo_path, push_data, os.environ) + sys.exit(result) + except Exception as error: + # TODO: johbo: Improve handling of this special case + if not getattr(error, '_vcs_kind', None) == 'repo_locked': + raise + print 'ERROR:', error + sys.exit(1) + sys.exit(0) + + + +if __name__ == '__main__': + main() diff --git a/vcsserver/hook_utils/hook_templates/svn_pre_commit_hook.py.tmpl b/vcsserver/hook_utils/hook_templates/svn_pre_commit_hook.py.tmpl new file mode 100644 --- /dev/null +++ b/vcsserver/hook_utils/hook_templates/svn_pre_commit_hook.py.tmpl @@ -0,0 +1,52 @@ +#!_ENV_ + +import os +import sys +path_adjust = [_PATH_] + +if path_adjust: + sys.path = path_adjust + +try: + from vcsserver import hooks +except ImportError: + if os.environ.get('RC_DEBUG_SVN_HOOK'): + import traceback + print traceback.format_exc() + hooks = None + + +# TIMESTAMP: _DATE_ +RC_HOOK_VER = '_TMPL_' + + +def main(): + if os.environ.get('SSH_READ_ONLY') == '1': + sys.stderr.write('Only read-only access is allowed') + sys.exit(1) + + if hooks is None: + # exit with success if we cannot import vcsserver.hooks !! + # this allows simply push to this repo even without rhodecode + sys.exit(0) + if os.environ.get('RC_SKIP_HOOKS'): + sys.exit(0) + repo_path = os.getcwd() + push_data = sys.argv[1:] + + os.environ['RC_HOOK_VER'] = RC_HOOK_VER + + try: + result = hooks.svn_pre_commit(repo_path, push_data, os.environ) + sys.exit(result) + except Exception as error: + # TODO: johbo: Improve handling of this special case + if not getattr(error, '_vcs_kind', None) == 'repo_locked': + raise + print 'ERROR:', error + sys.exit(1) + sys.exit(0) + + +if __name__ == '__main__': + main() diff --git a/vcsserver/hooks.py b/vcsserver/hooks.py --- a/vcsserver/hooks.py +++ b/vcsserver/hooks.py @@ -20,10 +20,10 @@ import io import os import sys -import json import logging import collections import importlib +import base64 from httplib import HTTPConnection @@ -46,7 +46,11 @@ class HooksHttpClient(object): def __call__(self, method, extras): connection = HTTPConnection(self.hooks_uri) body = self._serialize(method, extras) - connection.request('POST', '/', body) + try: + connection.request('POST', '/', body) + except Exception: + log.error('Connection failed on %s', connection) + raise response = connection.getresponse() return json.loads(response.read()) @@ -97,6 +101,17 @@ class GitMessageWriter(RemoteMessageWrit self.stdout.write(message.encode('utf-8')) +class SvnMessageWriter(RemoteMessageWriter): + """Writer that knows how to send messages to svn clients.""" + + def __init__(self, stderr=None): + # SVN needs data sent to stderr for back-to-client messaging + self.stderr = stderr or sys.stderr + + def write(self, message): + self.stderr.write(message.encode('utf-8')) + + def _handle_exception(result): exception_class = result.get('exception') exception_traceback = result.get('exception_traceback') @@ -122,8 +137,9 @@ def _get_hooks_client(extras): def _call_hook(hook_name, extras, writer): - hooks = _get_hooks_client(extras) - result = hooks(hook_name, extras) + hooks_client = _get_hooks_client(extras) + log.debug('Hooks, using client:%s', hooks_client) + result = hooks_client(hook_name, extras) log.debug('Hooks got result: %s', result) writer.write(result['output']) _handle_exception(result) @@ -465,3 +481,61 @@ def git_post_receive(unused_repo_path, r pass return _call_hook('post_push', extras, GitMessageWriter()) + + +def svn_pre_commit(repo_path, commit_data, env): + path, txn_id = commit_data + branches = [] + tags = [] + + cmd = ['svnlook', 'pget', + '-t', txn_id, + '--revprop', path, 'rc-scm-extras'] + stdout, stderr = subprocessio.run_command( + cmd, env=os.environ.copy()) + extras = json.loads(base64.urlsafe_b64decode(stdout)) + + extras['commit_ids'] = [] + extras['txn_id'] = txn_id + extras['new_refs'] = { + 'branches': branches, + 'bookmarks': [], + 'tags': tags, + } + sys.stderr.write(str(extras)) + return _call_hook('pre_push', extras, SvnMessageWriter()) + + +def svn_post_commit(repo_path, commit_data, env): + """ + commit_data is path, rev, txn_id + """ + path, commit_id, txn_id = commit_data + branches = [] + tags = [] + + cmd = ['svnlook', 'pget', + '-r', commit_id, + '--revprop', path, 'rc-scm-extras'] + stdout, stderr = subprocessio.run_command( + cmd, env=os.environ.copy()) + + extras = json.loads(base64.urlsafe_b64decode(stdout)) + + extras['commit_ids'] = [commit_id] + extras['txn_id'] = txn_id + extras['new_refs'] = { + 'branches': branches, + 'bookmarks': [], + 'tags': tags, + } + + if 'repo_size' in extras['hooks']: + try: + _call_hook('repo_size', extras, SvnMessageWriter()) + except: + pass + + return _call_hook('post_push', extras, SvnMessageWriter()) + + diff --git a/vcsserver/http_main.py b/vcsserver/http_main.py --- a/vcsserver/http_main.py +++ b/vcsserver/http_main.py @@ -92,6 +92,10 @@ class VCS(object): svn_repo_cache = self.cache.get_cache_region( 'svn', region='repo_object') svn_factory = SubversionFactory(svn_repo_cache) + # hg factory is used for svn url validation + hg_repo_cache = self.cache.get_cache_region( + 'hg', region='repo_object') + hg_factory = MercurialFactory(hg_repo_cache) self._svn_remote = SvnRemote(svn_factory, hg_factory=hg_factory) else: log.info("Subversion client import failed") @@ -190,6 +194,9 @@ class HTTPApplication(object): git_path = app_settings.get('git_path', None) if git_path: settings.GIT_EXECUTABLE = git_path + binary_dir = app_settings.get('core.binary_dir', None) + if binary_dir: + settings.BINARY_DIR = binary_dir def _configure(self): self.config.add_renderer( diff --git a/vcsserver/settings.py b/vcsserver/settings.py --- a/vcsserver/settings.py +++ b/vcsserver/settings.py @@ -17,3 +17,4 @@ WIRE_ENCODING = 'UTF-8' GIT_EXECUTABLE = 'git' +BINARY_DIR = '' diff --git a/vcsserver/svn.py b/vcsserver/svn.py --- a/vcsserver/svn.py +++ b/vcsserver/svn.py @@ -32,7 +32,7 @@ import svn.diff import svn.fs import svn.repos -from vcsserver import svn_diff, exceptions, subprocessio +from vcsserver import svn_diff, exceptions, subprocessio, settings from vcsserver.base import RepoFactory, raise_from_original log = logging.getLogger(__name__) @@ -414,6 +414,17 @@ class SvnRemote(object): def is_large_file(self, wire, path): return False + @reraise_safe_exceptions + def install_hooks(self, wire, force=False): + from vcsserver.hook_utils import install_svn_hooks + repo_path = wire['path'] + binary_dir = settings.BINARY_DIR + executable = None + if binary_dir: + executable = os.path.join(binary_dir, 'python') + return install_svn_hooks( + repo_path, executable=executable, force_create=force) + class SvnDiffer(object): """ @@ -576,6 +587,7 @@ class SvnDiffer(object): return content.splitlines(True) + class DiffChangeEditor(svn.delta.Editor): """ Records changes between two given revisions diff --git a/vcsserver/tests/__init__.py b/vcsserver/tests/__init__.py new file mode 100644 --- /dev/null +++ b/vcsserver/tests/__init__.py @@ -0,0 +1,16 @@ +# RhodeCode VCSServer provides access to different vcs backends via network. +# Copyright (C) 2014-2018 RhodeCode GmbH +# +# 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, write to the Free Software Foundation, +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA diff --git a/vcsserver/tests/conftest.py b/vcsserver/tests/conftest.py --- a/vcsserver/tests/conftest.py +++ b/vcsserver/tests/conftest.py @@ -55,3 +55,4 @@ def get_available_port(): mysocket.close() del mysocket return port + diff --git a/vcsserver/tests/fixture.py b/vcsserver/tests/fixture.py --- a/vcsserver/tests/fixture.py +++ b/vcsserver/tests/fixture.py @@ -69,3 +69,18 @@ class ContextINI(object): def __exit__(self, exc_type, exc_val, exc_tb): if self.destroy: os.remove(self.new_path) + + +def no_newline_id_generator(test_name): + """ + Generates a test name without spaces or newlines characters. Used for + nicer output of progress of test + """ + org_name = test_name + test_name = test_name\ + .replace('\n', '_N') \ + .replace('\r', '_N') \ + .replace('\t', '_T') \ + .replace(' ', '_S') + + return test_name or 'test-with-empty-name' diff --git a/vcsserver/tests/test_install_hooks.py b/vcsserver/tests/test_install_hooks.py new file mode 100644 --- /dev/null +++ b/vcsserver/tests/test_install_hooks.py @@ -0,0 +1,206 @@ +# RhodeCode VCSServer provides access to different vcs backends via network. +# Copyright (C) 2014-2018 RhodeCode GmbH +# +# 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, write to the Free Software Foundation, +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + +import os +import sys +import stat +import pytest +import vcsserver +import tempfile +from vcsserver import hook_utils +from vcsserver.tests.fixture import no_newline_id_generator +from vcsserver.utils import AttributeDict + + +class TestCheckRhodecodeHook(object): + + def test_returns_false_when_hook_file_is_wrong_found(self, tmpdir): + hook = os.path.join(str(tmpdir), 'fake_hook_file.py') + with open(hook, 'wb') as f: + f.write('dummy test') + result = hook_utils.check_rhodecode_hook(hook) + assert result is False + + def test_returns_true_when_no_hook_file_found(self, tmpdir): + hook = os.path.join(str(tmpdir), 'fake_hook_file_not_existing.py') + result = hook_utils.check_rhodecode_hook(hook) + assert result + + @pytest.mark.parametrize("file_content, expected_result", [ + ("RC_HOOK_VER = '3.3.3'\n", True), + ("RC_HOOK = '3.3.3'\n", False), + ], ids=no_newline_id_generator) + def test_signatures(self, file_content, expected_result, tmpdir): + hook = os.path.join(str(tmpdir), 'fake_hook_file_1.py') + with open(hook, 'wb') as f: + f.write(file_content) + + result = hook_utils.check_rhodecode_hook(hook) + + assert result is expected_result + + +class BaseInstallHooks(object): + HOOK_FILES = () + + def _check_hook_file_mode(self, file_path): + assert os.path.exists(file_path), 'path %s missing' % file_path + stat_info = os.stat(file_path) + + file_mode = stat.S_IMODE(stat_info.st_mode) + expected_mode = int('755', 8) + assert expected_mode == file_mode + + def _check_hook_file_content(self, file_path, executable): + executable = executable or sys.executable + with open(file_path, 'rt') as hook_file: + content = hook_file.read() + + expected_env = '#!{}'.format(executable) + expected_rc_version = "\nRC_HOOK_VER = '{}'\n".format( + vcsserver.__version__) + assert content.strip().startswith(expected_env) + assert expected_rc_version in content + + def _create_fake_hook(self, file_path, content): + with open(file_path, 'w') as hook_file: + hook_file.write(content) + + def create_dummy_repo(self, repo_type): + tmpdir = tempfile.mkdtemp() + repo = AttributeDict() + if repo_type == 'git': + repo.path = os.path.join(tmpdir, 'test_git_hooks_installation_repo') + os.makedirs(repo.path) + os.makedirs(os.path.join(repo.path, 'hooks')) + repo.bare = True + + elif repo_type == 'svn': + repo.path = os.path.join(tmpdir, 'test_svn_hooks_installation_repo') + os.makedirs(repo.path) + os.makedirs(os.path.join(repo.path, 'hooks')) + + return repo + + def check_hooks(self, repo_path, repo_bare=True): + for file_name in self.HOOK_FILES: + if repo_bare: + file_path = os.path.join(repo_path, 'hooks', file_name) + else: + file_path = os.path.join(repo_path, '.git', 'hooks', file_name) + self._check_hook_file_mode(file_path) + self._check_hook_file_content(file_path, sys.executable) + + +class TestInstallGitHooks(BaseInstallHooks): + HOOK_FILES = ('pre-receive', 'post-receive') + + def test_hooks_are_installed(self): + repo = self.create_dummy_repo('git') + result = hook_utils.install_git_hooks(repo.path, repo.bare) + assert result + self.check_hooks(repo.path, repo.bare) + + def test_hooks_are_replaced(self): + repo = self.create_dummy_repo('git') + hooks_path = os.path.join(repo.path, 'hooks') + for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]: + self._create_fake_hook( + file_path, content="RC_HOOK_VER = 'abcde'\n") + + result = hook_utils.install_git_hooks(repo.path, repo.bare) + assert result + self.check_hooks(repo.path, repo.bare) + + def test_non_rc_hooks_are_not_replaced(self): + repo = self.create_dummy_repo('git') + hooks_path = os.path.join(repo.path, 'hooks') + non_rc_content = 'echo "non rc hook"\n' + for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]: + self._create_fake_hook( + file_path, content=non_rc_content) + + result = hook_utils.install_git_hooks(repo.path, repo.bare) + assert result + + for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]: + with open(file_path, 'rt') as hook_file: + content = hook_file.read() + assert content == non_rc_content + + def test_non_rc_hooks_are_replaced_with_force_flag(self): + repo = self.create_dummy_repo('git') + hooks_path = os.path.join(repo.path, 'hooks') + non_rc_content = 'echo "non rc hook"\n' + for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]: + self._create_fake_hook( + file_path, content=non_rc_content) + + result = hook_utils.install_git_hooks( + repo.path, repo.bare, force_create=True) + assert result + self.check_hooks(repo.path, repo.bare) + + +class TestInstallSvnHooks(BaseInstallHooks): + HOOK_FILES = ('pre-commit', 'post-commit') + + def test_hooks_are_installed(self): + repo = self.create_dummy_repo('svn') + result = hook_utils.install_svn_hooks(repo.path) + assert result + self.check_hooks(repo.path) + + def test_hooks_are_replaced(self): + repo = self.create_dummy_repo('svn') + hooks_path = os.path.join(repo.path, 'hooks') + for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]: + self._create_fake_hook( + file_path, content="RC_HOOK_VER = 'abcde'\n") + + result = hook_utils.install_svn_hooks(repo.path) + assert result + self.check_hooks(repo.path) + + def test_non_rc_hooks_are_not_replaced(self): + repo = self.create_dummy_repo('svn') + hooks_path = os.path.join(repo.path, 'hooks') + non_rc_content = 'echo "non rc hook"\n' + for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]: + self._create_fake_hook( + file_path, content=non_rc_content) + + result = hook_utils.install_svn_hooks(repo.path) + assert result + + for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]: + with open(file_path, 'rt') as hook_file: + content = hook_file.read() + assert content == non_rc_content + + def test_non_rc_hooks_are_replaced_with_force_flag(self): + repo = self.create_dummy_repo('svn') + hooks_path = os.path.join(repo.path, 'hooks') + non_rc_content = 'echo "non rc hook"\n' + for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]: + self._create_fake_hook( + file_path, content=non_rc_content) + + result = hook_utils.install_svn_hooks( + repo.path, force_create=True) + assert result + self.check_hooks(repo.path, ) diff --git a/vcsserver/utils.py b/vcsserver/utils.py --- a/vcsserver/utils.py +++ b/vcsserver/utils.py @@ -73,3 +73,10 @@ def safe_str(unicode_, to_encoding=['utf return unicode_.encode(encoding) except (ImportError, UnicodeEncodeError): return unicode_.encode(to_encoding[0], 'replace') + + +class AttributeDict(dict): + def __getattr__(self, attr): + return self.get(attr, None) + __setattr__ = dict.__setitem__ + __delattr__ = dict.__delitem__