#1862 Move detecting image and type into a separate function
Opened 5 months ago by lsedlar. Modified 5 months ago
lsedlar/pungi deduplicate-helpers  into  master

@@ -48,6 +48,36 @@ 

  }

  

  

+ def find_type_and_format(path, specific_extensions=None):

+     """

+     Find image type and format based on file path. The values should be

+     suitable for using in the productmd metadata.

+ 

+     Args:

+         path (str): Path to the image file

+         specific_extensions (list): List of (type, suffixes, format) tuples for

+             specific mappings. If provided, these are checked first before

+             falling back to the global EXTENSIONS mapping.

+ 

+     Returns:

+         tuple: (type, format) where type is the image type and format is the

+                file format/extension. Returns (None, None) if no match found.

+     """

+     # Check specific extensions first if provided

+     if specific_extensions:

+         for type_, suffixes, format_ in specific_extensions:

+             if any(path.endswith(suffix) for suffix in suffixes):

+                 return type_, format_

+ 

+     # Fall back to global extensions mapping

+     for type_, suffixes in EXTENSIONS.items():

+         for suffix in suffixes:

+             if path.endswith(suffix):

+                 return type_, suffix

+ 

+     return None, None

+ 

+ 

  class ImageBuildPhase(

      base.PhaseLoggerMixin, base.ImageConfigMixin, base.ConfigGuardedPhase

  ):

file modified
+2 -17
@@ -9,7 +9,7 @@ 

  from .. import util

  from ..linker import Linker

  from ..wrappers import kojiwrapper

- from .image_build import EXTENSIONS

+ from .image_build import find_type_and_format

  from ..threading import TelemetryWorkerThread as WorkerThread

  

  
@@ -193,7 +193,7 @@ 

  

          for arch, paths in paths.items():

              for path in paths:

-                 type_, format_ = _find_type_and_format(path)

+                 type_, format_ = find_type_and_format(path, IMAGEBUILDEREXTENSIONS)

                  if not format_:

                      # Path doesn't match any known type.

                      continue
@@ -246,18 +246,3 @@ 

                  compose.im.add(variant=variant.uid, arch=arch, image=img)

  

          self.pool.log_info("[DONE ] %s (task id: %s)" % (msg, task_id))

- 

- 

- def _find_type_and_format(path):

-     # these are our image-builder-exclusive mappings for images whose extensions

-     # aren't quite the same as imagefactory. they come first as we

-     # want our oci.tar.xz mapping to win over the tar.xz one in

-     # EXTENSIONS

-     for type_, suffixes, format_ in IMAGEBUILDEREXTENSIONS:

-         if any(path.endswith(suffix) for suffix in suffixes):

-             return type_, format_

-     for type_, suffixes in EXTENSIONS.items():

-         for suffix in suffixes:

-             if path.endswith(suffix):

-                 return type_, suffix

-     return None, None

file modified
+2 -17
@@ -9,7 +9,7 @@ 

  from .. import util

  from ..linker import Linker

  from ..wrappers import kojiwrapper

- from .image_build import EXTENSIONS

