add test for None default parameter

sdk/python/kfp/local/executor_output_utils.py

@@ -49,17 +49,15 @@ def cast_protobuf_numbers(
     struct_pb2.Value to a dict/json, int will be upcast to float, even
     if the component output specifies int.
     """
-    int_output_keys = [
+    output_parameter_types = [
         output_param_name
         for output_param_name, parameter_spec in output_parameter_types.items()
         if parameter_spec.parameter_type ==
         pipeline_spec_pb2.ParameterType.ParameterTypeEnum.NUMBER_INTEGER
     ]
-    for int_output_key in int_output_keys:
-        # avoid KeyError when the user never writes to the dsl.OutputPath
-        if int_output_key in output_parameters:
-            output_parameters[int_output_key] = int(
-                output_parameters[int_output_key])
+    for float_output_key in output_parameter_types:
+        output_parameters[float_output_key] = int(
+            output_parameters[float_output_key])
     return output_parameters
 
 

sdk/python/kfp/local/subprocess_task_handler_test.py

@@ -14,7 +14,7 @@
 """Tests for subprocess_local_task_handler.py."""
 import contextlib
 import io
-from typing import NamedTuple
+from typing import NamedTuple, Optional
 import unittest
 from unittest import mock
 
@@ -417,6 +417,16 @@ def my_comp(out_param: dsl.OutputPath(int)):
         task = my_comp()
         self.assertEmpty(task.outputs)
 
+    def test_optional_param(self):
+        local.init(runner=local.SubprocessRunner(use_venv=True))
+
+        @dsl.component
+        def my_comp(string: Optional[str] = None) -> str:
+            return 'is none' if string is None else 'not none'
+
+        task = my_comp()
+        self.assertEqual(task.output, 'is none')
+
 
 if __name__ == '__main__':
     unittest.main()