WIP disk-image in patc

Assisted-by: Cursor AI
Signed-off-by: Scott Wickersham <swickers@redhat.com>

Author: swickersh

scripts/python/helpers/compress_artifacts.py

@@ -5,7 +5,8 @@
 * Pulls signed macOS and Windows OCI artifacts from Quay into a ``signed/`` directory.
 * Restores supplementary files (readme, license, changelog) that were held during signing.
 * Compresses each file entry into the final deliverable format:
-  - macOS / Linux → ``.tar.gz`` (from ``os/arch/`` directory)
+  - macOS / Linux (non-disk-image) → ``.tar.gz`` (from ``os/arch/`` directory)
+  - Linux disk images (``.qcow2``, ``.iso``) → copied as-is to ``ready_for_distribution/``
   - Windows → ``.zip`` (from ``os/arch/`` directory, extension corrected from
     ``.tar.gz``/``.tar``)
 * Updates ``SNAPSHOT_JSON`` to reflect corrected Windows filenames in ``files[]``.
@@ -39,6 +40,10 @@
 
 PROG = "compress_artifacts.py"
 
+# File extensions that identify disk images. These files are copied as-is to
+# ready_for_distribution without being wrapped in a tar archive.
+_DISK_IMAGE_EXTENSIONS: frozenset[str] = frozenset({".qcow2", ".iso"})
+
 QUAY_SECRET_MOUNT = Path(os.environ.get("QUAY_SECRET_MOUNT", "/mnt/quaySecret"))
 CONTENT_DIR = Path(os.environ.get("CONTENT_DIR", "/shared/artifacts"))
 SHARED_DIR = Path(os.environ.get("SHARED_DIR", "/shared"))
