code: def testRunWithIntermediateDeploymentConfig(self):
      self._pipeline.deployment_config.Pack(
          _INTERMEDIATE_DEPLOYMENT_CONFIG
      )
      beam_dag_runner.BeamDagRunner().run_with_ir(self._pipeline)
      self.assertEqual(
          _component_executors,
          {
              "my_example_gen": text_format.Parse(
                  'class_path: "tfx.components.example_gen_executor"',
                  _PythonClassExecutableSpec(),
              ),
              "my_transform": text_format.Parse(
                  'class_path: "tfx.components.transform_executor"',
                  _PythonClassExecutableSpec(),
              ),
              "my_trainer": text_format.Parse(
                  'image: "path/to/docker/image"',
                  _ContainerExecutableSpec(),
              ),
              "my_importer": None,
          },
      )
      self.assertEqual(
          _component_drivers,
          {
              "my_example_gen": text_format.Parse(
                  'class_path: "tfx.components.example_gen_driver"',
                  _PythonClassExecutableSpec(),
              ),
              "my_transform": None,
              "my_trainer": None,
              "my_importer": None,
          },
      )
      self.assertEqual(
          _component_platform_configs,
          {
              "my_example_gen": None,
              "my_transform": None,
              "my_trainer": text_format.Parse(
                  'docker_server_url: "docker/server/url"',
                  _DockerPlatformConfig(),
              ),
              "my_importer": None,
          },
      )
      # 'my_importer' has no upstream and can be executed in any order.
      self.assertIn("my_importer", _executed_components)
      _executed_components.remove("my_importer")
      self.assertEqual(
          _executed_components,
          ["my_example_gen", "my_transform", "my_trainer"],
      )
      # Verifies that every component gets a not-None pipeline_run.
      self.assertTrue(all(_component_to_pipeline_run.values()))
  
imports: import os
  from typing import Optional
  from unittest import mock
  import tensorflow
  from tfx.dsl.compiler import constants
  from tfx.orchestration import metadata
  from tfx.orchestration.beam import beam_dag_runner
  from tfx.orchestration.beam.legacy import beam_dag_runner
  from tfx.orchestration.config import pipeline_config
  from tfx.proto.orchestration import executable_spec_pb2
  from tfx.proto.orchestration import local_deployment_config_pb2
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.proto.orchestration import platform_config_pb2
  from tfx.utils import test_case_utils
  from google.protobuf import message
  from google.protobuf import text_format
label_desc: Use image node_id 3 for calling the BeamDagRunnerTest obj's underlying member method code with example usage: obj.testRunWithIntermediateDeploymentConfig() without return types