code: def test_multi_step_framework_processing_pipeline_same_source_dir(
      pipeline_session, role, pipeline_name
  ):
      default_bucket = pipeline_session.default_bucket()
      cache_config = CacheConfig(
          enable_caching=True, expire_after="PT1H"
      )
  
      SOURCE_DIR = "/pipeline/test_source_dir"
  
      role_param = ParameterString(name="Role", default_value=role)
  
      framework_processor_tf = FrameworkProcessor(
          role=role_param,
          instance_type="ml.m5.xlarge",
          instance_count=1,
          estimator_cls=TensorFlow,
          framework_version="2.9",
          py_version="py39",
          sagemaker_session=pipeline_session,
      )
  
      framework_processor_sk = FrameworkProcessor(
          framework_version="1.0-1",
          instance_type="ml.m5.xlarge",
          instance_count=1,
          base_job_name="my-job",
          role=role_param,
          estimator_cls=SKLearn,
          sagemaker_session=pipeline_session,
      )
  
      step_1 = ProcessingStep(
          name="Step-1",
          step_args=framework_processor_tf.run(
              code="script_1.py",
              source_dir=DATA_DIR + SOURCE_DIR,
              outputs=[
                  ProcessingOutput(
                      output_name="test",
                      source="/opt/ml/processing/test",
                  )
              ],
          ),
          cache_config=cache_config,
      )
  
      step_2 = ProcessingStep(
          name="Step-2",
          step_args=framework_processor_sk.run(
              code="script_2.py",
              source_dir=DATA_DIR + SOURCE_DIR,
              inputs=[
                  ProcessingInput(
                      source=step_1.properties.ProcessingOutputConfig.Outputs[
                          "test"
                      ].S3Output.S3Uri,
                      destination="/opt/ml/processing/test",
                  ),
              ],
          ),
          cache_config=cache_config,
      )
  
      pipeline = Pipeline(
          name=pipeline_name,
          steps=[step_1, step_2],
          sagemaker_session=pipeline_session,
          parameters=[role_param],
      )
      try:
          pipeline.create(role)
          definition = json.loads(pipeline.definition())
  
          (
              source_dir_1_tar_uri,
              entry_point_1,
          ) = _verify_code_artifacts_of_framework_processing_step(
              pipeline_session,
              framework_processor_tf,
              default_bucket,
              pipeline_name,
              definition["Steps"][0],
              DATA_DIR + SOURCE_DIR,
              "script_1.py",
          )
  
          (
              source_dir_2_tar_uri,
              entry_point_2,
          ) = _verify_code_artifacts_of_framework_processing_step(
              pipeline_session,
              framework_processor_sk,
              default_bucket,
              pipeline_name,
              definition["Steps"][1],
              DATA_DIR + SOURCE_DIR,
              "script_2.py",
          )
  
          # the tarred source dirs should have a different s3 uri since the entry_point code is different
          assert source_dir_1_tar_uri != source_dir_2_tar_uri
  
          # verify different entry_point paths
          assert entry_point_1 != entry_point_2
  
          execution = pipeline.start(parameters={})
          wait_pipeline_execution(
              execution=execution, delay=540, max_attempts=3
          )
  
          execution_steps = execution.list_steps()
          assert len(execution_steps) == 2
          for step in execution_steps:
              assert step["StepStatus"] == "Succeeded"
  
      finally:
          try:
              pipeline.delete()
          except Exception:
              pass
  
imports: from __future__ import absolute_import
  import json
  import os
  import re
  import subprocess
  from datetime import datetime
  from pathlib import Path
  import pytest
  from tests.integ.sagemaker.workflow.helpers import wait_pipeline_execution
  from sagemaker.workflow.utilities import hash_files_or_dirs, hash_object
  from sagemaker import image_uris, get_execution_role, utils
  from sagemaker.dataset_definition import DatasetDefinition, AthenaDatasetDefinition
  from sagemaker.processing import ProcessingInput, ProcessingOutput, FrameworkProcessor
  from sagemaker.s3 import S3Uploader, S3Downloader
  from sagemaker.sklearn import SKLearnProcessor, SKLearn
  from sagemaker.tensorflow import TensorFlow
  from sagemaker.workflow.parameters import ParameterInteger, ParameterString
  from sagemaker.workflow.pipeline import Pipeline
  from sagemaker.workflow.steps import ProcessingStep, CacheConfig
  from sagemaker.spark.processing import PySparkProcessor, SparkJarProcessor
  from sagemaker.wrangler.processing import DataWranglerProcessor
  from tests.integ import DATA_DIR
label_desc: Use image node_id 9 for calling a global function with example usage: test_multi_step_framework_processing_pipeline_same_source_dir(pipeline_session, role, pipeline_name) without return types