code: def testPenguinPipelineLocal(self, model_framework):
      if model_framework == "tfdf_experimental":
          # Skip if TFDF is not available or incompatible.
          try:
              importlib.import_module("tensorflow_decision_forests")
          except (ImportError, tf.errors.NotFoundError):
              self.skipTest(
                  "TensorflowDecisionForests is not available"
              )
      elif model_framework == "flax_experimental":
          # Skip if flax is not available or incompatible.
          try:
              importlib.import_module("flax")
          except (ImportError, tf.errors.NotFoundError):
              self.skipTest("flax is not available")
      module_file = self._module_file_name(model_framework)
      pipeline = penguin_pipeline_local.create_pipeline(
          pipeline_name=self._pipeline_name,
          data_root=self._data_root,
          module_file=module_file,
          accuracy_threshold=0.1,
          serving_model_dir=self._serving_model_dir,
          pipeline_root=self._pipeline_root,
          metadata_path=self._metadata_path,
          user_provided_schema_path=None,
          enable_tuning=False,
          enable_bulk_inferrer=False,
          examplegen_input_config=None,
          examplegen_range_config_date=None,
          resolver_range_config=None,
          beam_pipeline_args=self._make_beam_pipeline_args(),
          enable_transform_input_cache=False,
          enable_example_diff=False,
      )
  
      logging.info("Starting the first pipeline run.")
      LocalDagRunner().run(pipeline)
  
      self.assertTrue(fileio.exists(self._serving_model_dir))
      self.assertTrue(fileio.exists(self._metadata_path))
      expected_execution_count = 8  # 7 components + 1 resolver
      metadata_config = metadata.sqlite_metadata_connection_config(
          self._metadata_path
      )
      store = mlmd.MetadataStore(metadata_config)
      artifact_count = len(store.get_artifacts())
      execution_count = len(store.get_executions())
      self.assertGreaterEqual(artifact_count, execution_count)
      self.assertEqual(expected_execution_count, execution_count)
  
      self.assertPipelineExecution()
  
      logging.info(
          "Starting the second pipeline run. All components except "
          "Evaluator and Pusher will use cached results."
      )
      LocalDagRunner().run(pipeline)
  
      # Artifact count is increased by 3 caused by Evaluator and Pusher.
      self.assertLen(store.get_artifacts(), artifact_count + 3)
      artifact_count = len(store.get_artifacts())
      self.assertLen(
          store.get_executions(), expected_execution_count * 2
      )
  
      logging.info(
          "Starting the third pipeline run. "
          "All components will use cached results."
      )
      LocalDagRunner().run(pipeline)
  
      # Asserts cache execution.
      # Artifact count is unchanged.
      self.assertLen(store.get_artifacts(), artifact_count)
      self.assertLen(
          store.get_executions(), expected_execution_count * 3
      )
  
imports: import importlib
  import os
  from typing import List
  from absl import logging
  from absl.testing import parameterized
  import tensorflow
  from tfx.examples.penguin import penguin_pipeline_local
  from tfx.v1 import proto
  from tfx.v1.dsl.io import fileio
  from tfx.v1.orchestration import LocalDagRunner
  from tfx.v1.orchestration import metadata
  import ml_metadata
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 6 for calling the PenguinPipelineLocalEndToEndTest obj's underlying member method code with example usage: obj.testPenguinPipelineLocal(model_framework) without return types