Updated the V2 pipeline test to be more robust (#3129)

* Updated the V2 pipeline test  to be more robust

Signed-off-by: kunal-511 <[email protected]>

* Updated the version end to end test

Signed-off-by: kunal-511 <[email protected]>

* added quotes

Signed-off-by: kunal-511 <[email protected]>

* Updated the v2 test to make it use simple approach

Signed-off-by: kunal-511 <[email protected]>

* Updated experiment name

Signed-off-by: kunal-511 <[email protected]>

* Added some logs and simplified the pipeline

Signed-off-by: kunal-511 <[email protected]>

* making the pipeline more minimal

Signed-off-by: kunal-511 <[email protected]>

* removed the logs

Signed-off-by: kunal-511 <[email protected]>

* Updated test name

Signed-off-by: kunal-511 <[email protected]>

* Update pipeline_test.yaml

Signed-off-by: Julius von Kohout <[email protected]>

---------

Signed-off-by: kunal-511 <[email protected]>
Signed-off-by: Julius von Kohout <[email protected]>
Co-authored-by: Julius von Kohout <[email protected]>

Author: kunal-511

.github/workflows/full_kubeflow_integration_test.yaml

@@ -122,14 +122,14 @@ jobs:
       run: |
         pip3 install -U "kfp>=2.13.0"
         TOKEN="$(kubectl -n $KF_PROFILE create token default-editor)"
-        python3 tests/gh-actions/test_pipeline.py run_pipeline "${TOKEN}" "${KF_PROFILE}"
+        python3 tests/gh-actions/test_pipeline_v2.py run_pipeline "${TOKEN}" "${KF_PROFILE}"
 
     - name: Test Pipeline Access with Unauthorized Token
       run: |
         kubectl create namespace test-unauthorized
         kubectl create serviceaccount test-unauthorized -n test-unauthorized
         UNAUTHORIZED_TOKEN=$(kubectl -n test-unauthorized create token test-unauthorized)
-        python3 tests/gh-actions/test_pipeline.py test_unauthorized_access "$UNAUTHORIZED_TOKEN" "${KF_PROFILE}"
+        python3 tests/gh-actions/test_pipeline_v2.py test_unauthorized_access "$UNAUTHORIZED_TOKEN" "${KF_PROFILE}"
 
     - name: Test Volumes Web Application API
       run: ./tests/gh-actions/test_volumes_web_application.sh "${KF_PROFILE}"

.github/workflows/pipeline_test.yaml

@@ -11,8 +11,8 @@ on:
     - common/cert-manager/**
     - common/oauth2-proxy/**
     - common/istio*/**
-    - tests/gh-actions/test_pipeline.py
     - tests/gh-actions/test_pipeline_v1.py
+    - tests/gh-actions/test_pipeline_v2.py
     - experimental/security/PSS/*
 
 env:
@@ -64,6 +64,9 @@ jobs:
         fi
         kubectl get secret mlpipeline-minio-artifact -n "$KF_PROFILE" -o json | jq -r '.data | keys[] as $k | "\($k): \(. | .[$k] | @base64d)"' | tr '\n' ' '
 
+    - name: Wait for All Pods to be Ready
+      run: kubectl wait --for=condition=Ready pods --all --all-namespaces --timeout 60s --field-selector=status.phase!=Succeeded
+
     - name: Port forward
       run: ./tests/gh-actions/port_forward_gateway.sh
 
@@ -75,18 +78,16 @@ jobs:
 
     - name: V2 Pipeline Test
       run: |
-        pip3 install -U kfp==2.13.0
+        pip3 install "kfp>=2.13.0"
         TOKEN="$(kubectl -n $KF_PROFILE create token default-editor)"
-        python3 tests/gh-actions/test_pipeline.py run_pipeline "${TOKEN}" "${KF_PROFILE}"
+        python3 tests/gh-actions/test_pipeline_v2.py run_pipeline "${TOKEN}" "${KF_PROFILE}"
 
-    - name: Fail to list pipelines with unauthorized ServiceAccount Token
+    - name: Test unauthorized access
       run: |
-        pip3 install -U kfp==2.13.0
         TOKEN="$(kubectl -n default create token default)"
-        python3 tests/gh-actions/test_pipeline.py test_unauthorized_access "${TOKEN}" "${KF_PROFILE}"
+        python3 tests/gh-actions/test_pipeline_v2.py test_unauthorized_access "${TOKEN}" "${KF_PROFILE}"
         echo "Test succeeded. Token from unauthorized ServiceAccount cannot list pipelines in $KF_PROFILE namespace."
 
-
     - name: Apply Pod Security Standards baseline levels for static namespaces
       run: ./tests/gh-actions/enable_baseline_PSS.sh
 

tests/gh-actions/test_pipeline.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+
+import kfp
+import sys
+import time
+from kfp import dsl
+from kfp_server_api.exceptions import ApiException
+
+
+@dsl.component
+def hello_world_op() -> str:
+    print("Hello World from Kubeflow Pipelines V2!")
+    return "Hello World"
+
+
+@dsl.pipeline(
+    name="hello-world-v2",
+    description="A very simple hello world pipeline"
+)
+def hello_world_pipeline():
+    hello_world_op()
+
+
+def run_pipeline(token, namespace):
+    client = kfp.Client(host="http://localhost:8080/pipeline", existing_token=token)
+    
+    try:
+        pipelines = client.list_pipelines()
+        print(f"Successfully connected to KFP server, found {len(pipelines.pipelines)} pipelines")
+        
+        experiment = client.create_experiment("v2-pipeline-test", namespace=namespace)
+        print(f"Created experiment: v2-pipeline-test in namespace {namespace}")
+        
+        run = client.create_run_from_pipeline_func(
+            pipeline_func=hello_world_pipeline,
+            experiment_name="v2-pipeline-test",
+            run_name="v2-test-run",
+            arguments={},
+            namespace=namespace
+        )
+        
+        run_id = run.run_id
+        
+        for _ in range(30): 
+            status = client.get_run(run_id=run_id).state
+            
+            if status == "SUCCEEDED":
+                return
+            elif status not in ["PENDING", "RUNNING"]:
+                print(f"Pipeline failed with status: {status}")
+                
+                pods = client._get_k8s_client().list_namespaced_pod(
+                    namespace=namespace,
+                    label_selector=f"pipeline/runid={run_id}"
+                )
+                
+                print(f"Found {len(pods.items)} pods for this run")
+                for pod in pods.items:
+                    print(f"Pod {pod.metadata.name}: {pod.status.phase}")
+                
+                sys.exit(1)
+                
+            time.sleep(10)
+        
+        sys.exit(1)
+        
+    except Exception as exception:
+        print(f"Error in pipeline execution: {exception}")
+        sys.exit(1)
+
+
+def test_unauthorized_access(token, namespace):
+    client = kfp.Client(host="http://localhost:8080/pipeline", existing_token=token)
+
+    try:
+        pipeline = client.list_runs(namespace=namespace)
+        sys.exit(1)
+    except ApiException as exception:
+        if exception.status != 403:
+            sys.exit(1)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 3:
+        sys.exit(1)
+    
+    action = sys.argv[1]
+    token = sys.argv[2]
+    namespace = sys.argv[3]
+
+    if action == "run_pipeline":
+        run_pipeline(token, namespace)
+    elif action == "test_unauthorized_access":
+        test_unauthorized_access(token, namespace)
+    else:
+        sys.exit(1)