[WIP] [BUG] At least one of default_image or images must be provided in ImageConfig

flytekit/configuration/__init__.py

@@ -270,6 +270,15 @@ class ImageConfig(DataClassJsonMixin):
     default_image: Optional[Image] = None
     images: Optional[List[Image]] = None
 
+    def __post_init__(self) -> None:
+        """Ensure that at least one of 'default_image' or 'images' is provided.
+
+        If both are missing, an error is raised to prevent invalid configuration.
+        For more details, see https://github.com/flyteorg/flyte/issues/6157.
+        """
+        if self.default_image is None and self.images is None:
+            raise ValueError("Either 'default_image' or 'images' must be provided.")
+
     def find_image(self, name) -> Optional[Image]:
         """
         Return an image, by name, if it exists.

flytekit/core/python_auto_container.py

@@ -449,6 +449,8 @@ def get_registerable_container_image(img: Optional[Union[str, ImageSpec]], cfg:
             else:
                 raise AssertionError(f"Only fqn and version are supported replacements, {attr} is not supported")
         return img
+    # At least one of 'default_image' or 'images' must be provided
+    # Could we remove the following check?
     if cfg.default_image is None:
         raise ValueError("An image is required for PythonAutoContainer tasks")
     return cfg.default_image.full

flytekit/extend/backend/base_agent.py

@@ -283,7 +283,9 @@ def execute(self: PythonTask, **kwargs) -> LiteralMap:
         from flytekit.tools.translator import get_serializable
 
         ctx = FlyteContext.current_context()
-        ss = ctx.serialization_settings or SerializationSettings(ImageConfig())
+        # If ctx.serialization_settings is None, a default image must be provided.
+        # For more details, see https://github.com/flyteorg/flyte/issues/6157.
+        ss = ctx.serialization_settings or SerializationSettings(ImageConfig.auto_default_image())
         task_template = get_serializable(OrderedDict(), ss, self).template
         output_prefix = ctx.file_access.get_random_remote_directory()
 
@@ -329,7 +331,9 @@ class AsyncAgentExecutorMixin:
 
     def execute(self: PythonTask, **kwargs) -> LiteralMap:
         ctx = FlyteContext.current_context()
-        ss = ctx.serialization_settings or SerializationSettings(ImageConfig())
+        # If ctx.serialization_settings is None, a default image must be provided.
+        # For more details, see https://github.com/flyteorg/flyte/issues/6157.
+        ss = ctx.serialization_settings or SerializationSettings(ImageConfig.auto_default_image())
         output_prefix = ctx.file_access.get_random_remote_directory()
         self.resource_meta = None
 

tests/flytekit/unit/core/test_python_function_task.py

@@ -115,11 +115,10 @@ def test_container_image_conversion(mock_image_spec_builder):
     assert get_registerable_container_image(image_spec, cfg) == image_spec.image_name()
 
 
-def test_get_registerable_container_image_no_images():
-    cfg = ImageConfig()
-
-    with pytest.raises(ValueError):
-        get_registerable_container_image("", cfg)
+def test_empty_image_config():
+    ERR_MSG = "Either 'default_image' or 'images' must be provided."
+    with pytest.raises(ValueError, match=ERR_MSG):
+        cfg = ImageConfig()
 
 
 def test_py_func_task_get_container():