rhodecode/tests/vcs/test_archives.py

# Copyright (C) 2010-2023 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/

import datetime
import os
import shutil
import tarfile
import zipfile
import io

import mock
import pytest

import rhodecode
from rhodecode.lib.archive_cache import get_archival_config
from rhodecode.lib.str_utils import ascii_bytes
from rhodecode.lib.vcs.backends import base
from rhodecode.lib.vcs.exceptions import ImproperArchiveTypeError, VCSError
from rhodecode.lib.vcs.nodes import FileNode
from rhodecode.tests.vcs.conftest import BackendTestMixin


@pytest.fixture()
def d_cache_config():
    return get_archival_config(config=rhodecode.CONFIG)


@pytest.mark.usefixtures("vcs_repository_support")
class TestArchives(BackendTestMixin):

    @classmethod
    def _get_commits(cls):
        start_date = datetime.datetime(2010, 1, 1, 20)
        yield {
            'message': 'Initial Commit',
            'author': 'Joe Doe <[e-mail]>',
            'date': start_date + datetime.timedelta(hours=12),
            'added': [
                FileNode(b'executable_0o100755', b'mode_755', mode=0o100755),
                FileNode(b'executable_0o100500', b'mode_500', mode=0o100500),
                FileNode(b'not_executable', b'mode_644', mode=0o100644),
            ],
        }
        for x in range(5):
            yield {
                'message': 'Commit %d' % x,
                'author': 'Joe Doe <[e-mail]>',
                'date': start_date + datetime.timedelta(hours=12 * x),
                'added': [
                    FileNode(b'%d/file_%d.txt' % (x, x), content=b'Foobar %d' % x),
                ],
            }

    @pytest.mark.parametrize('compressor', ['gz', 'bz2'])
    def test_archive_tar(self, compressor, tmpdir, tmp_path, d_cache_config):

        archive_node = tmp_path / 'archive-node'
        archive_node.touch()

        archive_lnk = self.tip.archive_repo(
            str(archive_node), kind=f't{compressor}', archive_dir_name='repo', cache_config=d_cache_config)

        out_dir = tmpdir
        out_file = tarfile.open(str(archive_lnk), f'r|{compressor}')
        out_file.extractall(out_dir)
        out_file.close()

        for x in range(5):
            node_path = '%d/file_%d.txt' % (x, x)
            with open(os.path.join(out_dir, 'repo/' + node_path), 'rb') as f:
                file_content = f.read()
            assert file_content == self.tip.get_node(node_path).content

        shutil.rmtree(out_dir)

    @pytest.mark.parametrize('compressor', ['gz', 'bz2'])
    def test_archive_tar_symlink(self, compressor):
        pytest.skip('Not supported')

    @pytest.mark.parametrize('compressor', ['gz', 'bz2'])
    def test_archive_tar_file_modes(self, compressor, tmpdir, tmp_path, d_cache_config):
        archive_node = tmp_path / 'archive-node'
        archive_node.touch()

        archive_lnk = self.tip.archive_repo(
            str(archive_node), kind='t{}'.format(compressor), archive_dir_name='repo', cache_config=d_cache_config)

        out_dir = tmpdir
        out_file = tarfile.open(str(archive_lnk), 'r|{}'.format(compressor))
        out_file.extractall(out_dir)
        out_file.close()

        def dest(inp):
            return os.path.join(out_dir, "repo/" + inp)

        assert oct(os.stat(dest('not_executable')).st_mode) == '0o100644'

    def test_archive_zip(self, tmp_path, d_cache_config):
        archive_node = tmp_path / 'archive-node'
        archive_node.touch()

        archive_lnk = self.tip.archive_repo(str(archive_node), kind='zip',
                                            archive_dir_name='repo', cache_config=d_cache_config)
        zip_file = zipfile.ZipFile(str(archive_lnk))

        for x in range(5):
            node_path = '%d/file_%d.txt' % (x, x)
            data = zip_file.read(f'repo/{node_path}')

            decompressed = io.BytesIO()
            decompressed.write(data)
            assert decompressed.getvalue() == \
                self.tip.get_node(node_path).content
            decompressed.close()

    def test_archive_zip_with_metadata(self, tmp_path, d_cache_config):
        archive_node = tmp_path / 'archive-node'
        archive_node.touch()

        archive_lnk = self.tip.archive_repo(
            str(archive_node), kind='zip',
            archive_dir_name='repo', write_metadata=True, cache_config=d_cache_config)

        zip_file = zipfile.ZipFile(str(archive_lnk))
        metafile = zip_file.read('repo/.archival.txt')

        raw_id = ascii_bytes(self.tip.raw_id)
        assert b'commit_id:%b' % raw_id in metafile

        for x in range(5):
            node_path = '%d/file_%d.txt' % (x, x)
            data = zip_file.read(f'repo/{node_path}')
            decompressed = io.BytesIO()
            decompressed.write(data)
            assert decompressed.getvalue() == \
                self.tip.get_node(node_path).content
            decompressed.close()

    def test_archive_wrong_kind(self, tmp_path, d_cache_config):
        archive_node = tmp_path / 'archive-node'
        archive_node.touch()

        with pytest.raises(ImproperArchiveTypeError):
            self.tip.archive_repo(str(archive_node), kind='wrong kind', cache_config=d_cache_config)


@pytest.fixture()
def base_commit():
    """
    Prepare a `base.BaseCommit` just enough for `_validate_archive_prefix`.
    """
    commit = base.BaseCommit()
    commit.repository = mock.Mock()
    commit.repository.name = 'fake_repo'
    commit.short_id = 'fake_id'
    return commit


def test_validate_archive_prefix_enforces_non_ascii_as_prefix(base_commit):
    with pytest.raises(VCSError):
        base_commit._validate_archive_prefix("Ünïcödë")


def test_validate_archive_prefix_empty_prefix(base_commit):
    # TODO: johbo: Should raise a ValueError here.
    with pytest.raises(VCSError):
        base_commit._validate_archive_prefix('')


def test_validate_archive_prefix_with_leading_slash(base_commit):
    # TODO: johbo: Should raise a ValueError here.
    with pytest.raises(VCSError):
        base_commit._validate_archive_prefix('/any')


def test_validate_archive_prefix_falls_back_to_repository_name(base_commit):
    prefix = base_commit._validate_archive_prefix(None)
    expected_prefix = base_commit._ARCHIVE_PREFIX_TEMPLATE.format(
        repo_name='fake_repo',
        short_id='fake_id')
    assert isinstance(prefix, str)
    assert prefix == expected_prefix