update python formatting.

Signed-off-by: agoins <[email protected]>

Author: alyssacgoins

backend/src/v2/component/launcher_v2.go

@@ -910,7 +910,7 @@ func getExecutorOutputFile(path string) (*pipelinespec.ExecutorOutput, error) {
 
 func LocalPathForURI(uri string) (string, error) {
 	if strings.HasPrefix(uri, "gs://") {
-		return "/gcs/" + strings.TrimPrefix(uri, "gs://"), nil
+		return "/home/agoins/gcs/" + strings.TrimPrefix(uri, "gs://"), nil
 	}
 	if strings.HasPrefix(uri, "minio://") {
 		return "/minio/" + strings.TrimPrefix(uri, "minio://"), nil

backend/src/v2/component/launcher_v2_test.go

@@ -51,16 +51,17 @@ var addNumbersComponent = &pipelinespec.ComponentSpec{
 }
 
 // Tests that launcher correctly executes the user component and successfully writes output parameters to file.
-func Test_executeV2_Parameters(t *testing.T) {
-	customLocalPath := "/tmp/custom/artifact.txt"
+func Test_executeV2_Artifacts(t *testing.T) {
+	uri := "gs:///tmp/custom/artifact.txt"
+	custom := "mem://test-bucket/pipeline-root/temp.txt"
 	tests := []struct {
 		name          string
 		executorInput *pipelinespec.ExecutorInput
 		executorArgs  []string
 		wantErr       bool
 	}{
 		{
-			"happy pass",
+			"custom path",
 			&pipelinespec.ExecutorInput{
 				Inputs: &pipelinespec.ExecutorInput_Inputs{
 					ParameterValues: map[string]*structpb.Value{"a": structpb.NewNumberValue(1), "b": structpb.NewNumberValue(2)},
@@ -70,7 +71,11 @@ func Test_executeV2_Parameters(t *testing.T) {
 						"dataset": {
 							Artifacts: []*pipelinespec.RuntimeArtifact{
 								{
-									Uri: customLocalPath,
+									Uri:        uri,
+									CustomPath: &custom,
+									Type: &pipelinespec.ArtifactTypeSchema{
+										Kind: &pipelinespec.ArtifactTypeSchema_InstanceSchema{InstanceSchema: "title: kfp.Model\ntype: object\nproperties:\n  framework:\n    type: string\n  framework_version:\n    type: string\n"},
+									},
 								},
 							},
 						},
@@ -80,6 +85,83 @@ func Test_executeV2_Parameters(t *testing.T) {
 			[]string{"-c", "test {{$.inputs.parameters['a']}} -eq 1 || exit 1\ntest {{$.inputs.parameters['b']}} -eq 2 || exit 1"},
 			false,
 		},
+		{
+			"default path",
+			&pipelinespec.ExecutorInput{
+				Inputs: &pipelinespec.ExecutorInput_Inputs{
+					ParameterValues: map[string]*structpb.Value{"b": structpb.NewNumberValue(2)},
+				},
+				Outputs: &pipelinespec.ExecutorInput_Outputs{
+					Artifacts: map[string]*pipelinespec.ArtifactList{
+						"dataset": {
+							Artifacts: []*pipelinespec.RuntimeArtifact{
+								{
+									Uri:        uri,
+									CustomPath: &custom,
+									Type: &pipelinespec.ArtifactTypeSchema{
+										Kind: &pipelinespec.ArtifactTypeSchema_InstanceSchema{InstanceSchema: "title: kfp.Model\ntype: object\nproperties:\n  framework:\n    type: string\n  framework_version:\n    type: string\n"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			[]string{"-c", "test {{$.inputs.parameters['a']}} -eq 5 || exit 1\ntest {{$.inputs.parameters['b']}} -eq 2 || exit 1"},
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeKubernetesClientset := &fake.Clientset{}
+			fakeMetadataClient := metadata.NewFakeClient()
+			bucket, err := blob.OpenBucket(context.Background(), "mem://test-bucket")
+			assert.Nil(t, err)
+			bucketConfig, err := objectstore.ParseBucketConfig("mem://test-bucket/pipeline-root/", nil)
+			assert.Nil(t, err)
+			_, _, err = executeV2(
+				context.Background(),
+				test.executorInput,
+				addNumbersComponent,
+				"sh",
+				test.executorArgs,
+				bucket,
+				bucketConfig,
+				fakeMetadataClient,
+				"namespace",
+				fakeKubernetesClientset,
+				"false",
+			)
+			//
+			//if test.wantErr {
+			//	assert.NotNil(t, err)
+			//} else {
+			//	assert.Nil(t, err)
+			//}
+			assert.NotEmpty(t, bucket)
+		})
+	}
+}
+
+// Tests that launcher correctly executes the user component and successfully writes output parameters to file.
+func Test_executeV2_Parameters(t *testing.T) {
+	tests := []struct {
+		name          string
+		executorInput *pipelinespec.ExecutorInput
+		executorArgs  []string
+		wantErr       bool
+	}{
+		{
+			"happy pass",
+			&pipelinespec.ExecutorInput{
+				Inputs: &pipelinespec.ExecutorInput_Inputs{
+					ParameterValues: map[string]*structpb.Value{"a": structpb.NewNumberValue(1), "b": structpb.NewNumberValue(2)},
+				},
+			},
+			[]string{"-c", "test {{$.inputs.parameters['a']}} -eq 1 || exit 1\ntest {{$.inputs.parameters['b']}} -eq 2 || exit 1"},
+			false,
+		},
 		{
 			"use default value",
 			&pipelinespec.ExecutorInput{

samples/v2/component_with_runtime_artifct_custom_path.py renamed to samples/v2/component_with_runtime_artifact_custom_path.py

@@ -21,4 +21,4 @@ def validate_custom_path(exp_path: str, input_list: Output[Artifact]) -> bool:
 def pipeline_with_custom_path_artifact():
     task1 = create_list()
     task1.output.set_custom_path('/etc/test/file/path')
-    # task2 = validate_custom_path(path='/etc/test/file/path', input_list=task1.output)
+    # task2 = validate_custom_path(path='/etc/test/file/path', input_list=task1.output)

samples/v2/sample_test.py

@@ -41,12 +41,13 @@ def get_kfp_pipeline_spec_path() -> str:
     return path
 
 def get_package_path(subdir: str) -> str:
-    repo_name = os.environ.get('REPO_NAME', 'kubeflow/pipelines')
-    if os.environ.get('PULL_NUMBER'):
-        path = f'git+https://github.com/{repo_name}.git@refs/pull/{os.environ["PULL_NUMBER"]}/merge#subdirectory={subdir}'
-    else:
-        path = f'git+https://github.com/{repo_name}.git@master#subdirectory={subdir}'
-    return path
+    return f'git+https://github.com/alyssacgoins/data-science-pipelines.git@refs/pull/12248/merge#subdirectory=sdk/python'
+    # repo_name = os.environ.get('REPO_NAME', 'kubeflow/pipelines')
+    # if os.environ.get('PULL_NUMBER'):
+    #     path = f'git+https://github.com/{repo_name}.git@refs/pull/{os.environ["PULL_NUMBER"]}/merge#subdirectory={subdir}'
+    # else:
+    #     path = f'git+https://github.com/{repo_name}.git@master#subdirectory={subdir}'
+    # return path
 
 # Set the component configuration BEFORE importing any pipeline modules
 # To have pipeline execution code leverage source kfp-pipeline-spec

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

@@ -82,7 +82,7 @@ def __init__(self,
         self.uri = uri or ''
         self.name = name or ''
         self.metadata = metadata or {}
-        self.custom_path: str = ''
+        self.custom_path = ''
 
     @property
     def path(self) -> str:
@@ -93,7 +93,7 @@ def path(self, path: str) -> None:
         self._set_path(path)
 
     def _get_path(self) -> Optional[str]:
-        if self.custom_path != "":
+        if self.custom_path is not None:
             return self.custom_path
         if self.uri.startswith(RemotePrefix.GCS.value):
             return _GCS_LOCAL_MOUNT_PREFIX + self.uri[len(RemotePrefix.GCS.value
@@ -116,14 +116,12 @@ def _set_path(self, path: str) -> None:
 
     @property
     def custom_path(self) -> str:
-        return self.custom_path
+        return self._custom_path
 
     @custom_path.setter
     def custom_path(self, value):
-        self._set_custom_path = value
+        self._custom_path = value
 
-    def _set_custom_path(self, path: str) -> None:
-        self.custom_path = path
 
 def convert_local_path_to_remote_path(path: str) -> str:
     if path.startswith(_GCS_LOCAL_MOUNT_PREFIX):

sdk/python/test_data/pipelines/component_with_runtime_artifact_custom_path.py

@@ -1,25 +1,30 @@
+from kfp import dsl
 from kfp.dsl import Output
 from kfp.dsl.types.artifact_types import Artifact
-from kfp.v2 import dsl
 
 
 @dsl.component
 def create_list() -> list:
     return [1, 2, 3, 4]
 
+
 @dsl.component
 def append_to_list(digit: int, input_list: Output[Artifact]) -> list:
     input_list.append(digit)
     return input_list
 
+
 @dsl.component
 def validate_custom_path(exp_path: str, input_list: Output[Artifact]) -> bool:
     #todo: is this the correct comparison? (or should use != instead?)
     if input_list.path is not exp_path:
-        raise ValueError(f"File uri is {input_list.path} but should be {exp_path}.")
+        raise ValueError(
+            f"File uri is {input_list.path} but should be {exp_path}.")
+
 
 @dsl.pipeline
 def pipeline_with_custom_path_artifact():
     task1 = create_list()
     task1.output.set_custom_path('/etc/test/file/path')
-    task2 = validate_custom_path(path='/etc/test/file/path', input_list=task1.output)
+    task2 = validate_custom_path(
+        path='/etc/test/file/path', input_list=task1.output)