vcsserver/tests/test_install_hooks.py

# RhodeCode VCSServer provides access to different vcs backends via network.
# Copyright (C) 2014-2024 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