code: def testImdbPipelineNativeKeras(self):
      pipeline = imdb_pipeline_native_keras._create_pipeline(
          pipeline_name=self._pipeline_name,
          data_root=self._data_root,
          module_file=self._module_file,
          serving_model_dir=self._serving_model_dir,
          pipeline_root=self._pipeline_root,
          metadata_path=self._metadata_path,
          beam_pipeline_args=[],
      )
  
      BeamDagRunner().run(pipeline)
  
      self.assertTrue(fileio.exists(self._serving_model_dir))
      self.assertTrue(fileio.exists(self._metadata_path))
      expected_execution_count = 9  # 8 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()
  
      # Runs pipeline the second time.
      BeamDagRunner().run(pipeline)
  
      # 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.assertEqual(
              artifact_count + 3, len(m.store.get_artifacts())
          )
          artifact_count = len(m.store.get_artifacts())
          self.assertEqual(
              expected_execution_count * 2,
              len(m.store.get_executions()),
          )
  
      # Runs pipeline the third time.
      BeamDagRunner().run(pipeline)
  
      # Asserts cache execution.
      with metadata.Metadata(metadata_config) as m:
          # Artifact count is unchanged.
          self.assertEqual(artifact_count, len(m.store.get_artifacts()))
          self.assertEqual(
              expected_execution_count * 3,
              len(m.store.get_executions()),
          )
  
file_path: tfx/tfx/examples/imdb/imdb_pipeline_native_keras_e2e_test.py
imports: import os
  import tensorflow
  from tfx.dsl.io import fileio
  from tfx.examples.imdb import imdb_pipeline_native_keras
  from tfx.orchestration import metadata
  from tfx.orchestration.beam.beam_dag_runner import BeamDagRunner
label_desc: Use image node_id 4 for calling the ImdbPipelineNativeKerasEndToEndTest obj's underlying member method code with example usage: obj.testImdbPipelineNativeKeras() without return types