##// END OF EJS Templates
release: Bump version 5.1.0 to 5.1.1
release: Bump version 5.1.0 to 5.1.1

File last commit:

r1249:5745b11f default
r1269:20bfc127 v5.1.1 stable
Show More
test_install_hooks.py
289 lines | 10.3 KiB | text/x-python | PythonLexer
# RhodeCode VCSServer provides access to different vcs backends via network.
# Copyright (C) 2014-2023 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.hook_utils import set_permissions_if_needed, HOOKS_DIR_MODE, HOOKS_FILE_MODE
from vcsserver.tests.fixture import no_newline_id_generator
from vcsserver.lib.str_utils import safe_bytes
from vcsserver.utils import AttributeDict
class TestCheckRhodecodeHook:
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(b'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(safe_bytes(file_content))
result = hook_utils.check_rhodecode_hook(hook)
assert result is expected_result
class BaseInstallHooks:
HOOK_FILES = ()
def _check_hook_file_dir_mode(self, file_path):
dir_path = os.path.dirname(file_path)
assert os.path.exists(dir_path), f'dir {file_path} missing'
stat_info = os.stat(dir_path)
file_mode = stat.S_IMODE(stat_info.st_mode)
expected_mode = int(HOOKS_DIR_MODE)
assert expected_mode == file_mode, f'expected mode: {oct(expected_mode)} got: {oct(file_mode)} for {dir_path}'
def _check_hook_file_mode(self, file_path):
assert os.path.exists(file_path), f'path {file_path} missing'
stat_info = os.stat(file_path)
file_mode = stat.S_IMODE(stat_info.st_mode)
expected_mode = int(HOOKS_FILE_MODE)
assert expected_mode == file_mode, f'expected mode: {oct(expected_mode)} got: {oct(file_mode)} for {file_path}'
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.get_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_dir_mode(file_path)
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, )
def create_test_file(filename):
"""Utility function to create a test file."""
with open(filename, 'w') as f:
f.write("Test file")
def remove_test_file(filename):
"""Utility function to remove a test file."""
if os.path.exists(filename):
os.remove(filename)
@pytest.fixture
def test_file():
filename = 'test_file.txt'
create_test_file(filename)
yield filename
remove_test_file(filename)
def test_increase_permissions(test_file):
# Set initial lower permissions
initial_perms = 0o644
os.chmod(test_file, initial_perms)
# Set higher permissions
new_perms = 0o666
set_permissions_if_needed(test_file, new_perms)
# Check if permissions were updated
assert (os.stat(test_file).st_mode & 0o777) == new_perms
def test_no_permission_change_needed(test_file):
# Set initial permissions
initial_perms = 0o666
os.chmod(test_file, initial_perms)
# Attempt to set the same permissions
set_permissions_if_needed(test_file, initial_perms)
# Check if permissions were unchanged
assert (os.stat(test_file).st_mode & 0o777) == initial_perms
def test_no_permission_reduction(test_file):
# Set initial higher permissions
initial_perms = 0o666
os.chmod(test_file, initial_perms)
# Attempt to set lower permissions
lower_perms = 0o644
set_permissions_if_needed(test_file, lower_perms)
# Check if permissions were not reduced
assert (os.stat(test_file).st_mode & 0o777) == initial_perms
def test_no_permission_reduction_when_on_777(test_file):
# Set initial higher permissions
initial_perms = 0o777
os.chmod(test_file, initial_perms)
# Attempt to set lower permissions
lower_perms = 0o755
set_permissions_if_needed(test_file, lower_perms)
# Check if permissions were not reduced
assert (os.stat(test_file).st_mode & 0o777) == initial_perms