rhodecode/tests/vcs/test_archives.py

# Copyright (C) 2010-2024 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, safe_bytes, safe_str
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 = b"%d/file_%d.txt" % (x, x)
            with open(os.path.join(safe_bytes(str(out_dir)), b"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 = b"%d/file_%d.txt" % (x, x)
            # NOTE: zipfile operates only on strings inside the archive
            data = zip_file.read(safe_str(b"repo/%s" % 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 = b"%d/file_%d.txt" % (x, x)
            # NOTE: zipfile operates only on strings inside the archive
            data = zip_file.read(safe_str(b"repo/%s" % 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