chores<test>Improve integration test pipelines: add inputs, outputs, and multi-task chaining Fixes #12017

.gitignore

@@ -93,3 +93,4 @@ kubeconfig_dev-pipelines-api
 backend/Dockerfile.driver-debug
 
 backend/src/crd/kubernetes/bin
+venv/

backend/test/v2/resources/arguments-parameters.py

@@ -16,15 +16,23 @@
 
 
 @dsl.container_component
-def echo(param1: str, param2: str):
+def echo(param1: str, param2: str, output: dsl.OutputPath(str)):
     return dsl.ContainerSpec(
         image='public.ecr.aws/docker/library/python:3.12',
-        command=['echo'],
-        args=[f'{param1}-{param2}'],
+        command=['sh', '-c'],
+        args=[f'echo {param1}-{param2} > {output}'],
     )
 
+
+@dsl.pipeline
+def arguments_pipeline(param1: str, param2: str) -> str:
+    first = echo(param1=param1, param2=param2)
+    second = echo(param1=first.outputs['output'], param2=param2)
+    return second.outputs['output']
+
+
 if __name__ == '__main__':
     compiler.Compiler().compile(
-        echo,
+        arguments_pipeline,
         package_path=__file__.replace('.py', '.yaml'),
-        pipeline_parameters={'param1': 'hello'})
+        pipeline_parameters={'param1': 'hello', 'param2': 'world'})

backend/test/v2/resources/arguments-parameters.yaml

@@ -1,8 +1,10 @@
 # PIPELINE DEFINITION
-# Name: echo
+# Name: arguments-pipeline
 # Inputs:
 #    param1: str [Default: 'hello']
-#    param2: str
+#    param2: str [Default: 'world']
+# Outputs:
+#    Output: str
 components:
   comp-echo:
     executorLabel: exec-echo
@@ -12,19 +14,52 @@ components:
           parameterType: STRING
         param2:
           parameterType: STRING
+    outputDefinitions:
+      parameters:
+        output:
+          parameterType: STRING
+  comp-echo-2:
+    executorLabel: exec-echo-2
+    inputDefinitions:
+      parameters:
+        param1:
+          parameterType: STRING
+        param2:
+          parameterType: STRING
+    outputDefinitions:
+      parameters:
+        output:
+          parameterType: STRING
 deploymentSpec:
   executors:
     exec-echo:
       container:
         args:
-        - '{{$.inputs.parameters[''param1'']}}-{{$.inputs.parameters[''param2'']}}'
+        - echo {{$.inputs.parameters['param1']}}-{{$.inputs.parameters['param2']}}
+          > {{$.outputs.parameters['output'].output_file}}
         command:
-        - echo
+        - sh
+        - -c
+        image: public.ecr.aws/docker/library/python:3.12
+    exec-echo-2:
+      container:
+        args:
+        - echo {{$.inputs.parameters['param1']}}-{{$.inputs.parameters['param2']}}
+          > {{$.outputs.parameters['output'].output_file}}
+        command:
+        - sh
+        - -c
         image: public.ecr.aws/docker/library/python:3.12
 pipelineInfo:
-  name: echo
+  name: arguments-pipeline
 root:
   dag:
+    outputs:
+      parameters:
+        Output:
+          valueFromParameter:
+            outputParameterKey: output
+            producerSubtask: echo-2
     tasks:
       echo:
         cachingOptions:
@@ -39,12 +74,34 @@ root:
               componentInputParameter: param2
         taskInfo:
           name: echo
+      echo-2:
+        cachingOptions:
+          enableCache: true
+        componentRef:
+          name: comp-echo-2
+        dependentTasks:
+        - echo
+        inputs:
+          parameters:
+            param1:
+              taskOutputParameter:
+                outputParameterKey: output
+                producerTask: echo
+            param2:
+              componentInputParameter: param2
+        taskInfo:
+          name: echo-2
   inputDefinitions:
     parameters:
       param1:
         defaultValue: hello
         parameterType: STRING
       param2:
+        defaultValue: world
+        parameterType: STRING
+  outputDefinitions:
+    parameters:
+      Output:
         parameterType: STRING
 schemaVersion: 2.1.0
 sdkVersion: kfp-2.13.0

backend/test/v2/resources/env-var.py

@@ -13,8 +13,15 @@ def comp(env_var: str) -> str:
     return value
 
 
+@dsl.component(base_image="public.ecr.aws/docker/library/python:3.12")
+def process_value(value: str) -> str:
+    # For demonstration, just append a string
+    return f"Processed: {value}"
+
+
 @dsl.pipeline
 def test_env_exists(env_var: str) -> str:
     comp_task = comp(env_var=env_var)
     comp_task.set_caching_options(False)
-    return comp_task.output
+    process_task = process_value(value=comp_task.output)
+    return process_task.output

