code: def testPenguinPipelineLocalWithRollingWindow(self):
      module_file = self._module_file_name("keras")
      examplegen_input_config = proto.Input(
          splits=[
              proto.Input.Split(name="test", pattern="day{SPAN}/*"),
          ]
      )
      resolver_range_config = proto.RangeConfig(
          rolling_range=proto.RollingRange(num_spans=2)
      )
  
      def run_pipeline(examplegen_range_config_date):
          LocalDagRunner().run(
              penguin_pipeline_local.create_pipeline(
                  pipeline_name=self._pipeline_name,
                  data_root=self._data_root_span,
                  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=examplegen_input_config,
                  examplegen_range_config_date=examplegen_range_config_date,
                  resolver_range_config=resolver_range_config,
                  beam_pipeline_args=self._make_beam_pipeline_args(),
                  enable_transform_input_cache=True,
                  enable_example_diff=False,
              )
          )
  
      # Trigger the pipeline for the first span.
      run_pipeline(examplegen_range_config_date="20220102")
  
      self.assertTrue(fileio.exists(self._serving_model_dir))
      self.assertTrue(fileio.exists(self._metadata_path))
      self.assertPipelineExecution()
      transform_execution_type = (
          "tfx.components.transform.component.Transform"
      )
      trainer_execution_type = (
          "tfx.components.trainer.component.Trainer"
      )
      expected_execution_count = 11  # 8 components + 3 resolvers
      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)
      # Verify Transform's input examples artifacts.
      tft_input_examples_artifacts = self._get_input_examples_artifacts(
          store, transform_execution_type
      )
      self.assertLen(tft_input_examples_artifacts, 1)
      # SpansResolver (controlled by resolver_range_config) returns span 1.
      self.assertEqual(
          1,
          tft_input_examples_artifacts[0]
          .custom_properties[_SPAN_PROPERTY_NAME]
          .int_value,
      )
  
      # Trigger the pipeline for the second span.
      run_pipeline(examplegen_range_config_date="20220103")
  
      execution_count = len(store.get_executions())
      # In the second run, evaluator may not bless the model. As a result,
      # conditional-guarded Pusher may not be executed.
      expected_minimum_execution_count = (
          expected_execution_count * 2 - 1
      )
      self.assertLessEqual(
          expected_minimum_execution_count, execution_count
      )
      # Verify Transform's input examples artifacts.
      tft_input_examples_artifacts = self._get_input_examples_artifacts(
          store, transform_execution_type
      )
      self.assertLen(tft_input_examples_artifacts, 2)
      spans = {
          tft_input_examples_artifacts[0]
          .custom_properties[_SPAN_PROPERTY_NAME]
          .int_value,
          tft_input_examples_artifacts[1]
          .custom_properties[_SPAN_PROPERTY_NAME]
          .int_value,
      }
      # SpansResolver (controlled by resolver_range_config) returns span 1 & 2.
      self.assertSetEqual({1, 2}, spans)
      # Verify Trainer's input examples artifacts.
      self.assertLen(
          self._get_input_examples_artifacts(
              store, trainer_execution_type
          ),
          2,
      )
  
      # Trigger the pipeline for the thrid span.
      run_pipeline(examplegen_range_config_date="20220104")
  
      metadata_config = metadata.sqlite_metadata_connection_config(
          self._metadata_path
      )
      execution_count = len(store.get_executions())
      # Similar to the second run, conditional-guarded Pusher may not be executed.
      expected_minimum_execution_count = (
          expected_execution_count * 3 - 2
      )
      self.assertLessEqual(
          expected_minimum_execution_count, execution_count
      )
      # Verify Transform's input examples artifacts.
      tft_input_examples_artifacts = self._get_input_examples_artifacts(
          store, transform_execution_type
      )
      self.assertLen(tft_input_examples_artifacts, 2)
      spans = {
          tft_input_examples_artifacts[0]
          .custom_properties[_SPAN_PROPERTY_NAME]
          .int_value,
          tft_input_examples_artifacts[1]
          .custom_properties[_SPAN_PROPERTY_NAME]
          .int_value,
      }
      # SpansResolver (controlled by resolver_range_config) returns span 2 & 3.
      self.assertSetEqual({2, 3}, spans)
      # Verify Trainer's input examples artifacts.
      self.assertLen(
          self._get_input_examples_artifacts(
              store, trainer_execution_type
          ),
          2,
      )
  
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 11 for calling the PenguinPipelineLocalEndToEndTest obj's underlying member method code with example usage: obj.testPenguinPipelineLocalWithRollingWindow() without return types