docstr: End-to-End test for primitive artifacts passing.
docstr_tok: ["End-to-End","test","for","primitive","artifacts","passing","."]
code: def testPrimitiveEnd2EndPipeline(self):
      """End-to-End test for primitive artifacts passing."""
      pipeline_name = "kubeflow-primitive-e2e-test-{}".format(
          test_utils.random_id()
      )
      components = kubeflow_test_utils.create_primitive_type_components(
          pipeline_name
      )
      # Test that the pipeline can be executed successfully.
      pipeline = self._create_pipeline(pipeline_name, components)
      self._compile_and_run_pipeline(
          pipeline=pipeline, workflow_name=pipeline_name + "-run-1"
      )
      # Test if the correct value has been passed.
      str_artifacts = self._get_artifacts_with_type_and_pipeline(
          type_name="String", pipeline_name=pipeline_name
      )
      # There should be exactly one string artifact.
      self.assertEqual(1, len(str_artifacts))
      self.assertEqual(
          self._get_value_of_string_artifact(str_artifacts[0]),
          "hello %s\n" % pipeline_name,
      )
      # Test caching.
      self._compile_and_run_pipeline(
          pipeline=pipeline, workflow_name=pipeline_name + "-run-2"
      )
      cached_execution = (
          self._get_executions_by_pipeline_name_and_state(
              pipeline_name=pipeline_name,
              state=metadata_store_pb2.Execution.State.CACHED,
          )
      )
      self.assertEqual(2, len(cached_execution))
  
imports: import os
  import subprocess
  import time
  from typing import List
  from absl import logging
  from grpc import insecure_channel
  import tensorflow
  from tfx.dsl.io import fileio
  from tfx.orchestration import test_utils
  from tfx.orchestration.experimental.core.testing import test_dynamic_exec_properties_pipeline
  from tfx.orchestration.kubeflow import test_utils
  from tfx.orchestration.test_pipelines import download_grep_print_pipeline
  from tfx.types import standard_artifacts
  from ml_metadata.proto import metadata_store_pb2
  from ml_metadata.proto import metadata_store_service_pb2
  from ml_metadata.proto import metadata_store_service_pb2_grpc
label_desc: Use image node_id 11 for calling the KubeflowEndToEndTest obj's underlying member method code with example usage: obj.testPrimitiveEnd2EndPipeline() without return types