code: def testPenguinPipelineLocal(self, make_warmup):
      LocalDagRunner().run(
          penguin_pipeline_local_infraval._create_pipeline(
              pipeline_name=self._pipeline_name,
              data_root=self._data_root,
              module_file=self._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=self._schema_path,
              beam_pipeline_args=[],
              infra_validator_host_ip_address=self._infra_validator_host_ip_address,
              make_warmup=make_warmup,
          )
      )
  
      self.assertTrue(fileio.exists(self._serving_model_dir))
      self.assertTrue(fileio.exists(self._metadata_path))
      expected_execution_count = 10  # 9 components + 1 resolver
      metadata_config = metadata.sqlite_metadata_connection_config(
          self._metadata_path
      )
      with metadata.Metadata(metadata_config) as m:
          artifact_count = len(m.store.get_artifacts())
          execution_count = len(m.store.get_executions())
          self.assertGreaterEqual(artifact_count, execution_count)
          self.assertEqual(expected_execution_count, execution_count)
  
      self._assertPipelineExecution()
      self._assertInfraValidatorPassed()
  
      # Runs pipeline the second time.
      LocalDagRunner().run(
          penguin_pipeline_local_infraval._create_pipeline(
              pipeline_name=self._pipeline_name,
              data_root=self._data_root,
              module_file=self._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=self._schema_path,
              beam_pipeline_args=[],
              infra_validator_host_ip_address=self._infra_validator_host_ip_address,
              make_warmup=make_warmup,
          )
      )
  
      # All executions but Evaluator and Pusher are cached.
      with metadata.Metadata(metadata_config) as m:
          # Artifact count is increased by 3 caused by Evaluator and Pusher.
          self.assertLen(m.store.get_artifacts(), artifact_count + 3)
          artifact_count = len(m.store.get_artifacts())
          self.assertLen(
              m.store.get_executions(), expected_execution_count * 2
          )
  
      # Runs pipeline the third time.
      LocalDagRunner().run(
          penguin_pipeline_local_infraval._create_pipeline(
              pipeline_name=self._pipeline_name,
              data_root=self._data_root,
              module_file=self._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=self._schema_path,
              beam_pipeline_args=[],
              infra_validator_host_ip_address=self._infra_validator_host_ip_address,
              make_warmup=make_warmup,
          )
      )
  
      # Asserts cache execution.
      with metadata.Metadata(metadata_config) as m:
          # Artifact count is unchanged.
          self.assertLen(m.store.get_artifacts(), artifact_count)
          self.assertLen(
              m.store.get_executions(), expected_execution_count * 3
          )
  
imports: import os
  import subprocess
  from absl.testing import parameterized
  import tensorflow
  from tfx.dsl.io import fileio
  from tfx.examples.penguin import penguin_pipeline_local_infraval
  from tfx.orchestration import metadata
  from tfx.orchestration.local.local_dag_runner import LocalDagRunner
  from tfx.utils import path_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 8 for calling the PenguinPipelineLocalInfravalEndToEndTest obj's underlying member method code with example usage: obj.testPenguinPipelineLocal(make_warmup) without return types