code: def testPipeline(self):
      component_a = _make_fake_component_instance(
          "component_a", _OutputTypeA, {}, {}
      )
      component_b = _make_fake_component_instance(
          "component_b",
          _OutputTypeB,
          {"a": component_a.outputs["output"]},
          {},
      )
      component_c = _make_fake_component_instance(
          "component_c",
          _OutputTypeC,
          {"a": component_a.outputs["output"]},
          {},
      )
      component_d = _make_fake_component_instance(
          "component_d",
          _OutputTypeD,
          {
              "b": component_b.outputs["output"],
              "c": component_c.outputs["output"],
          },
          {},
      )
      component_e = _make_fake_component_instance(
          "component_e",
          _OutputTypeE,
          {
              "a": component_a.outputs["output"],
              "b": component_b.outputs["output"],
              "d": component_d.outputs["output"],
          },
          {},
          with_beam=True,
      )
  
      my_pipeline = pipeline.Pipeline(
          pipeline_name="a",
          pipeline_root="b",
          components=[
              component_d,
              component_c,
              component_a,
              component_b,
              component_e,
              component_a,
          ],
          enable_cache=True,
          metadata_connection_config=self._metadata_connection_config,
          beam_pipeline_args=["--runner=PortableRunner"],
          additional_pipeline_args={},
      )
      self.assertCountEqual(
          my_pipeline.components,
          [
              component_a,
              component_b,
              component_c,
              component_d,
              component_e,
          ],
      )
      self.assertCountEqual(
          my_pipeline.components[0].downstream_nodes,
          [component_b, component_c, component_e],
      )
      self.assertEqual(my_pipeline.components[-1], component_e)
      self.assertEqual(my_pipeline.pipeline_info.pipeline_name, "a")
      self.assertEqual(my_pipeline.pipeline_info.pipeline_root, "b")
      self.assertEqual(
          my_pipeline.metadata_connection_config,
          self._metadata_connection_config,
      )
      self.assertTrue(my_pipeline.enable_cache)
      self.assertEqual(
          component_e.executor_spec.beam_pipeline_args,
          ["--runner=PortableRunner"],
      )
      self.assertCountEqual(
          my_pipeline.beam_pipeline_args, ["--runner=PortableRunner"]
      )
      self.assertDictEqual(my_pipeline.additional_pipeline_args, {})
  
imports: import itertools
  import os
  from typing import Any, Dict, Optional, Type
  import tensorflow
  from tfx import types
  from tfx.dsl.components.base import base_beam_component
  from tfx.dsl.components.base import base_component
  from tfx.dsl.components.base import base_executor
  from tfx.dsl.components.base import base_node
  from tfx.dsl.components.base import executor_spec
  from tfx.dsl.placeholder import placeholder
  from tfx.orchestration import metadata
  from tfx.orchestration import pipeline
  from tfx.types.component_spec import ChannelParameter
  from tfx.types.component_spec import ExecutionParameter
  from tfx.utils import test_case_utils
label_desc: Use image node_id 3 for calling the PipelineTest obj's underlying member method code with example usage: obj.testPipeline() without return types