diff --git a/test/test_debian.py b/test/test_debian.py new file mode 100644 index 0000000..48ae41b --- /dev/null +++ b/test/test_debian.py @@ -0,0 +1,103 @@ +# -*- coding: utf-8 -*- +# Copyright 2025, Red Hat, Inc. +# License: GPL-2.0+ +# See the LICENSE file for more details on Licensing + +"""Tests for Debian cloud image URL generation.""" + +import pytest + +from testcloud import config, exceptions +from testcloud.distro_utils import debian +from testcloud.distro_utils.debian import get_debian_image_url + + +PRIMARY = "https://cdimage.debian.org/images/cloud" +FALLBACK = "https://cloud.debian.org/images/cloud" + + +class TestGetDebianImageUrl(object): + + def setup_method(self, method): + config._config = None + + def teardown_method(self, method): + config._config = None + + def test_latest(self, monkeypatch): + monkeypatch.setattr(config, "CONF_DIRS", []) + urls = get_debian_image_url(version="latest", arch="x86_64") + assert len(urls) == 2 + assert "bookworm" in urls[0] + assert "genericcloud-amd64" in urls[0] + assert urls[0].startswith(PRIMARY) + assert urls[1].startswith(FALLBACK) + + def test_version_number(self, monkeypatch): + monkeypatch.setattr(config, "CONF_DIRS", []) + urls = get_debian_image_url(version="12", arch="x86_64") + assert "bookworm" in urls[0] + assert "debian-12-genericcloud" in urls[0] + + def test_codename(self, monkeypatch): + monkeypatch.setattr(config, "CONF_DIRS", []) + urls = get_debian_image_url(version="bookworm", arch="x86_64") + assert "bookworm" in urls[0] + assert "debian-12-genericcloud" in urls[0] + + def test_sid(self, monkeypatch): + monkeypatch.setattr(config, "CONF_DIRS", []) + urls = get_debian_image_url(version="sid", arch="x86_64") + assert "sid/daily/latest/debian-sid-genericcloud-amd64-daily.qcow2" in urls[0] + + def test_invalid_version(self, monkeypatch): + monkeypatch.setattr(config, "CONF_DIRS", []) + with pytest.raises(exceptions.TestcloudImageError): + get_debian_image_url(version="99", arch="x86_64") + + def test_unsupported_arch(self, monkeypatch): + monkeypatch.setattr(config, "CONF_DIRS", []) + with pytest.raises(exceptions.TestcloudImageError): + get_debian_image_url(version="latest", arch="s390x") + + def test_aarch64_uses_generic(self, monkeypatch): + monkeypatch.setattr(config, "CONF_DIRS", []) + urls = get_debian_image_url(version="12", arch="aarch64") + for url in urls: + assert "generic-arm64" in url + assert "genericcloud" not in url + + def test_str_config_compat(self, monkeypatch): + monkeypatch.setattr(config, "CONF_DIRS", []) + monkeypatch.setattr(debian.config_data, "DEBIAN_IMG_URL", + "https://example.com/%s/%s/%s.qcow2") + urls = get_debian_image_url(version="12", arch="x86_64") + assert len(urls) == 1 + assert urls[0].startswith("https://example.com/") + + def test_list_config(self, monkeypatch): + monkeypatch.setattr(config, "CONF_DIRS", []) + monkeypatch.setattr(debian.config_data, "DEBIAN_IMG_URL", [ + "https://mirror1.example.com/%s/%s/%s.qcow2", + "https://mirror2.example.com/%s/%s/%s.qcow2", + "https://mirror3.example.com/%s/%s/%s.qcow2", + ]) + urls = get_debian_image_url(version="12", arch="x86_64") + assert len(urls) == 3 + assert urls[0].startswith("https://mirror1") + assert urls[2].startswith("https://mirror3") + + def test_no_config_mutation(self, monkeypatch): + monkeypatch.setattr(config, "CONF_DIRS", []) + cfg = config.get_config() + original = list(cfg.DEBIAN_IMG_URL) if isinstance(cfg.DEBIAN_IMG_URL, list) else cfg.DEBIAN_IMG_URL + + get_debian_image_url(version="12", arch="aarch64") + + if isinstance(original, list): + assert cfg.DEBIAN_IMG_URL == original + for template in cfg.DEBIAN_IMG_URL: + assert "genericcloud" in template + else: + assert cfg.DEBIAN_IMG_URL == original + assert "genericcloud" in cfg.DEBIAN_IMG_URL diff --git a/test/test_image.py b/test/test_image.py index 3085038..ac5db1c 100644 --- a/test/test_image.py +++ b/test/test_image.py @@ -5,6 +5,7 @@ """ This module is for testing the behaviour of the Image class.""" +from unittest import mock from unittest.mock import patch import pytest @@ -28,6 +29,68 @@ def test_download(self): pass +class TestImageDownloadFallback(object): + + def setup_method(self, method): + DB.bind([DBImage], bind_refs=False, bind_backrefs=False) + DB.connect() + DB.create_tables([DBImage]) + self.patcher = patch('testcloud.sql.data_dir_changed', return_value=None) + self.mocked_method = self.patcher.start() + + def teardown_method(self, method): + self.patcher.stop() + DB.drop_tables([DBImage]) + DB.close() + + def test_download_fallback_on_failure(self, monkeypatch): + primary = "https://mirror1.example.com/image.qcow2" + fallback = "https://mirror2.example.com/image.qcow2" + + stub_download = mock.Mock(side_effect=[ + exceptions.TestcloudImageError, + exceptions.TestcloudImageError, + exceptions.TestcloudImageError, + None, + ]) + monkeypatch.setattr(image.Image, "_download_remote_image", stub_download) + monkeypatch.setattr("os.path.exists", lambda p: False) + + img = image.Image(primary) + img.fallback_urls = [fallback] + img.download() + + assert any(fallback in str(c) for c in stub_download.call_args_list) + + def test_download_no_fallback_on_success(self, monkeypatch): + primary = "https://mirror1.example.com/image.qcow2" + fallback = "https://mirror2.example.com/image.qcow2" + + stub_download = mock.Mock(return_value=None) + monkeypatch.setattr(image.Image, "_download_remote_image", stub_download) + monkeypatch.setattr("os.path.exists", lambda p: False) + + img = image.Image(primary) + img.fallback_urls = [fallback] + img.download() + + assert stub_download.call_count == 1 + assert primary in str(stub_download.call_args_list[0]) + + def test_download_all_mirrors_fail(self, monkeypatch): + primary = "https://mirror1.example.com/image.qcow2" + fallback = "https://mirror2.example.com/image.qcow2" + + stub_download = mock.Mock(side_effect=exceptions.TestcloudImageError) + monkeypatch.setattr(image.Image, "_download_remote_image", stub_download) + monkeypatch.setattr("os.path.exists", lambda p: False) + + img = image.Image(primary) + img.fallback_urls = [fallback] + with pytest.raises(exceptions.TestcloudImageError, match="all mirrors"): + img.download() + + # def test_save_pristine(self): # pass # diff --git a/testcloud/cli.py b/testcloud/cli.py index 3673cd4..05979ec 100644 --- a/testcloud/cli.py +++ b/testcloud/cli.py @@ -271,6 +271,13 @@ def _generate_name(): return name +def _unpack_image_url(url_result): + """Unpack image URL result into primary URL and fallback list.""" + if isinstance(url_result, list): + return url_result[0], url_result[1:] + return url_result, [] + + def _download_image(args): if not args.url: log.error("Url wasn't specified.") @@ -278,18 +285,25 @@ def _download_image(args): try: # FIXME maybe change to smth like ... if args.url[:4] not in ["http", "file"] - url = get_image_url(args.url, arch=args.arch) if not any([prot in args.url for prot in ["http", "file"]]) else args.url + url_result = get_image_url(args.url, arch=args.arch) if not any([prot in args.url for prot in ["http", "file"]]) else args.url + url, fallback_urls = _unpack_image_url(url_result) except TestcloudImageError: log.error("Couldn't find the desired image ( %s )..." % args.url) sys.exit(1) - try: - image.Image._download_remote_image(url, os.path.join(args.dest_path, os.path.basename(urlparse(url).path))) - except TestcloudImageError: - log.error("Couldn't download the requested image due to an error.") - sys.exit(1) - except TestcloudPermissionsError: - log.error("Couldn't write to the requested target location ( %s )." % args.dest_path) + dest_path = os.path.join(args.dest_path, os.path.basename(urlparse(url).path)) + for url_idx, download_url in enumerate([url] + fallback_urls): + try: + image.Image._download_remote_image(download_url, dest_path) + break + except TestcloudImageError: + if url_idx < len(fallback_urls): + log.warning("Download from %s failed, trying next mirror...", download_url) + continue + log.error("Couldn't download the requested image due to an error.") + sys.exit(1) + except TestcloudPermissionsError: + log.error("Couldn't write to the requested target location ( %s )." % args.dest_path) def _create_instance(args): @@ -326,7 +340,8 @@ def _create_instance(args): sys.exit(1) try: - url = get_image_url(args.url, arch=args.arch) if not any([prot in args.url for prot in ["http", "file"]]) else args.url + url_result = get_image_url(args.url, arch=args.arch) if not any([prot in args.url for prot in ["http", "file"]]) else args.url + url, fallback_urls = _unpack_image_url(url_result) assert url except (TestcloudImageError, AssertionError): log.error("Couldn't find the desired image ( %s )..." % args.url) @@ -350,6 +365,7 @@ def _create_instance(args): sys.exit(1) tc_image = image.Image(url) + tc_image.fallback_urls = fallback_urls try: tc_image.prepare() except TestcloudPermissionsError as error: diff --git a/testcloud/config.py b/testcloud/config.py index 16a2d63..a13305f 100644 --- a/testcloud/config.py +++ b/testcloud/config.py @@ -275,7 +275,10 @@ def DATA_DIR(self, value): DEBIAN_RELEASE_MAP = {"10": "buster", "11": "bullseye", "12": "bookworm"} DEBIAN_LATEST = "12" - DEBIAN_IMG_URL = "https://cloud.debian.org/images/cloud/%s/daily/latest/debian-%s-genericcloud-%s-daily.qcow2" + DEBIAN_IMG_URL = [ + "https://cdimage.debian.org/images/cloud/%s/daily/latest/debian-%s-genericcloud-%s-daily.qcow2", + "https://cloud.debian.org/images/cloud/%s/daily/latest/debian-%s-genericcloud-%s-daily.qcow2", + ] UBUNTU_RELEASES_API = "https://api.launchpad.net/devel/ubuntu/series" UBUNTU_IMG_URL = "https://cloud-images.ubuntu.com/%s/current/%s-server-cloudimg-%s.img" diff --git a/testcloud/distro_utils/debian.py b/testcloud/distro_utils/debian.py index 171f176..8086c4a 100644 --- a/testcloud/distro_utils/debian.py +++ b/testcloud/distro_utils/debian.py @@ -12,33 +12,39 @@ config_data = config.get_config() -def get_debian_image_url(version: str, arch: str) -> str: +def get_debian_image_url(version: str, arch: str) -> list[str]: arch_map = {"x86_64": "amd64", "aarch64": "arm64", "ppc64le": "ppc64el"} if arch not in arch_map: log.error("Requested architecture is not supported by testcloud for Debian.") raise exceptions.TestcloudImageError + url_templates = config_data.DEBIAN_IMG_URL + if isinstance(url_templates, str): + url_templates = [url_templates] + if arch != "x86_64": - config_data.DEBIAN_IMG_URL = config_data.DEBIAN_IMG_URL.replace("genericcloud", "generic") + url_templates = [t.replace("genericcloud", "generic") for t in url_templates] inverted_releases = {v: k for k, v in config_data.DEBIAN_RELEASE_MAP.items()} if version == "latest": - return config_data.DEBIAN_IMG_URL % ( + args = ( config_data.DEBIAN_RELEASE_MAP[config_data.DEBIAN_LATEST], config_data.DEBIAN_LATEST, arch_map[arch], ) elif version == "sid": - return config_data.DEBIAN_IMG_URL % (version, version, arch_map[arch]) + args = (version, version, arch_map[arch]) elif version in config_data.DEBIAN_RELEASE_MAP: - return config_data.DEBIAN_IMG_URL % (config_data.DEBIAN_RELEASE_MAP[version], version, arch_map[arch]) - elif version in config_data.DEBIAN_RELEASE_MAP.values(): - return config_data.DEBIAN_IMG_URL % (version, inverted_releases[version], arch_map[arch]) + args = (config_data.DEBIAN_RELEASE_MAP[version], version, arch_map[arch]) + elif version in inverted_releases: + args = (version, inverted_releases[version], arch_map[arch]) else: log.error( "Unknown Debian release, valid releases are: " "latest, %s, %s" % (", ".join(config_data.DEBIAN_RELEASE_MAP), ", ".join(inverted_releases)) ) raise exceptions.TestcloudImageError + + return [t % args for t in url_templates] diff --git a/testcloud/image.py b/testcloud/image.py index 28070e4..ba00d6b 100644 --- a/testcloud/image.py +++ b/testcloud/image.py @@ -88,6 +88,8 @@ def __init__(self, uri: str): status = "ready" self.sqldata = DBImage.create(name=uri_data["name"], status=status, remote_path=uri, local_path=local_path) + self.fallback_urls = [] + @property def name(self): return self.sqldata.name @@ -301,15 +303,27 @@ def download(self): raise TestcloudPermissionsError("Problem writing to {}. Are you in group testcloud?".format(self.local_path)) from None elif rpls.startswith("http://") or rpls.startswith("https://"): - retries = 0 - while True: - try: - Image._download_remote_image(self.remote_path, raw_local_path, self._download_callback) + urls_to_try = [self.remote_path] + self.fallback_urls + for url_idx, download_url in enumerate(urls_to_try): + retries = 0 + success = False + while retries <= config_data.DOWNLOAD_RETRIES: + try: + Image._download_remote_image(download_url, raw_local_path, self._download_callback) + success = True + break + except TestcloudImageError: + retries += 1 + if success: break - except TestcloudImageError: - retries += 1 - if retries > config_data.DOWNLOAD_RETRIES: - raise TestcloudImageError("Image download failed after %d attempts." % retries) + if url_idx < len(urls_to_try) - 1: + log.warning( + "Download from %s failed after %d attempts, " + "trying next mirror...", download_url, retries) + else: + raise TestcloudImageError( + "Image download failed from all mirrors after %d attempts each." + % (config_data.DOWNLOAD_RETRIES + 1)) else: raise TestcloudImageError("Testcloud only supports file, http and https URLs") diff --git a/testcloud/util.py b/testcloud/util.py index fcdcd2a..72d71bf 100644 --- a/testcloud/util.py +++ b/testcloud/util.py @@ -167,7 +167,12 @@ def get_image_url(distro_str: str, arch="x86_64", verify=False, additional_handl for _, distro in MERGED_HANDLES.items(): match = re.match(distro["re"], distro_str) if match: - return (verify_url if verify else lambda x: x)(distro["fn"](version=match.group(3) or "latest", arch=arch)) + result = distro["fn"](version=match.group(3) or "latest", arch=arch) + if isinstance(result, list): + if verify: + verify_url(result[0]) + return result + return (verify_url if verify else lambda x: x)(result) log.error("Invalid url handle (distro or distro{-,:}version) passed, supported handles are: %s" % HELP_LIST) raise exceptions.TestcloudImageError