+ from .image_build import find_type_and_format

  from ..threading import TelemetryWorkerThread as WorkerThread

  

  KIWIEXTENSIONS = [
@@ -193,7 +193,7 @@ 

  

          for arch, paths in paths.items():

              for path in paths:

-                 type_, format_ = _find_type_and_format(path)

+                 type_, format_ = find_type_and_format(path, KIWIEXTENSIONS)

                  if not format_:

                      # Path doesn't match any known type.

                      continue
@@ -246,18 +246,3 @@ 

                  compose.im.add(variant=variant.uid, arch=arch, image=img)

  

          self.pool.log_info("[DONE ] %s (task id: %s)" % (msg, task_id))

- 

- 

- def _find_type_and_format(path):

-     # these are our kiwi-exclusive mappings for images whose extensions

-     # aren't quite the same as imagefactory. they come first as we

-     # want our oci.tar.xz mapping to win over the tar.xz one in

-     # EXTENSIONS

-     for type_, suffixes, format_ in KIWIEXTENSIONS:

-         if any(path.endswith(suffix) for suffix in suffixes):

-             return type_, format_

-     for type_, suffixes in EXTENSIONS.items():

-         for suffix in suffixes:

-             if path.endswith(suffix):

-                 return type_, suffix

-     return None, None

file modified
+1
@@ -1,3 +1,4 @@ 

  parameterized

  pytest

  pytest-cov

+ pytest-subtests

file modified
+89 -1
@@ -3,7 +3,13 @@ 

  from unittest import mock

  import os

  

- from pungi.phases.image_build import ImageBuildPhase, CreateImageBuildThread

+ import pytest

+ 

+ from pungi.phases.image_build import (

+     ImageBuildPhase,

+     CreateImageBuildThread,

+     find_type_and_format,

+ )

  from tests.helpers import DummyCompose, PungiTestCase, boom

  

  
@@ -1098,3 +1104,85 @@ 

          with self.assertRaises(RuntimeError):

              with mock.patch("time.sleep"):

                  t.process((compose, cmd, None), 1)

+ 

+ 

+ @pytest.mark.parametrize(

+     "path,expected_type,expected_format",

+     [

+         ("test-image.qcow2", "qcow2", "qcow2"),

+         ("test-image.tar.xz", "docker", "tar.xz"),

+         ("test-image.vhd", "vpc", "vhd"),

+         ("test-image.unknown", None, None),

+         # Test edge cases like empty paths, paths with dots, etc.

+         ("", None, None),

+         (".", None, None),

+         ("test-image.", None, None),

+         ("test.image.with.dots.qcow2", "qcow2", "qcow2"),

+         (".test-image.qcow2.", None, None),

+         ("qcow2", "qcow2", "qcow2"),

+         ("test-image.qcow2.", None, None),

+         (".qcow2", "qcow2", "qcow2"),

+     ],

+ )

+ def test_find_type_and_format_with_global_extensions(

+     path, expected_type, expected_format

+ ):

+     """Test finding type and format using global EXTENSIONS mapping."""

+     result = find_type_and_format(path)

+     assert result == (expected_type, expected_format)

+ 

+ 

+ @pytest.mark.parametrize(

+     "path,specific_extensions,expected_type,expected_format",

+     [

+         (

+             "test-image.vhdfixed.xz",

+             [("vhd-compressed", ["vhdfixed.xz"], "vhd.xz")],

+             "vhd-compressed",

+             "vhd.xz",

+         ),

+         (

+             "test-image.multi1",

+             [("multi-format", ["multi1", "multi2"], "multi.format")],

+             "multi-format",

+             "multi.format",

+         ),

+         (

+             "test-image.multi2",

+             [("multi-format", ["multi1", "multi2"], "multi.format")],

+             "multi-format",

+             "multi.format",

+         ),

+         # Test fallback to global EXTENSIONS when specific extensions don't

+         # match.

+         (

+             "test-image.qcow2",

+             [("custom-only", ["custom.only"], "custom.only")],

+             "qcow2",

+             "qcow2",

+         ),

+         ("test-image.qcow2", [], "qcow2", "qcow2"),

+         ("test-image.qcow2", None, "qcow2", "qcow2"),

+         # Test that specific extensions take priority over global ones.

+         (

+             "test-image.tar.xz",

+             [("custom-docker", ["tar.xz"], "custom-tar.xz")],

+             "custom-docker",

+             "custom-tar.xz",

+         ),

+         # Test behavior when no extensions match.

+         (

+             "test-image.unknown",

+             [("custom-only", ["custom.only"], "custom.only")],

+             None,

+             None,

+         ),

+         ("test-image.nonexistent", [], None, None),

+     ],

+ )

+ def test_find_type_and_format_with_specific_extensions(

+     path, specific_extensions, expected_type, expected_format

+ ):

+     """Test finding type and format with specific extensions parameter."""

+     result = find_type_and_format(path, specific_extensions)

+     assert result == (expected_type, expected_format)

The logic was duplicated in two locations, this patch makes it a common function.


There are a few things I'm not excited about here:

  • The location of the new function is not great. The module where it is defined doesn't actually use it. But it can not be in the existing utils module as that would cause circular imports. And a new utils module doesn't seem better than this.
  • The tests are maybe a little excessive?
  • The tests are not unittest anymore, but pure pytest. We use pytest as a runner anyway, so it's not a new dependency, but the suite will not work with unittest runner anymore if someone was doing that. It could be rewritted with unittest.subTest, but that doesn't play great with pytest (unless we install pytest-subtests).

CC @supakeen

Also, I will admit here to using Cursor to aid with creating this change.