code: def test_one_step_data_wrangler_processing_pipeline(
      sagemaker_session, role, pipeline_name
  ):
      instance_count = ParameterInteger(
          name="InstanceCount", default_value=1
      )
      instance_type = ParameterString(
          name="InstanceType", default_value="ml.m5.4xlarge"
      )
  
      recipe_file_path = os.path.join(
          DATA_DIR, "workflow", "dummy_recipe.flow"
      )
      input_file_path = os.path.join(
          DATA_DIR, "workflow", "dummy_data.csv"
      )
  
      output_name = "3f74973c-fd1e-4845-89f8-0dd400031be9.default"
      output_content_type = "CSV"
      output_config = {
          output_name: {"content_type": output_content_type}
      }
      job_argument = [f"--output-config '{json.dumps(output_config)}'"]
  
      inputs = [
          ProcessingInput(
              input_name="dummy_data.csv",
              source=input_file_path,
              destination="/opt/ml/processing/dummy_data.csv",
          )
      ]
  
      output_s3_uri = (
          f"s3://{sagemaker_session.default_bucket()}/output"
      )
      outputs = [
          ProcessingOutput(
              output_name=output_name,
              source="/opt/ml/processing/output",
              destination=output_s3_uri,
              s3_upload_mode="EndOfJob",
          )
      ]
  
      data_wrangler_processor = DataWranglerProcessor(
          role=role,
          data_wrangler_flow_source=recipe_file_path,
          instance_count=instance_count,
          instance_type=instance_type,
          sagemaker_session=sagemaker_session,
          max_runtime_in_seconds=86400,
      )
  
      data_wrangler_step = ProcessingStep(
          name="data-wrangler-step",
          processor=data_wrangler_processor,
          inputs=inputs,
          outputs=outputs,
          job_arguments=job_argument,
      )
  
      pipeline = Pipeline(
          name=pipeline_name,
          parameters=[instance_count, instance_type],
          steps=[data_wrangler_step],
          sagemaker_session=sagemaker_session,
      )
  
      definition = json.loads(pipeline.definition())
      expected_image_uri = image_uris.retrieve(
          "data-wrangler", region=sagemaker_session.boto_region_name
      )
      assert len(definition["Steps"]) == 1
      assert (
          definition["Steps"][0]["Arguments"]["AppSpecification"][
              "ImageUri"
          ]
          is not None
      )
      assert (
          definition["Steps"][0]["Arguments"]["AppSpecification"][
              "ImageUri"
          ]
          == expected_image_uri
      )
  
      assert (
          definition["Steps"][0]["Arguments"]["ProcessingInputs"]
          is not None
      )
      processing_inputs = definition["Steps"][0]["Arguments"][
          "ProcessingInputs"
      ]
      assert len(processing_inputs) == 2
      for processing_input in processing_inputs:
          if processing_input["InputName"] == "flow":
              assert processing_input["S3Input"]["S3Uri"].endswith(
                  ".flow"
              )
              assert (
                  processing_input["S3Input"]["LocalPath"]
                  == "/opt/ml/processing/flow"
              )
          elif processing_input["InputName"] == "dummy_data.csv":
              assert processing_input["S3Input"]["S3Uri"].endswith(
                  ".csv"
              )
              assert (
                  processing_input["S3Input"]["LocalPath"]
                  == "/opt/ml/processing/dummy_data.csv"
              )
          else:
              raise AssertionError("Unknown input name")
      assert (
          definition["Steps"][0]["Arguments"]["ProcessingOutputConfig"]
          is not None
      )
      processing_outputs = definition["Steps"][0]["Arguments"][
          "ProcessingOutputConfig"
      ]["Outputs"]
      assert len(processing_outputs) == 1
      assert processing_outputs[0]["OutputName"] == output_name
      assert processing_outputs[0]["S3Output"] is not None
      assert (
          processing_outputs[0]["S3Output"]["LocalPath"]
          == "/opt/ml/processing/output"
      )
      assert processing_outputs[0]["S3Output"]["S3Uri"] == output_s3_uri
  
      try:
          response = pipeline.create(role)
          create_arn = response["PipelineArn"]
  
          execution = pipeline.start()
          response = execution.describe()
          assert response["PipelineArn"] == create_arn
  
          wait_pipeline_execution(
              execution=execution, delay=60, max_attempts=10
          )
  
          execution_steps = execution.list_steps()
          assert len(execution_steps) == 1
          assert execution_steps[0]["StepName"] == "data-wrangler-step"
      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 13 for calling a global function with example usage: test_one_step_data_wrangler_processing_pipeline(sagemaker_session, role, pipeline_name) without return types