Add mirror fallback for Debian cloud image downloads

test/test_debian.py

@@ -0,0 +1,103 @@
+# -*- coding: utf-8 -*-
+# Copyright 2025, Red Hat, Inc.
+# License: GPL-2.0+ <http://spdx.org/licenses/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

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
 #

testcloud/cli.py

@@ -271,25 +271,39 @@ 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.")
         sys.exit(1)
 
     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:

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"

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]

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")
 

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