code: def testStubbedTaxiPipelineBeam(self):
      pipeline_ir = compiler.Compiler().compile(self.taxi_pipeline)
  
      logging.info(
          "Replacing with test_data_dir:%s", self._recorded_output_dir
      )
      pipeline_mock.replace_executor_with_stub(
          pipeline_ir, self._recorded_output_dir, []
      )
  
      BeamDagRunner().run_with_ir(pipeline_ir)
  
      self.assertTrue(fileio.exists(self._metadata_path))
  
      metadata_config = metadata.sqlite_metadata_connection_config(
          self._metadata_path
      )
  
      # Verify that recorded files are successfully copied to the output uris.
      with metadata.Metadata(metadata_config) as m:
          artifacts = m.store.get_artifacts()
          artifact_count = len(artifacts)
          executions = m.store.get_executions()
          execution_count = len(executions)
          # Artifact count is greater by 7 due to extra artifacts produced by
          # Evaluator(blessing and evaluation), Trainer(model and model_run) and
          # Transform(example, graph, cache, pre_transform_statistics,
          # pre_transform_schema, post_transform_statistics, post_transform_schema,
          # post_transform_anomalies) minus Resolver which doesn't generate
          # new artifact.
          self.assertEqual(artifact_count, execution_count + 7)
          self.assertLen(self.taxi_pipeline.components, execution_count)
  
          for execution in executions:
              component_id = pipeline_recorder_utils.get_component_id_from_execution(
                  m, execution
              )
              if component_id.startswith("Resolver"):
                  continue
              eid = [execution.id]
              events = m.store.get_events_by_execution_ids(eid)
              output_events = [
                  x
                  for x in events
                  if x.type == metadata_store_pb2.Event.OUTPUT
              ]
              for event in output_events:
                  steps = event.path.steps
                  self.assertTrue(steps[0].HasField("key"))
                  name = steps[0].key
                  artifacts = m.store.get_artifacts_by_id(
                      [event.artifact_id]
                  )
                  for idx, artifact in enumerate(artifacts):
                      self.assertDirectoryEqual(
                          artifact.uri,
                          os.path.join(
                              self._recorded_output_dir,
                              component_id,
                              name,
                              str(idx),
                          ),
                      )
  
      # Calls verifier for pipeline output artifacts, excluding the resolver node.
      BeamDagRunner().run(self.taxi_pipeline)
      pipeline_outputs = executor_verifier_utils.get_pipeline_outputs(
          self.taxi_pipeline.metadata_connection_config,
          self._pipeline_name,
      )
  
      verifier_map = {
          "model": self._verify_model,
          "model_run": self._verify_model,
          "examples": self._verify_examples,
          "schema": self._verify_schema,
          "anomalies": self._verify_anomalies,
          "evaluation": self._verify_evaluation,
          # A subdirectory of updated_analyzer_cache has changing name.
          "updated_analyzer_cache": self._veryify_root_dir,
      }
  
      # List of components to verify. Resolver is ignored because it
      # doesn't have an executor.
      verify_component_ids = [
          component.id
          for component in self.taxi_pipeline.components
          if not component.id.startswith("Resolver")
      ]
  
      for component_id in verify_component_ids:
          logging.info("Verifying %s", component_id)
          for key, artifact_dict in pipeline_outputs[
              component_id
          ].items():
              for idx, artifact in artifact_dict.items():
                  recorded_uri = os.path.join(
                      self._recorded_output_dir,
                      component_id,
                      key,
                      str(idx),
                  )
                  verifier_map.get(key, self._verify_file_path)(
                      artifact.uri, recorded_uri
                  )
  
imports: import os
  from absl import logging
  import tensorflow
  from tfx.dsl.compiler import compiler
  from tfx.dsl.io import fileio
  from tfx.examples.chicago_taxi_pipeline import taxi_pipeline_local
  from tfx.experimental.pipeline_testing import executor_verifier_utils
  from tfx.experimental.pipeline_testing import pipeline_mock
  from tfx.experimental.pipeline_testing import pipeline_recorder_utils
  from tfx.orchestration import metadata
  from tfx.orchestration.beam.beam_dag_runner import BeamDagRunner
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 10 for calling the TaxiPipelineRegressionEndToEndTest obj's underlying member method code with example usage: obj.testStubbedTaxiPipelineBeam() without return types