docstr: End-to-End test for a successful pipeline with exit handler.
docstr_tok: ["End-to-End","test","for","a","successful","pipeline","with","exit","handler","."]
code: def testExitHandlerPipelineSuccess(self):
      """End-to-End test for a successful pipeline with exit handler."""
      pipeline_name = "kubeflow-v2-exit-handler-test-{}".format(
          orchestration_test_utils.random_id()
      )
  
      components = test_utils.simple_pipeline_components(
          _TEST_DATA_ROOT
      )
  
      beam_pipeline_args = [
          "--temp_location="
          + os.path.join(
              self._pipeline_root(pipeline_name), "dataflow", "temp"
          ),
          "--project={}".format(self._GCP_PROJECT_ID),
      ]
  
      pipeline = self._create_pipeline(
          pipeline_name, components, beam_pipeline_args
      )
  
      output_file_dir = os.path.join(
          self._pipeline_root(pipeline_name), _success_file_name
      )
  
      exit_handler = custom_exit_handler.test_exit_handler(
          final_status=tfx.orchestration.experimental.FinalStatusStr(),
          file_dir=output_file_dir,
      )
  
      self._run_pipeline(pipeline=pipeline, exit_handler=exit_handler)
  
      # verify execution results
      actual_final_status_str = io_utils.read_string_file(
          output_file_dir
      )
      expected_successful_final_status_str = """
        {
          "state":"SUCCEEDED",
          "error":{},
          "pipeline_task_name":"tfx-dag"
        }
      """
  
      expected_successful_final_status = (
          pipeline_spec_pb2.PipelineTaskFinalStatus()
      )
      json_format.Parse(
          expected_successful_final_status_str,
          expected_successful_final_status,
      )
  
      actual_final_status = pipeline_spec_pb2.PipelineTaskFinalStatus()
      json_format.Parse(actual_final_status_str, actual_final_status)
  
      self.assertProtoPartiallyEquals(
          expected_successful_final_status,
          actual_final_status,
          ignored_fields=["pipeline_job_resource_name"],
      )
  
file_path: tfx/tfx/orchestration/kubeflow/v2/e2e_tests/exit_handler_e2e_test.py
imports: import os
  from kfp.pipeline_spec import pipeline_spec_pb2
  import tensorflow
  from tfx import v1
  from tfx.orchestration import test_utils
  from tfx.orchestration.kubeflow.v2 import test_utils
  from tfx.orchestration.kubeflow.v2.e2e_tests import base_test_case
  from tfx.orchestration.test_pipelines import custom_exit_handler
  from tfx.utils import io_utils
  from google.protobuf import json_format
label_desc: Use image node_id 1 for calling the ExitHandlerE2ETest obj's underlying member method code with example usage: obj.testExitHandlerPipelineSuccess() without return types