resolve python 3.12 distutils issue

.python-version

@@ -0,0 +1 @@
+3.12.1

sdk/python/kfp/dsl/types/type_utils.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 """Utilities for component I/O type mapping."""
 
-from distutils import util
 import inspect
 import json
 from typing import Any, Callable, Dict, Optional, Type, Union
@@ -71,9 +70,27 @@
 }
 
 
+# copied from distutils.util, which was removed in Python 3.12
+# https://github.com/pypa/distutils/blob/fb5c5704962cd3f40c69955437da9a88f4b28567/distutils/util.py#L340-L353
+def strtobool(val):
+    """Convert a string representation of truth to true (1) or false (0).
+
+    True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values
+    are 'n', 'no', 'f', 'false', 'off', and '0'.  Raises ValueError if
+    'val' is anything else.
+    """
+    val = val.lower()
+    if val in ('y', 'yes', 't', 'true', 'on', '1'):
+        return 1
+    elif val in ('n', 'no', 'f', 'false', 'off', '0'):
+        return 0
+    else:
+        raise ValueError('invalid truth value %r' % (val,))
+
+
 def bool_cast_fn(default: Union[str, bool]) -> bool:
     if isinstance(default, str):
-        default = util.strtobool(default) == 1
+        default = strtobool(default) == 1
     return default
 
 

sdk/python/kfp/local/pipeline_orchestrator_test.py

@@ -28,57 +28,15 @@
 from kfp.local import testing_utilities
 
 
-@dsl.component
-def identity(string: str) -> str:
-    return string
-
-
-@dsl.component
-def pass_op():
-    pass
-
-
-@dsl.component
-def many_parameter_component(
-    a_float: float,
-    a_boolean: bool,
-    a_dict: dict,
-    a_string: str = 'default',
-    an_integer: int = 12,
-    a_list: list = ['item1', 'item2'],
-) -> NamedTuple(
-        'outputs',
-        a_string=str,
-        a_float=float,
-        an_integer=int,
-        a_boolean=bool,
-        a_list=list,
-        a_dict=dict,
-):
-    outputs = NamedTuple(
-        'outputs',
-        a_string=str,
-        a_float=float,
-        an_integer=int,
-        a_boolean=bool,
-        a_list=list,
-        a_dict=dict,
-    )
-    return outputs(
-        a_string=a_string,
-        a_float=a_float,
-        an_integer=an_integer,
-        a_boolean=a_boolean,
-        a_list=a_list,
-        a_dict=a_dict,
-    )
-
-
 class TestRunPipelineSubprocessRunner(
-        testing_utilities.LocalRunnerEnvironmentTestCase, unittest.TestCase):
+        testing_utilities.LocalRunnerEnvironmentTestCase):
 
     def test_must_initialize(self):
 
+        @dsl.component
+        def identity(string: str) -> str:
+            return string
+
         @dsl.pipeline
         def my_pipeline():
             identity(string='foo')
@@ -92,6 +50,10 @@ def my_pipeline():
     def test_no_io(self):
         local.init(local.SubprocessRunner())
 
+        @dsl.component
+        def pass_op():
+            pass
+
         @dsl.pipeline
         def my_pipeline():
             pass_op()
@@ -104,6 +66,10 @@ def my_pipeline():
     def test_missing_args(self):
         local.init(local.SubprocessRunner())
 
+        @dsl.component
+        def identity(string: str) -> str:
+            return string
+
         @dsl.pipeline
         def my_pipeline(string: str) -> str:
             t1 = identity(string=string)
@@ -118,6 +84,10 @@ def my_pipeline(string: str) -> str:
     def test_simple_return(self):
         local.init(local.SubprocessRunner())
 
+        @dsl.component
+        def identity(string: str) -> str:
+            return string
+
         @dsl.pipeline
         def my_pipeline(string: str = 'text') -> str:
             t1 = identity(string=string)
@@ -137,6 +107,41 @@ def test_all_param_io(self):
         # - passing args from pipeline param, upstream output, and constant
         # - pipeline surfacing outputs from last run component and not last run component
 
+        @dsl.component
+        def many_parameter_component(
+            a_float: float,
+            a_boolean: bool,
+            a_dict: dict,
+            a_string: str = 'default',
+            an_integer: int = 12,
+            a_list: list = ['item1', 'item2'],
+        ) -> NamedTuple(
+                'outputs',
+                a_string=str,
+                a_float=float,
+                an_integer=int,
+                a_boolean=bool,
+                a_list=list,
+                a_dict=dict,
+        ):
+            outputs = NamedTuple(
+                'outputs',
+                a_string=str,
+                a_float=float,
+                an_integer=int,
+                a_boolean=bool,
+                a_list=list,
+                a_dict=dict,
+            )
+            return outputs(
+                a_string=a_string,
+                a_float=a_float,
+                an_integer=an_integer,
+                a_boolean=a_boolean,
+                a_list=a_list,
+                a_dict=a_dict,
+            )
+
         @dsl.pipeline
         def my_pipeline(
             flt: float,
@@ -249,6 +254,10 @@ def my_pipeline():
     def test_pipeline_in_pipeline_not_supported(self):
         local.init(local.SubprocessRunner())
 
+        @dsl.component
+        def identity(string: str) -> str:
+            return string
+
         @dsl.pipeline
         def inner_pipeline():
             identity(string='foo')