Add files via upload

hello/hello_world.py

@@ -0,0 +1,29 @@
+import os
+
+from kfp import dsl
+from kfp import compiler
+
+# In tests, we install a KFP package from the PR under test. Users should not
+# normally need to specify `kfp_package_path` in their component definitions.
+_KFP_PACKAGE_PATH = os.getenv('KFP_PACKAGE_PATH')
+
+
+@dsl.component(kfp_package_path=_KFP_PACKAGE_PATH)
+def hello_world(text: str) -> str:
+    print(text)
+    return text
+
+
+@dsl.pipeline(name='hello-world', description='A simple intro pipeline')
+def pipeline_hello_world(text: str = 'hi there'):
+    """Pipeline that passes small pipeline parameter string to consumer op."""
+
+    consume_task = hello_world(
+        text=text)  # Passing pipeline parameter as argument to consumer op
+
+
+if __name__ == "__main__":
+    # execute only if run as a script
+    compiler.Compiler().compile(
+        pipeline_func=pipeline_hello_world,
+        package_path='hello_world_pipeline.json')

hello/hello_world.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+#
+# Copyright 2021 The Kubeflow Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -ex
+
+pushd ./backend/src/v2/test
+
+python3 -m pip install --upgrade pip
+python3 -m pip install -r ./requirements-sample-test.txt
+
+popd
+
+# The -u flag makes python output unbuffered, so that we can see real time log.
+# Reference: https://stackoverflow.com/a/107717
+python3 -u ./hello/sample.py

hello/sample.py

@@ -0,0 +1,49 @@
+import os
+import unittest
+from dataclasses import dataclass
+from pprint import pprint
+from typing import List
+
+import kfp
+from kfp.dsl.graph_component import GraphComponent
+import hello_world
+
+_MINUTE = 60  # seconds
+_DEFAULT_TIMEOUT = 5 * _MINUTE
+
+
+@dataclass
+class TestCase:
+    pipeline_func: GraphComponent
+    timeout: int = _DEFAULT_TIMEOUT
+
+
+class SampleTest(unittest.TestCase):
+    _kfp_host_and_port = os.getenv('KFP_API_HOST_AND_PORT', 'http://localhost:8888')
+    _kfp_ui_and_port = os.getenv('KFP_UI_HOST_AND_PORT', 'http://localhost:8080')
+    _client = kfp.Client(host=_kfp_host_and_port, ui_host=_kfp_ui_and_port)
+
+    def test(self):
+        # 只測試 hello_world 範例
+        test_case = TestCase(pipeline_func=hello_world.pipeline_hello_world)
+
+        # 直接執行測試案例
+        self.run_test_case(test_case.pipeline_func, test_case.timeout)
+
+    def run_test_case(self, pipeline_func: GraphComponent, timeout: int):
+        # 執行指定的管道函數，並等待完成
+        with self.subTest(pipeline=pipeline_func, msg=pipeline_func.name):
+            run_result = self._client.create_run_from_pipeline_func(pipeline_func=pipeline_func)
+            run_response = run_result.wait_for_run_completion(timeout)
+
+            # 打印運行的詳細信息
+            pprint(run_response.run_details)
+            print("Run details page URL:")
+            print(f"{self._kfp_ui_and_port}/#/runs/details/{run_response.run_id}")
+
+            # 檢查運行狀態是否成功
+            self.assertEqual(run_response.state, "SUCCEEDED")
+
+
+if __name__ == '__main__':
+    unittest.main()