backend/test/v2/resources/hello-world-with-returning-component.py

@@ -7,6 +7,14 @@ def comp(message: str) -> str:
     return message
 
 
+@dsl.component(base_image="public.ecr.aws/docker/library/python:3.12")
+def process_message(msg: str) -> str:
+    # For demonstration, just append a string
+    return f"Processed: {msg}"
+
+
 @dsl.pipeline
 def my_pipeline(message: str) -> str:
-    return comp(message=message).output
+    first = comp(message=message)
+    second = process_message(msg=first.output)
+    return second.output

backend/test/v2/resources/hello-world.py

@@ -16,14 +16,21 @@
 
 
 @dsl.container_component
-def echo():
+def echo(message: str, output: dsl.OutputPath(str)):
     return dsl.ContainerSpec(
         image='public.ecr.aws/docker/library/python:3.12',
-        command=['echo'],
-        args=['hello world'],
+        command=['sh', '-c'],
+        args=[f'echo {message} > {output}'],
     )
 
 
+@dsl.pipeline(name='hello-world-pipeline')
+def hello_world_pipeline(input_message: str) -> str:
+    first = echo(message=input_message)
+    second = echo(message=first.outputs['output'])
+    return second.outputs['output']
+
+
 if __name__ == '__main__':
     compiler.Compiler().compile(
-        echo, package_path=__file__.replace('.py', '.yaml'))
+        hello_world_pipeline, package_path=__file__.replace('.py', '.yaml'))

backend/test/v2/resources/hello-world.yaml

@@ -1,28 +1,92 @@
 # PIPELINE DEFINITION
-# Name: echo
+# Name: hello-world-pipeline
+# Inputs:
+#    input_message: str
+# Outputs:
+#    Output: str
 components:
   comp-echo:
     executorLabel: exec-echo
+    inputDefinitions:
+      parameters:
+        message:
+          parameterType: STRING
+    outputDefinitions:
+      parameters:
+        output:
+          parameterType: STRING
+  comp-echo-2:
+    executorLabel: exec-echo-2
+    inputDefinitions:
+      parameters:
+        message:
+          parameterType: STRING
+    outputDefinitions:
+      parameters:
+        output:
+          parameterType: STRING
 deploymentSpec:
   executors:
     exec-echo:
       container:
         args:
-        - hello world
+        - echo {{$.inputs.parameters['message']}} > {{$.outputs.parameters['output'].output_file}}
         command:
-        - echo
+        - sh
+        - -c
+        image: public.ecr.aws/docker/library/python:3.12
+    exec-echo-2:
+      container:
+        args:
+        - echo {{$.inputs.parameters['message']}} > {{$.outputs.parameters['output'].output_file}}
+        command:
+        - sh
+        - -c
         image: public.ecr.aws/docker/library/python:3.12
 pipelineInfo:
-  name: echo
+  name: hello-world-pipeline
 root:
   dag:
+    outputs:
+      parameters:
+        Output:
+          valueFromParameter:
+            outputParameterKey: output
+            producerSubtask: echo-2
     tasks:
       echo:
         cachingOptions:
           enableCache: true
         componentRef:
           name: comp-echo
+        inputs:
+          parameters:
+            message:
+              componentInputParameter: input_message
         taskInfo:
           name: echo
+      echo-2:
+        cachingOptions:
+          enableCache: true
+        componentRef:
+          name: comp-echo-2
+        dependentTasks:
+        - echo
+        inputs:
+          parameters:
+            message:
+              taskOutputParameter:
+                outputParameterKey: output
+                producerTask: echo
+        taskInfo:
+          name: echo-2
+  inputDefinitions:
+    parameters:
+      input_message:
+        parameterType: STRING
+  outputDefinitions:
+    parameters:
+      Output:
+        parameterType: STRING
 schemaVersion: 2.1.0
 sdkVersion: kfp-2.13.0

backend/test/v2/resources/long-running.py

@@ -16,18 +16,23 @@
 
 
 @dsl.container_component
-def wait_op():
+def wait_op(
+    duration: int, message: str, output: dsl.OutputPath(str)
+):
     return dsl.ContainerSpec(
         image='alpine:latest',
         command=['sh', '-c'],
-        args=['echo step-1 sleeping for 5m; sleep 300; echo done1'],
+        args=[
+            f'echo {message} sleeping for {duration}s; sleep {duration}; echo done > {output}'
+        ],
     )
 
 
 @dsl.pipeline
-def wait_awhile():
-    task1 = wait_op()
-    task2 = wait_op().after(task1)
+def wait_awhile(duration: int = 300) -> str:
+    task1 = wait_op(duration=duration, message='step-1',)
+    task2 = wait_op(duration=duration, message=task1.outputs['output'])
+    return task2.outputs['output']
 
 
 if __name__ == '__main__':