rpms/cobbler
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
cobbler/cobbler-python3.13.patch

972 lines
33 KiB
Diff
Raw Permalink Blame History

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
Reference in a new issue View git blame Copy permalink