@@ -111,11 +116,10 @@ def _compress_file_entry(
     the archive is created as a ``.zip`` instead of ``.tar.gz``/``.tar``, and the returned
     source path reflects the corrected filename so the snapshot can be updated accordingly.
 
-    Raises RuntimeError on failure (missing source, unknown OS, or empty arch directory).
+    Disk images (``.qcow2``, ``.iso``) are copied directly to ``ready_dir`` without
+    being wrapped in a tarball, since they are already in their final deliverable format.
 
-    Note: all files are currently compressed regardless of type. ISOs should be
-    passed through as-is rather than wrapped in a tarball — this will need to be
-    addressed before ISO delivery is supported.
+    Raises RuntimeError on failure (missing source, unknown OS, or empty arch directory).
     """
     source = entry.get("source")
     if not source:
@@ -139,12 +143,22 @@ def _compress_file_entry(
 
     # macOS and Linux follow the Unix convention of tar.gz archives; Windows uses zip
     # because that is the standard expected by Windows users and Developer Portal tooling.
+    # Disk images are an exception: they are delivered as-is without any archiving.
     if os_name in ("darwin", "linux"):
         out_path = ready_dir / source_filename
-        with tarfile.open(str(out_path), "w:gz") as tf:
-            for item in sorted(arch_dir.rglob("*")):
-                if item.is_file():
-                    tf.add(str(item), arcname=str(item.relative_to(arch_dir)))
+        if Path(source_filename).suffix.lower() in _DISK_IMAGE_EXTENSIONS:
+            files = [f for f in arch_dir.rglob("*") if f.is_file()]
+            if len(files) != 1:
+                raise RuntimeError(
+                    f"Expected exactly one disk image file in {arch_dir}, "
+                    f"found {len(files)}"
+                )
+            shutil.copy2(str(files[0]), str(out_path))
+        else:
+            with tarfile.open(str(out_path), "w:gz") as tf:
+                for item in sorted(arch_dir.rglob("*")):
+                    if item.is_file():
+                        tf.add(str(item), arcname=str(item.relative_to(arch_dir)))
         logger.info("  Created (%s): %s", array_name, source_filename)
         return source
 

scripts/python/helpers/push_unsigned.py

@@ -40,6 +40,10 @@
 SUPPLEMENTARY_NAMES = {"readme", "license", "changelog"}
 SUPPLEMENTARY_EXTS = {".md", ".txt"}
 
+# File extensions that identify disk images. These files are moved directly to
+# the target directory without tar extraction.
+_DISK_IMAGE_EXTENSIONS: frozenset[str] = frozenset({".qcow2", ".iso"})
+
 logger = logging.getLogger(__name__)
 
 
@@ -81,7 +85,11 @@ def move_supplementary_out(src_root: Path, hold_root: Path) -> None:
 
 
 def _unpack_file_entries(entries: list[dict], component_dir: Path, unsigned_dir: Path) -> None:
-    """Extract each archive from entries into its OS/arch subdirectory under unsigned_dir."""
+    """Extract each archive from entries into its OS/arch subdirectory under unsigned_dir.
+
+    Disk images (.qcow2, .iso) are moved directly to the target directory without
+    unpacking, since they are not archives.
+    """
     for entry in entries:
         source = entry.get("source", "")
         os_name = entry.get("os", "")
@@ -103,9 +111,12 @@ def _unpack_file_entries(entries: list[dict], component_dir: Path, unsigned_dir:
             continue
 
         target_dir.mkdir(parents=True, exist_ok=True)
-        with tarfile.open(str(archive_path)) as tf:
-            _safe_extract_archive(tf, target_dir, archive_name)
-        archive_path.unlink()
+        if archive_path.suffix.lower() in _DISK_IMAGE_EXTENSIONS:
+            shutil.move(str(archive_path), str(target_dir / archive_name))
+        else:
+            with tarfile.open(str(archive_path)) as tf:
+                _safe_extract_archive(tf, target_dir, archive_name)
+            archive_path.unlink()
 
 
 def _safe_extract_archive(tf: tarfile.TarFile, target_dir: Path, archive_name: str) -> None:

scripts/python/helpers/test_compress_artifacts.py

@@ -148,6 +148,92 @@ def test_compress_file_entry_windows(tmp_path: Path, monkeypatch: pytest.MonkeyP
     assert result is not None and result.endswith(".zip")
 
 
+def test_compress_file_entry_qcow2_passthrough(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A qcow2 disk image is copied directly to ready_for_distribution without archiving."""
+    monkeypatch.setattr(compress_artifacts, "CONTENT_DIR", tmp_path)
+    comp_dir = tmp_path / "prod"
+    ready_dir = comp_dir / "ready_for_distribution"
+    ready_dir.mkdir(parents=True)
+
+    linux_arch_dir = comp_dir / "linux" / "x86_64"
+    linux_arch_dir.mkdir(parents=True)
+    (linux_arch_dir / "rhel-10.0-x86_64-kvm.qcow2").write_bytes(b"qcow2 content")
+
+    result = compress_artifacts._compress_file_entry(
+        {
+            "source": "/releases/rhel-10.0-x86_64-kvm.qcow2",
+            "os": "linux",
+            "arch": "x86_64",
+        },
+        "files",
+        comp_dir,
+        ready_dir,
+    )
+    out = ready_dir / "rhel-10.0-x86_64-kvm.qcow2"
+    assert out.exists()
+    assert out.read_bytes() == b"qcow2 content"
+    assert not tarfile.is_tarfile(str(out))
+    assert result == "/releases/rhel-10.0-x86_64-kvm.qcow2"
+
+
+def test_compress_file_entry_iso_passthrough(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An iso disk image is copied directly to ready_for_distribution without archiving."""
+    monkeypatch.setattr(compress_artifacts, "CONTENT_DIR", tmp_path)
+    comp_dir = tmp_path / "prod"
+    ready_dir = comp_dir / "ready_for_distribution"
+    ready_dir.mkdir(parents=True)
+
+    linux_arch_dir = comp_dir / "linux" / "x86_64"
+    linux_arch_dir.mkdir(parents=True)
+    (linux_arch_dir / "rhel-10.0-x86_64-boot.iso").write_bytes(b"iso content")
+
+    result = compress_artifacts._compress_file_entry(
+        {
+            "source": "/releases/rhel-10.0-x86_64-boot.iso",
+            "os": "linux",
+            "arch": "x86_64",
+        },
+        "files",
+        comp_dir,
+        ready_dir,
+    )
+    out = ready_dir / "rhel-10.0-x86_64-boot.iso"
+    assert out.exists()
+    assert out.read_bytes() == b"iso content"
+    assert result == "/releases/rhel-10.0-x86_64-boot.iso"
+
+
+def test_compress_file_entry_disk_image_multiple_files_raises(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """RuntimeError is raised when a disk image arch dir contains more than one file."""
+    monkeypatch.setattr(compress_artifacts, "CONTENT_DIR", tmp_path)
+    comp_dir = tmp_path / "prod"
+    ready_dir = comp_dir / "ready_for_distribution"
+    ready_dir.mkdir(parents=True)
+
+    linux_arch_dir = comp_dir / "linux" / "x86_64"
+    linux_arch_dir.mkdir(parents=True)
+    (linux_arch_dir / "rhel-10.0-x86_64-kvm.qcow2").write_bytes(b"a")
+    (linux_arch_dir / "extra.qcow2").write_bytes(b"b")
+
+    with pytest.raises(RuntimeError, match="exactly one disk image"):
+        compress_artifacts._compress_file_entry(
+            {
+                "source": "/releases/rhel-10.0-x86_64-kvm.qcow2",
+                "os": "linux",
+                "arch": "x86_64",
+            },
+            "files",
+            comp_dir,
+            ready_dir,
+        )
+
+
 def test_compress_file_entry_missing_arch_dir_raises(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

scripts/python/helpers/test_push_unsigned.py

@@ -283,6 +283,42 @@ def test_unpack_file_entries_skips_missing_fields(tmp_path: Path) -> None:
     # no crash
 
 
+def test_unpack_file_entries_qcow2_passthrough(tmp_path: Path) -> None:
+    """A qcow2 disk image is moved directly to the OS/arch dir without tar extraction."""
+    comp_dir = tmp_path / "prod"
+    comp_dir.mkdir()
+    disk_image = comp_dir / "rhel-10.0-x86_64-kvm.qcow2"
+    disk_image.write_bytes(b"qcow2 raw content")
+
+    push_unsigned._unpack_file_entries(
+        [{"source": "/releases/rhel-10.0-x86_64-kvm.qcow2", "os": "linux", "arch": "x86_64"}],
+        comp_dir,
+        comp_dir / "unsigned",
+    )
+    dest = comp_dir / "linux" / "x86_64" / "rhel-10.0-x86_64-kvm.qcow2"
+    assert dest.exists()
+    assert dest.read_bytes() == b"qcow2 raw content"
+    assert not disk_image.exists()
+
+
+def test_unpack_file_entries_iso_passthrough(tmp_path: Path) -> None:
+    """An iso disk image is moved directly to the OS/arch dir without tar extraction."""
+    comp_dir = tmp_path / "prod"
+    comp_dir.mkdir()
+    disk_image = comp_dir / "rhel-10.0-x86_64-boot.iso"
+    disk_image.write_bytes(b"iso raw content")
+
+    push_unsigned._unpack_file_entries(
+        [{"source": "/releases/rhel-10.0-x86_64-boot.iso", "os": "linux", "arch": "x86_64"}],
+        comp_dir,
+        comp_dir / "unsigned",
+    )
+    dest = comp_dir / "linux" / "x86_64" / "rhel-10.0-x86_64-boot.iso"
+    assert dest.exists()
+    assert dest.read_bytes() == b"iso raw content"
+    assert not disk_image.exists()
+
+
 def test_unpack_file_entries_rejects_path_traversal(tmp_path: Path) -> None:
     """RuntimeError is raised when a tar entry contains an unsafe path traversal sequence."""
     comp_dir = tmp_path / "prod"