code: def test_caching_behavior(
      pipeline_session,
      role,
      cpu_instance_type,
      pipeline_name,
      script_dir,
      athena_dataset_definition,
      region_name,
  ):
      default_bucket = pipeline_session.default_bucket()
      data_path = os.path.join(DATA_DIR, "workflow")
  
      framework_version = "0.20.0"
      instance_type = "ml.m5.xlarge"
      instance_count = ParameterInteger(
          name="InstanceCount", default_value=1
      )
      output_prefix = ParameterString(
          name="OutputPrefix", default_value="output"
      )
  
      input_data = f"s3://sagemaker-sample-data-{region_name}/processing/census/census-income.csv"
  
      # additionally add abalone input, so we can test input s3 file from local upload
      abalone_input = ProcessingInput(
          input_name="abalone_data",
          source=os.path.join(data_path, "abalone-dataset.csv"),
          destination="/opt/ml/processing/input",
      )
  
      # define processing step
      sklearn_processor = SKLearnProcessor(
          framework_version=framework_version,
          instance_type=instance_type,
          instance_count=instance_count,
          base_job_name="test-sklearn",
          sagemaker_session=pipeline_session,
          role=role,
      )
      processor_args = sklearn_processor.run(
          inputs=[
              ProcessingInput(
                  source=input_data,
                  destination="/opt/ml/processing/input",
              ),
              ProcessingInput(
                  dataset_definition=athena_dataset_definition
              ),
              abalone_input,
          ],
          outputs=[
              ProcessingOutput(
                  output_name="train_data",
                  source="/opt/ml/processing/train",
              ),
              ProcessingOutput(
                  output_name="test_data",
                  source="/opt/ml/processing/test",
                  destination=Join(
                      on="/",
                      values=[
                          "s3:/",
                          pipeline_session.default_bucket(),
                          "test-sklearn",
                          output_prefix,
                          ExecutionVariables.PIPELINE_EXECUTION_ID,
                      ],
                  ),
              ),
          ],
          code=os.path.join(script_dir, "preprocessing.py"),
      )
      step_process = ProcessingStep(
          name="my-process",
          display_name="ProcessingStep",
          description="description for Processing step",
          step_args=processor_args,
      )
  
      # define training step
      sklearn_train = SKLearn(
          framework_version=framework_version,
          source_dir=script_dir,
          entry_point=os.path.join(script_dir, "train.py"),
          instance_type=instance_type,
          sagemaker_session=pipeline_session,
          role=role,
      )
      train_args = sklearn_train.fit(
          inputs=TrainingInput(
              s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                  "train_data"
              ].S3Output.S3Uri
          ),
      )
      step_train = TrainingStep(
          name="my-train",
          display_name="TrainingStep",
          description="description for Training step",
          step_args=train_args,
      )
  
      # define pipeline
      pipeline = Pipeline(
          name=pipeline_name,
          parameters=[instance_count, output_prefix],
          steps=[step_process, step_train],
          sagemaker_session=pipeline_session,
      )
  
      try:
          # create pipeline
          pipeline.create(role)
          definition = json.loads(pipeline.definition())
  
          # verify input path
          expected_abalone_input_path = (
              f"{pipeline_name}/{step_process.name}"
              f"/input/abalone_data"
          )
          expected_abalone_input_file = (
              f"{expected_abalone_input_path}/abalone-dataset.csv"
          )
  
          s3_input_objects = pipeline_session.list_s3_files(
              bucket=default_bucket,
              key_prefix=expected_abalone_input_path,
          )
          assert expected_abalone_input_file in s3_input_objects
  
          # verify code path
          expected_code_path = (
              f"{pipeline_name}/code/"
              f"{hash_files_or_dirs([script_dir])}"
          )
          expected_training_file = (
              f"{expected_code_path}/sourcedir.tar.gz"
          )
  
          s3_code_objects = pipeline_session.list_s3_files(
              bucket=default_bucket, key_prefix=expected_code_path
          )
          assert expected_training_file in s3_code_objects
  
          # update pipeline
          pipeline.update(role)
  
          # verify no changes
          definition2 = json.loads(pipeline.definition())
          assert definition == definition2
  
          # add dummy file to source_dir
          shutil.copyfile(
              DATA_DIR + "/dummy_script.py",
              script_dir + "/dummy_script.py",
          )
  
          # update pipeline again
          pipeline.update(role)
  
          # verify changes
          definition3 = json.loads(pipeline.definition())
          assert definition != definition3
  
      finally:
          try:
              os.remove(script_dir + "/dummy_script.py")
              pipeline.delete()
          except Exception:
              os.remove(script_dir + "/dummy_script.py")
              pass
  
imports: from __future__ import absolute_import
  import json
  import os
  import re
  import tempfile
  import time
  import shutil
  from contextlib import contextmanager
  import pytest
  import pandas
  from sagemaker.utils import retry_with_backoff
  from tests.integ.sagemaker.workflow.helpers import wait_pipeline_execution
  from tests.integ.s3_utils import extract_files_from_s3
  from sagemaker.workflow.model_step import ModelStep, _REGISTER_MODEL_NAME_BASE, _REPACK_MODEL_NAME_BASE
  from sagemaker.parameter import IntegerParameter
  from sagemaker.pytorch import PyTorch, PyTorchModel
  from sagemaker.tuner import HyperparameterTuner
  from tests.integ.timeout import timeout
  from sagemaker.session import Session
  from sagemaker import image_uris
  from sagemaker.estimator import Estimator
  from sagemaker.inputs import CreateModelInput, TrainingInput
  from sagemaker.model import Model
  from sagemaker.model_metrics import MetricsSource, ModelMetrics
  from sagemaker.processing import ProcessingInput, ProcessingOutput, FeatureStoreOutput, ScriptProcessor
  from sagemaker.s3 import S3Uploader
  from sagemaker.session import get_execution_role
  from sagemaker.sklearn.estimator import SKLearn
  from sagemaker.transformer import Transformer
  from sagemaker.sklearn.processing import SKLearnProcessor
  from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo, ConditionLessThanOrEqualTo
  from sagemaker.workflow.condition_step import ConditionStep
  from sagemaker.workflow.callback_step import CallbackStep, CallbackOutput, CallbackOutputTypeEnum
  from sagemaker.wrangler.processing import DataWranglerProcessor
  from sagemaker.dataset_definition.inputs import DatasetDefinition, AthenaDatasetDefinition
  from sagemaker.workflow.execution_variables import ExecutionVariables
  from sagemaker.workflow.functions import Join, JsonGet
  from sagemaker.wrangler.ingestion import generate_data_ingestion_flow_from_s3_input
  from sagemaker.workflow.parameters import ParameterInteger, ParameterString
  from sagemaker.workflow.steps import CreateModelStep, ProcessingStep, TrainingStep, TransformStep, TransformInput, PropertyFile, TuningStep
  from sagemaker.workflow.step_collections import RegisterModel
  from sagemaker.workflow.pipeline import Pipeline
  from sagemaker.workflow.utilities import hash_files_or_dirs
  from sagemaker.feature_store.feature_group import FeatureGroup, FeatureDefinition, FeatureTypeEnum
  from tests.integ import DATA_DIR
label_desc: Use image node_id 21 for calling a global function with example usage: test_caching_behavior(pipeline_session, role, cpu_instance_type, pipeline_name, script_dir, athena_dataset_definition, region_name) without return types