code: def test_one_step_ingestion_pipeline(
      sagemaker_session,
      feature_store_session,
      feature_definitions,
      role,
      pipeline_name,
  ):
      instance_count = ParameterInteger(
          name="InstanceCount", default_value=1
      )
      instance_type = ParameterString(
          name="InstanceType", default_value="ml.m5.4xlarge"
      )
  
      input_name = "features.csv"
      input_file_path = os.path.join(
          DATA_DIR, "workflow", "features.csv"
      )
      input_data_uri = os.path.join(
          "s3://",
          sagemaker_session.default_bucket(),
          "py-sdk-ingestion-test-input/features.csv",
      )
  
      with open(input_file_path, "r") as data:
          body = data.read()
          S3Uploader.upload_string_as_file_body(
              body=body,
              desired_s3_uri=input_data_uri,
              sagemaker_session=sagemaker_session,
          )
  
      inputs = [
          ProcessingInput(
              input_name=input_name,
              source=input_data_uri,
              destination="/opt/ml/processing/features.csv",
          )
      ]
  
      feature_group_name = f"py-sdk-integ-fg-{int(time.time() * 10**7)}"
      feature_group = FeatureGroup(
          name=feature_group_name,
          feature_definitions=feature_definitions,
          sagemaker_session=feature_store_session,
      )
  
      (
          ingestion_only_flow,
          output_name,
      ) = generate_data_ingestion_flow_from_s3_input(
          input_name,
          input_data_uri,
          s3_content_type="csv",
          s3_has_header=True,
      )
  
      outputs = [
          ProcessingOutput(
              output_name=output_name,
              app_managed=True,
              feature_store_output=FeatureStoreOutput(
                  feature_group_name=feature_group_name
              ),
          )
      ]
  
      output_content_type = "CSV"
      output_config = {
          output_name: {"content_type": output_content_type}
      }
      job_argument = [f"--output-config '{json.dumps(output_config)}'"]
  
      temp_flow_path = "./ingestion.flow"
      with cleanup_feature_group(feature_group):
          json.dump(ingestion_only_flow, open(temp_flow_path, "w"))
  
          data_wrangler_processor = DataWranglerProcessor(
              role=role,
              data_wrangler_flow_source=temp_flow_path,
              instance_count=instance_count,
              instance_type=instance_type,
              sagemaker_session=sagemaker_session,
              max_runtime_in_seconds=86400,
          )
  
          data_wrangler_step = ProcessingStep(
              name="ingestion-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,
          )
  
          try:
              response = pipeline.create(role)
              create_arn = response["PipelineArn"]
  
              offline_store_s3_uri = os.path.join(
                  "s3://",
                  sagemaker_session.default_bucket(),
                  feature_group_name,
              )
              feature_group.create(
                  s3_uri=offline_store_s3_uri,
                  record_identifier_name="f11",
                  event_time_feature_name="f10",
                  role_arn=role,
                  enable_online_store=False,
              )
              _wait_for_feature_group_create(feature_group)
  
              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"] == "ingestion-step"
              assert execution_steps[0]["StepStatus"] == "Succeeded"
  
              athena_query = feature_group.athena_query()
              with timeout(minutes=10):
                  athena_query.run(
                      query_string=f'SELECT * FROM "{athena_query.table_name}"',
                      output_location=f"{offline_store_s3_uri}/query_results",
                  )
                  athena_query.wait()
                  assert (
                      "SUCCEEDED"
                      == athena_query.get_query_execution()
                      .get("QueryExecution")
                      .get("Status")
                      .get("State")
                  )
  
                  df = athena_query.as_dataframe()
                  assert (
                      pd.read_csv(input_file_path).shape[0]
                      == df.shape[0]
                  )
          finally:
              try:
                  pipeline.delete()
              except Exception as e:
                  print(f"Delete pipeline failed with error: {e}")
              os.remove(temp_flow_path)
  
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 10 for calling a global function with example usage: test_one_step_ingestion_pipeline(sagemaker_session, feature_store_session, feature_definitions, role, pipeline_name) without return types