# 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.rc_cache.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 <joe.doe@example.com>', '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 <joe.doe@example.com>', '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