diff --git a/changelog.d/3842.fixed b/changelog.d/3842.fixed new file mode 100644 index 00000000..6c6d6313 --- /dev/null +++ b/changelog.d/3842.fixed @@ -0,0 +1 @@ +Fix compatibility with Python 3.13 diff --git a/cobbler/actions/reposync.py b/cobbler/actions/reposync.py index c0163350..ec5745fb 100644 --- a/cobbler/actions/reposync.py +++ b/cobbler/actions/reposync.py @@ -23,9 +23,9 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA import logging import os import os.path -import pipes -import stat +import shlex import shutil +import stat from typing import Optional, Union from cobbler import utils @@ -272,9 +272,9 @@ class RepoSync: blended = utils.blender(self.api, False, repo) flags = blended.get("createrepo_flags", "(ERROR: FLAGS)") try: - cmd = "createrepo %s %s %s" % (" ".join(mdoptions), flags, pipes.quote(dirname)) - utils.subprocess_call(cmd) - except: + cmd = ["createrepo"] + mdoptions + flags + [shlex.quote(dirname)] + utils.subprocess_call(cmd, shell=False) + except Exception: utils.log_exc() self.logger.error("createrepo failed.") del fnames[:] # we're in the right place @@ -302,8 +302,19 @@ class RepoSync: dest_path = os.path.join(self.settings.webdir, "repo_mirror", repo.name) # FIXME: wrapper for subprocess that logs to logger - cmd = ["wget", "-N", "-np", "-r", "-l", "inf", "-nd", "-P", pipes.quote(dest_path), pipes.quote(repo.mirror)] - rc = utils.subprocess_call(cmd) + cmd = [ + "wget", + "-N", + "-np", + "-r", + "-l", + "inf", + "-nd", + "-P", + shlex.quote(dest_path), + shlex.quote(repo.mirror), + ] + return_value = utils.subprocess_call(cmd, shell=False) if rc != 0: raise CX("cobbler reposync failed") @@ -347,9 +358,14 @@ class RepoSync: if flags == '': flags = self.settings.reposync_rsync_flags - cmd = "rsync %s --delete-after %s --delete --exclude-from=/etc/cobbler/rsync.exclude %s %s" \ - % (flags, spacer, pipes.quote(repo.mirror), pipes.quote(dest_path)) - rc = utils.subprocess_call(cmd) + cmd = ["rsync"] + flags + ["--delete-after"] + cmd += spacer + [ + "--delete", + "--exclude-from=/etc/cobbler/rsync.exclude", + shlex.quote(repo.mirror), + shlex.quote(dest_path), + ] + return_code = utils.subprocess_call(cmd, shell=False) if rc != 0: raise CX("cobbler reposync failed") @@ -386,10 +402,11 @@ class RepoSync: if not HAS_LIBREPO: raise CX("no librepo found, please install python3-librepo") - if os.path.exists("/usr/bin/dnf"): - cmd = "/usr/bin/dnf reposync" - elif os.path.exists("/usr/bin/reposync"): - cmd = "/usr/bin/reposync" + if os.path.exists("/usr/bin/reposync"): + cmd = ["/usr/bin/reposync"] + # DNF5 does not have a reposync subcommand + elif os.path.exists("/usr/bin/dnf"): + cmd = ["/usr/bin/dnf", "reposync"] else: # Warn about not having yum-utils. We don't want to require it in the package because Fedora 22+ has moved # to dnf. @@ -451,6 +468,11 @@ class RepoSync: # Counter-intuitive, but we want the newish kernels too arch = "i686" + cmd = self.reposync_cmd() + cmd += self.rflags + [ + f"--repo={shlex.quote(rest)}", + f"--download-path={shlex.quote(repos_path)}", + ] if arch != "none": cmd = "%s -a %s" % (cmd, arch) @@ -544,9 +566,11 @@ class RepoSync: if not has_rpm_list: # If we have not requested only certain RPMs, use reposync - cmd = "%s %s --config=%s --repoid=%s -p %s" \ - % (cmd, self.rflags, temp_file, pipes.quote(repo.name), - pipes.quote(repos_path)) + cmd += self.rflags + [ + f"--config={temp_file}", + f"--repoid={shlex.quote(repo.name)}", + f"--download-path={shlex.quote(repos_path)}", + ] if arch != "none": cmd = "%s -a %s" % (cmd, arch) @@ -557,14 +581,14 @@ class RepoSync: use_source = "" if arch == "src": - use_source = "--source" - - # Older yumdownloader sometimes explodes on --resolvedeps if this happens to you, upgrade yum & yum-utils - extra_flags = self.settings.yumdownloader_flags - cmd = "/usr/bin/dnf download" - cmd = "%s %s %s --disablerepo=* --enablerepo=%s -c %s --destdir=%s %s" \ - % (cmd, extra_flags, use_source, pipes.quote(repo.name), temp_file, pipes.quote(dest_path), - " ".join(repo.rpm_list)) + cmd.append("--source") + cmd += [ + "--disablerepo=*", + f"--enablerepo={shlex.quote(repo.name)}", + f"-c={temp_file}", + f"--destdir={shlex.quote(dest_path)}", + ] + cmd += repo.rpm_list # Now regardless of whether we're doing yumdownloader or reposync or whether the repo was http://, ftp://, or # rhn://, execute all queued commands here. Any failure at any point stops the operation. @@ -669,17 +693,21 @@ class RepoSync: dists = ",".join(repo.apt_dists) components = ",".join(repo.apt_components) - mirror_data = "--method=%s --host=%s --root=%s --dist=%s --section=%s" \ - % (pipes.quote(method), pipes.quote(host), pipes.quote(mirror), pipes.quote(dists), - pipes.quote(components)) + mirror_data = [ + f"--method={shlex.quote(method)}", + f"--host={shlex.quote(host)}", + f"--root={shlex.quote(mirror)}", + f"--dist={shlex.quote(dists)}", + f"--section={shlex.quote(components)}", + ] rflags = "--nocleanup" for x in repo.yumopts: if repo.yumopts[x]: rflags += " %s=%s" % (x, repo.yumopts[x]) else: - rflags += " %s" % x - cmd = "%s %s %s %s" % (mirror_program, rflags, mirror_data, pipes.quote(dest_path)) + rflags.append(repo_yumoption) + cmd = [mirror_program] + rflags + mirror_data + [shlex.quote(dest_path)] if repo.arch == RepoArchs.SRC: cmd = "%s --source" % cmd else: diff --git a/tests/actions/reposync_test.py b/tests/actions/reposync_test.py index 0bee772c..ee8d1549 100644 --- a/tests/actions/reposync_test.py +++ b/tests/actions/reposync_test.py @@ -1,251 +1,592 @@ +""" +Tests that validate the functionality of the module that is responsible for repository synchronization. +""" + import os -import glob +from pathlib import Path +from typing import TYPE_CHECKING, Any, Dict, List, Union import pytest -from cobbler import enums +from cobbler import cexceptions, enums +from cobbler.actions import reposync from cobbler.api import CobblerAPI -from cobbler.actions.reposync import RepoSync from cobbler.items.repo import Repo -from cobbler import cexceptions -from tests.conftest import does_not_raise +from tests.conftest import does_not_raise -@pytest.fixture(scope="class") -def api(): - return CobblerAPI() +if TYPE_CHECKING: + from pytest_mock import MockerFixture -@pytest.fixture(scope="class") -def reposync(api): - test_reposync = RepoSync(api, tries=2, nofail=False) +@pytest.fixture(name="reposync_object", scope="function") +def fixture_reposync_object( + mocker: "MockerFixture", cobbler_api: CobblerAPI +) -> reposync.RepoSync: + settings_mock = mocker.MagicMock() + settings_mock.webdir = "/srv/www/cobbler" + settings_mock.server = "localhost" + settings_mock.http_port = 80 + settings_mock.proxy_url_ext = "" + settings_mock.yumdownloader_flags = "--testflag" + settings_mock.reposync_rsync_flags = "--testflag" + settings_mock.reposync_flags = "--testflag" + mocker.patch.object(cobbler_api, "settings", return_value=settings_mock) + test_reposync = reposync.RepoSync(cobbler_api, tries=2, nofail=False) return test_reposync -@pytest.fixture -def repo(api): +@pytest.fixture(name="repo") +def fixture_repo(cobbler_api: CobblerAPI) -> Repo: """ Creates a Repository "testrepo0" with a keep_updated=True and mirror_locally=True". """ - test_repo = Repo(api) + test_repo = Repo(cobbler_api) test_repo.name = "testrepo0" test_repo.mirror_locally = True test_repo.keep_updated = True - api.add_repo(test_repo) return test_repo @pytest.fixture -def remove_repo(api): +def remove_repo(cobbler_api: CobblerAPI): """ Removes the Repository "testrepo0" which can be created with repo. """ yield - test_repo = api.find_repo("testrepo0") - if test_repo is not None: - api.remove_repo(test_repo.name) + test_repo = cobbler_api.find_repo("testrepo0") + if test_repo is not None and not isinstance(test_repo, list): + cobbler_api.remove_repo(test_repo.name) -class TestRepoSync: - @pytest.mark.usefixtures("remove_repo") - @pytest.mark.parametrize( - "input_mirror_type,input_mirror,expected_exception", - [ - ( - enums.MirrorType.BASEURL, - "http://download.fedoraproject.org/pub/fedora/linux/development/rawhide/Everything/x86_64/os", - does_not_raise() - ), - ( - enums.MirrorType.MIRRORLIST, - "https://mirrors.fedoraproject.org/mirrorlist?repo=rawhide&arch=x86_64", - does_not_raise() - ), - ( - enums.MirrorType.METALINK, - "https://mirrors.fedoraproject.org/metalink?repo=rawhide&arch=x86_64", - does_not_raise() - ), - ( - enums.MirrorType.BASEURL, - "http://www.example.com/path/to/some/repo", - pytest.raises(cexceptions.CX) - ), +@pytest.fixture(scope="function", autouse=True) +def reset_librepo(): + has_librepo = reposync.HAS_LIBREPO + yield + reposync.HAS_LIBREPO = has_librepo + + +def test_repo_walker(mocker: "MockerFixture", tmp_path: Path): + # Arrange + def test_fun(arg: Any, top: Any, names: Any): + pass + + subdir1 = tmp_path / "sub1" + subdir2 = tmp_path / "sub2" + subdir1.mkdir() + subdir2.mkdir() + spy = mocker.Mock(wraps=test_fun) + + # Act + reposync.repo_walker(tmp_path, spy, None) # type: ignore + + # Assert + assert spy.mock_calls == [ + # settings.yaml is here because of our autouse fixture that we use to restore the settings + mocker.call(None, tmp_path, ["settings.yaml", "sub1", "sub2"]), + mocker.call(None, str(subdir1), []), + mocker.call(None, str(subdir2), []), + ] + + +@pytest.mark.parametrize( + "input_has_librepo,input_path_exists_side_effect,expected_exception,expected_result", + [ + (True, [False, True], does_not_raise(), ["/usr/bin/dnf", "reposync"]), + (True, [True, False], does_not_raise(), ["/usr/bin/reposync"]), + (True, [False, False], pytest.raises(cexceptions.CX), ""), + (False, [False, True], pytest.raises(cexceptions.CX), ""), + ], +) +def test_reposync_cmd( + mocker: "MockerFixture", + reposync_object: reposync.RepoSync, + input_has_librepo: bool, + input_path_exists_side_effect: List[bool], + expected_exception: Any, + expected_result: Union[List[str], str], +): + # Arrange + mocker.patch("os.path.exists", side_effect=input_path_exists_side_effect) + reposync.HAS_LIBREPO = input_has_librepo + + # Act + with expected_exception: + result = reposync_object.reposync_cmd() + + # Assert + assert result == expected_result + + +def test_run(mocker: "MockerFixture", reposync_object: reposync.RepoSync, repo: Repo): + # Arrange + env_vars: Dict[str, Any] = {} + mocker.patch("os.makedirs") + mocker.patch("os.path.isdir", return_value=True) + mocker.patch( + "os.path.join", + side_effect=[ + "/srv/www/cobbler/repo_mirror", + "/srv/www/cobbler/repo_mirror/%s" % repo.name, ], ) - def test_reposync_yum( - self, - input_mirror_type, - input_mirror, - expected_exception, - api, - repo, - reposync - ): - # Arrange - test_repo = repo - test_repo.breed = enums.RepoBreeds.YUM - test_repo.mirror = input_mirror - test_repo.mirror_type = input_mirror_type - test_repo.rpm_list = "fedora-gpg-keys" - test_settings = api.settings() - repo_path = os.path.join(test_settings.webdir, "repo_mirror", test_repo.name) - - # Act & Assert - with expected_exception: - reposync.run(test_repo.name) - result = os.path.exists(repo_path) - if test_repo.rpm_list and test_repo.rpm_list != []: - for rpm in test_repo.rpm_list: - assert glob.glob(os.path.join(repo_path, "**", rpm) + "*.rpm", recursive=True) != [] - assert result - # Test that re-downloading the metadata in .origin/repodata will not result in an error - reposync.run(test_repo.name) - - @pytest.mark.usefixtures("remove_repo") - @pytest.mark.parametrize( - "input_mirror_type,input_mirror,input_arch,input_rpm_list,expected_exception", + mocker.patch("os.environ", return_value=env_vars) + mocker.patch.object(reposync_object, "repos", return_value=[repo]) + mocker.patch.object(reposync_object, "sync") + mocker.patch.object(reposync_object, "update_permissions") + reposync_object.repos = [repo] # type: ignore + + # Act + reposync_object.run() + + # Assert + # This has to be 0 since all env vars need to be removed after reposync has run. + assert len(env_vars) == 0 + + +def test_gen_urlgrab_ssl_opts(reposync_object: reposync.RepoSync): + # Arrange + input_dict: Dict[str, Any] = {} + + # Act + result = reposync_object.gen_urlgrab_ssl_opts(input_dict) + + # Assert + assert isinstance(result, tuple) + assert len(result) == 2 + # The data of the first element is kind of flexible let's skip asserting it for now + assert isinstance(result[1], bool) + + +@pytest.mark.usefixtures("remove_repo") +@pytest.mark.parametrize( + "input_mirror_type,input_mirror,expected_exception", + [ + ( + enums.MirrorType.BASEURL, + "http://download.fedoraproject.org/pub/fedora/linux/development/rawhide/Everything/x86_64/os", + does_not_raise(), + ), + ( + enums.MirrorType.MIRRORLIST, + "https://mirrors.fedoraproject.org/mirrorlist?repo=rawhide&arch=x86_64", + does_not_raise(), + ), + ( + enums.MirrorType.METALINK, + "https://mirrors.fedoraproject.org/metalink?repo=rawhide&arch=x86_64", + does_not_raise(), + ), + ], +) +def test_reposync_yum( + mocker: "MockerFixture", + input_mirror_type: enums.MirrorType, + input_mirror: str, + expected_exception: Any, + cobbler_api: CobblerAPI, + repo: Repo, + reposync_object: reposync.RepoSync, +): + # Arrange + test_repo = repo + test_repo.breed = enums.RepoBreeds.YUM + test_repo.mirror = input_mirror + test_repo.mirror_type = input_mirror_type + test_repo.rpm_list = "fedora-gpg-keys" + test_settings = cobbler_api.settings() + repo_path = os.path.join(test_settings.webdir, "repo_mirror", test_repo.name) + mocked_subprocess = mocker.patch( + "cobbler.utils.subprocess_call", autospec=True, return_value=0 + ) + mocker.patch.object( + reposync_object, "create_local_file", return_value="/create/local/file" + ) + mocker.patch.object( + reposync_object, "reposync_cmd", return_value=["/my/fake/dnf", "reposync"] + ) + mocker.patch.object(reposync_object, "rflags", return_value="--fake-r-flakg") + mocker.patch.object( + reposync_object, + "gen_urlgrab_ssl_opts", + return_value=(("TODO", "TODO", "TODO"), False), + ) + mocker.patch("os.path.exists", return_value=True) + mocker.patch("shutil.rmtree") + mocker.patch("os.makedirs") + mocked_repo_walker = mocker.patch("cobbler.actions.reposync.repo_walker") + handle_mock = mocker.MagicMock() + result_mock = mocker.MagicMock() + mocker.patch("librepo.Handle", return_value=handle_mock) + mocker.patch("librepo.Result", return_value=result_mock) + + # Act & Assert + with expected_exception: + reposync_object.yum_sync(repo) + + mocked_subprocess.assert_called_with( + [ + "/usr/bin/dnf", + "download", + "--testflag", + "--disablerepo=*", + f"--enablerepo={repo.name}", + "-c=/create/local/file", + f"--destdir={repo_path}", + "fedora-gpg-keys", + ], + shell=False, + ) + handle_mock.perform.assert_called_with(result_mock) + assert mocked_repo_walker.call_count == 1 + + +@pytest.mark.usefixtures("remove_repo") +@pytest.mark.parametrize( + "input_mirror_type,input_mirror,input_arch,input_rpm_list,expected_exception", + [ + ( + enums.MirrorType.BASEURL, + "http://ftp.debian.org/debian", + enums.RepoArchs.X86_64, + "", + does_not_raise(), + ), + ( + enums.MirrorType.MIRRORLIST, + "http://ftp.debian.org/debian", + enums.RepoArchs.X86_64, + "", + pytest.raises(cexceptions.CX), + ), + ( + enums.MirrorType.METALINK, + "http://ftp.debian.org/debian", + enums.RepoArchs.X86_64, + "", + pytest.raises(cexceptions.CX), + ), + ( + enums.MirrorType.BASEURL, + "http://ftp.debian.org/debian", + enums.RepoArchs.NONE, + "", + pytest.raises(cexceptions.CX), + ), + ( + enums.MirrorType.BASEURL, + "http://ftp.debian.org/debian", + enums.RepoArchs.X86_64, + "dpkg", + pytest.raises(cexceptions.CX), + ), + ], +) +def test_reposync_apt( + mocker: "MockerFixture", + input_mirror_type: enums.MirrorType, + input_mirror: str, + input_arch: enums.RepoArchs, + input_rpm_list: str, + expected_exception: Any, + cobbler_api: CobblerAPI, + repo: Repo, + reposync_object: reposync.RepoSync, +): + # Arrange + test_repo = repo + test_repo.breed = enums.RepoBreeds.APT + test_repo.arch = input_arch + test_repo.apt_components = "main" + test_repo.apt_dists = "stable" + test_repo.mirror = input_mirror + test_repo.mirror_type = input_mirror_type + test_repo.rpm_list = input_rpm_list + test_settings = cobbler_api.settings() + repo_path = os.path.join(test_settings.webdir, "repo_mirror", test_repo.name) + mocked_subprocess = mocker.patch( + "cobbler.utils.subprocess_call", autospec=True, return_value=0 + ) + mocker.patch("os.path.exists", return_value=True) + + # Act + with expected_exception: + reposync_object.apt_sync(repo) + + # Assert + mocked_subprocess.assert_called_with( + [ + "/usr/bin/debmirror", + "--nocleanup", + "--method=http", + "--host=ftp.debian.org", + "--root=/debian", + "--dist=stable", + "--section=main", + repo_path, + "--nosource", + "-a=amd64", + ], + shell=False, + ) + + +@pytest.mark.usefixtures("remove_repo") +@pytest.mark.parametrize( + "input_mirror_type,input_mirror,expected_exception", + [ + ( + enums.MirrorType.BASEURL, + "http://download.fedoraproject.org/pub/fedora/linux/development/rawhide/Everything/x86_64/os/Packages/2", + does_not_raise(), + ), + ( + enums.MirrorType.MIRRORLIST, + "http://download.fedoraproject.org/pub/fedora/linux/development/rawhide/Everything/x86_64/os/Packages/2", + pytest.raises(cexceptions.CX), + ), + ( + enums.MirrorType.METALINK, + "http://download.fedoraproject.org/pub/fedora/linux/development/rawhide/Everything/x86_64/os/Packages/2", + pytest.raises(cexceptions.CX), + ), + ], +) +def test_reposync_wget( + mocker: "MockerFixture", + input_mirror_type: enums.MirrorType, + input_mirror: str, + expected_exception: Any, + cobbler_api: CobblerAPI, + repo: Repo, + reposync_object: reposync.RepoSync, +): + # Arrange + test_repo = repo + test_repo.breed = enums.RepoBreeds.WGET + test_repo.mirror = input_mirror + test_repo.mirror_type = input_mirror_type + repo_path = os.path.join( + reposync_object.settings.webdir, "repo_mirror", test_repo.name + ) + mocked_subprocess = mocker.patch( + "cobbler.utils.subprocess_call", autospec=True, return_value=0 + ) + mocker.patch("cobbler.actions.reposync.repo_walker") + mocker.patch.object(reposync_object, "create_local_file") + + # Act + with expected_exception: + reposync_object.wget_sync(test_repo) + + # Assert + mocked_subprocess.assert_called_with( + [ + "wget", + "-N", + "-np", + "-r", + "-l", + "inf", + "-nd", + "-P", + repo_path, + input_mirror, + ], + shell=False, + ) + + +def test_reposync_rhn( + mocker: "MockerFixture", reposync_object: reposync.RepoSync, repo: Repo +): + # Arrange + repo.mirror = "rhn://%s" % repo.name + mocked_subprocess = mocker.patch( + "cobbler.utils.subprocess_call", autospec=True, return_value=0 + ) + mocker.patch("os.path.isdir", return_value=True) + mocker.patch("os.makedirs") + mocker.patch("cobbler.actions.reposync.repo_walker") + mocker.patch.object(reposync_object, "create_local_file") + mocker.patch.object( + reposync_object, "reposync_cmd", return_value=["/my/fake/reposync"] + ) + + # Act + reposync_object.rhn_sync(repo) + + # Assert + # TODO: Check this more and document how its actually working + mocked_subprocess.assert_called_with( [ - ( - enums.MirrorType.BASEURL, - "http://ftp.debian.org/debian", - enums.RepoArchs.X86_64, - "", - does_not_raise() - ), - ( - enums.MirrorType.MIRRORLIST, - "http://ftp.debian.org/debian", - enums.RepoArchs.X86_64, - "", - pytest.raises(cexceptions.CX) - ), - ( - enums.MirrorType.METALINK, - "http://ftp.debian.org/debian", - enums.RepoArchs.X86_64, - "", - pytest.raises(cexceptions.CX) - ), - ( - enums.MirrorType.BASEURL, - "http://www.example.com/path/to/some/repo", - enums.RepoArchs.X86_64, - "", - pytest.raises(cexceptions.CX) - ), - ( - enums.MirrorType.BASEURL, - "http://ftp.debian.org/debian", - enums.RepoArchs.NONE, - "", - pytest.raises(cexceptions.CX) - ), - ( - enums.MirrorType.BASEURL, - "http://ftp.debian.org/debian", - enums.RepoArchs.X86_64, - "dpkg", - pytest.raises(cexceptions.CX) - ), + "/my/fake/reposync", + "--testflag", + "--repo=testrepo0", + "--download-path=/srv/www/cobbler/repo_mirror", ], + shell=False, ) - def test_reposync_apt( - self, - input_mirror_type, - input_mirror, - input_arch, - input_rpm_list, - expected_exception, - api, - repo, - reposync - ): - # Arrange - test_repo = repo - test_repo.breed = enums.RepoBreeds.APT - test_repo.arch = input_arch - test_repo.apt_components = "main" - test_repo.apt_dists = "stable" - test_repo.mirror = input_mirror - test_repo.mirror_type = input_mirror_type - test_repo.rpm_list = input_rpm_list - test_repo.yumopts = "--exclude=.* --include=dpkg.* --no-check-gpg --rsync-extra=none" - test_settings = api.settings() - repo_path = os.path.join(test_settings.webdir, "repo_mirror", test_repo.name) - - # Act & Assert - with expected_exception: - reposync.run(test_repo.name) - result = os.path.exists(repo_path) - for rpm in ["dpkg"]: - assert glob.glob(os.path.join(repo_path, "**", "dpkg") + "*", recursive=True) != [] - assert result - - @pytest.mark.skip("To flaky and thus not reliable. Needs to be mocked to be of use.") - @pytest.mark.usefixtures("remove_repo") - @pytest.mark.parametrize( - "input_mirror_type,input_mirror,expected_exception", + + +def test_reposync_rsync( + mocker: "MockerFixture", reposync_object: reposync.RepoSync, repo: Repo +): + # Arrange + mocked_subprocess = mocker.patch("cobbler.utils.subprocess_call", return_value=0) + mocker.patch("cobbler.actions.reposync.repo_walker") + mocker.patch.object(reposync_object, "create_local_file") + repo_path = os.path.join(reposync_object.settings.webdir, "repo_mirror", repo.name) + + # Act + reposync_object.rsync_sync(repo) + + # Assert + mocked_subprocess.assert_called_with( [ - ( - enums.MirrorType.BASEURL, - "http://download.fedoraproject.org/pub/fedora/linux/development/rawhide/Everything/x86_64/os/Packages/2", - does_not_raise() - ), - ( - enums.MirrorType.MIRRORLIST, - "http://download.fedoraproject.org/pub/fedora/linux/development/rawhide/Everything/x86_64/os/Packages/2", - pytest.raises(cexceptions.CX) - ), - ( - enums.MirrorType.METALINK, - "http://download.fedoraproject.org/pub/fedora/linux/development/rawhide/Everything/x86_64/os/Packages/2", - pytest.raises(cexceptions.CX) - ), - ( - enums.MirrorType.BASEURL, - "http://www.example.com/path/to/some/repo", - pytest.raises(cexceptions.CX) - ), + "rsync", + "--testflag", + "--delete-after", + "-e ssh", + "--delete", + "--exclude-from=/etc/cobbler/rsync.exclude", + "/", + repo_path, ], + shell=False, ) - def test_reposync_wget( - self, - input_mirror_type, - input_mirror, - expected_exception, - api, - repo, - reposync - ): - # Arrange - test_repo = repo - test_repo.breed = enums.RepoBreeds.WGET - test_repo.mirror = input_mirror - test_repo.mirror_type = input_mirror_type - test_settings = api.settings() - repo_path = os.path.join(test_settings.webdir, "repo_mirror", test_repo.name) - - # Act & Assert - with expected_exception: - reposync.run(test_repo.name) - result = os.path.exists(repo_path) - for rpm in ["rpm"]: - assert glob.glob(os.path.join(repo_path, "**", "2") + "*", recursive=True) != [] - assert result - - -@pytest.mark.skip("TODO") -def test_reposync_rhn(): + + +def test_createrepo_walker( + mocker: "MockerFixture", reposync_object: reposync.RepoSync, repo: Repo +): # Arrange + input_repo = repo + input_repo.breed = enums.RepoBreeds.RSYNC + input_dirname = "" + input_fnames = [] + expected_call = ["createrepo", "--testflags", f"'{input_dirname}'"] + mocked_subprocess = mocker.patch( + "cobbler.utils.subprocess_call", autospec=True, return_value=0 + ) + mocker.patch( + "cobbler.utils.blender", + autospec=True, + return_value={"createrepo_flags": "--testflags"}, + ) + mocker.patch("cobbler.utils.remove_yum_olddata") + mocker.patch("cobbler.utils.subprocess_get", return_value="5") + mocker.patch("cobbler.utils.get_family", return_value="TODO") + mocker.patch("os.path.exists", return_value=True) + mocker.patch("os.path.isfile", return_value=True) + mocker.patch.object(reposync_object, "librepo_getinfo", return_value={}) + # Act + reposync_object.createrepo_walker(input_repo, input_dirname, input_fnames) + # Assert - assert False + # TODO: Improve coverage over different cases in method + mocked_subprocess.assert_called_with(expected_call, shell=False) -@pytest.mark.skip("TODO") -def test_reposync_rsync(): +@pytest.mark.parametrize( + "input_repotype,expected_exception", + [ + (enums.RepoBreeds.YUM, does_not_raise()), + (enums.RepoBreeds.RHN, does_not_raise()), + (enums.RepoBreeds.APT, does_not_raise()), + (enums.RepoBreeds.RSYNC, does_not_raise()), + (enums.RepoBreeds.WGET, does_not_raise()), + (enums.RepoBreeds.NONE, pytest.raises(cexceptions.CX)), + ], +) +def test_sync( + mocker: "MockerFixture", + cobbler_api: CobblerAPI, + reposync_object: reposync.RepoSync, + input_repotype: enums.RepoBreeds, + expected_exception: Any, +): # Arrange + test_repo = Repo(cobbler_api) + test_repo.breed = input_repotype + rhn_sync_mock = mocker.patch.object(reposync_object, "rhn_sync") + yum_sync_mock = mocker.patch.object(reposync_object, "yum_sync") + apt_sync_mock = mocker.patch.object(reposync_object, "apt_sync") + rsync_sync_mock = mocker.patch.object(reposync_object, "rsync_sync") + wget_sync_mock = mocker.patch.object(reposync_object, "wget_sync") + # Act + with expected_exception: + reposync_object.sync(test_repo) + + # Assert + call_count = sum( + ( + rhn_sync_mock.call_count, + yum_sync_mock.call_count, + apt_sync_mock.call_count, + rsync_sync_mock.call_count, + wget_sync_mock.call_count, + ) + ) + assert call_count == 1 + + +def test_librepo_getinfo( + mocker: "MockerFixture", reposync_object: reposync.RepoSync, tmp_path: Path +): + # Arrange + handle_mock = mocker.MagicMock() + result_mock = mocker.MagicMock() + mocker.patch("librepo.Handle", return_value=handle_mock) + mocker.patch("librepo.Result", return_value=result_mock) + + # Act + reposync_object.librepo_getinfo(str(tmp_path)) + + # Assert + handle_mock.perform.assert_called_with(result_mock) + result_mock.getinfo.assert_called() + + +def test_create_local_file( + mocker: "MockerFixture", reposync_object: reposync.RepoSync, repo: Repo +): + # Arrange + mocker.patch("cobbler.utils.filesystem_helpers.mkdir", autospec=True) + mock_open = mocker.patch("builtins.open", mocker.mock_open()) + input_dest_path = "" + input_repo = repo + input_output = True + + # Act + reposync_object.create_local_file(input_dest_path, input_repo, output=input_output) + + # Assert + # TODO: Extend checks + assert mock_open.call_count == 1 + assert mock_open.mock_calls[0] == mocker.call("config.repo", "w", encoding="UTF-8") + mock_open_handle = mock_open() + assert mock_open_handle.write.mock_calls[0] == mocker.call("[testrepo0]\n") + assert mock_open_handle.write.mock_calls[1] == mocker.call("name=testrepo0\n") + + +def test_update_permissions( + mocker: "MockerFixture", reposync_object: reposync.RepoSync +): + # Arrange + mocked_subprocess = mocker.patch( + "cobbler.utils.subprocess_call", autospec=True, return_value=0 + ) + path_to_update = "/my/fake/path" + expected_calls = [ + mocker.call(["chown", "-R", "root:www", path_to_update], shell=False), + mocker.call(["chmod", "-R", "755", path_to_update], shell=False), + ] + + # Act + reposync_object.update_permissions(path_to_update) + # Assert - assert False + assert mocked_subprocess.mock_calls == expected_calls