code: def test_to_pipeline(
      get_execution_role,
      session,
      mock_get_pipeline_lineage_names,
      mock_create_lineage,
      mock_upload_callable,
      mock_runtime_manager,
      mock_script_upload,
      mock_dependency_upload,
      mock_spark_dependency_upload,
      mock_estimator,
      mock_training_step,
      mock_training_input,
      mock_spark_image,
      pipeline,
      lineage_validator,
  ):
      session.sagemaker_config = None
      session.boto_region_name = TEST_REGION
      session.expand_role.return_value = EXECUTION_ROLE_ARN
      spark_config = SparkConfig(
          submit_files=["file_a", "file_b", "file_c"]
      )
      job_settings = _JobSettings(
          spark_config=spark_config,
          s3_root_uri=S3_URI,
          role=EXECUTION_ROLE_ARN,
          include_local_workdir=True,
          instance_type="ml.m5.large",
          encrypt_inter_container_traffic=True,
          sagemaker_session=session,
      )
      jobs_container_entrypoint = [
          "/bin/bash",
          f"/opt/ml/input/data/{RUNTIME_SCRIPTS_CHANNEL_NAME}/{ENTRYPOINT_SCRIPT_NAME}",
      ]
      jobs_container_entrypoint.extend(["--jars", "path_a"])
      jobs_container_entrypoint.extend(["--py-files", "path_b"])
      jobs_container_entrypoint.extend(["--files", "path_c"])
      jobs_container_entrypoint.extend([SPARK_APP_SCRIPT_PATH])
      container_args = ["--s3_base_uri", f"{S3_URI}/pipeline_name"]
      container_args.extend(["--region", session.boto_region_name])
  
      mock_feature_processor_config = Mock(
          mode=FeatureProcessorMode.PYSPARK,
          inputs=[tdh.FEATURE_PROCESSOR_INPUTS],
          output="some_fg",
      )
      mock_feature_processor_config.mode.return_value = (
          FeatureProcessorMode.PYSPARK
      )
  
      wrapped_func = Mock(
          Callable,
          feature_processor_config=mock_feature_processor_config,
          job_settings=job_settings,
          wrapped_func=job_function,
      )
      wrapped_func.feature_processor_config.return_value = (
          mock_feature_processor_config
      )
      wrapped_func.job_settings.return_value = job_settings
      wrapped_func.wrapped_func.return_value = job_function
  
      pipeline_arn = to_pipeline(
          pipeline_name="pipeline_name",
          step=wrapped_func,
          role=EXECUTION_ROLE_ARN,
          max_retries=1,
          tags=[
              ("tag_key_1", "tag_value_1"),
              ("tag_key_2", "tag_value_2"),
          ],
          sagemaker_session=session,
      )
      assert pipeline_arn == PIPELINE_ARN
  
      assert mock_upload_callable.called_once_with(job_function)
      local_dependencies_path = mock_runtime_manager().snapshot()
      mock_python_version = (
          mock_runtime_manager()._current_python_version()
      )
      container_args.extend(
          ["--client_python_version", mock_python_version]
      )
  
      mock_script_upload.assert_called_once_with(
          spark_config,
          f"{S3_URI}/pipeline_name",
          None,
          session,
      )
  
      mock_dependency_upload.assert_called_once_with(
          local_dependencies_path,
          True,
          None,
          None,
          f"{S3_URI}/pipeline_name",
          None,
          session,
          None,
      )
  
      mock_spark_dependency_upload.assert_called_once_with(
          spark_config,
          f"{S3_URI}/pipeline_name",
          None,
          session,
      )
  
      mock_estimator.assert_called_once_with(
          role=EXECUTION_ROLE_ARN,
          max_run=86400,
          max_retry_attempts=1,
          output_path=f"{S3_URI}/pipeline_name",
          output_kms_key=None,
          image_uri="some_image_uri",
          input_mode="File",
          environment={
              "AWS_DEFAULT_REGION": "us-west-2",
              "REMOTE_FUNCTION_SECRET_KEY": job_settings.environment_variables[
                  "REMOTE_FUNCTION_SECRET_KEY"
              ],
              "scheduled_time": Parameter(
                  name="scheduled_time",
                  parameter_type=ParameterTypeEnum.STRING,
              ),
          },
          volume_size=30,
          volume_kms_key=None,
          instance_count=1,
          instance_type="ml.m5.large",
          encrypt_inter_container_traffic=True,
          container_entry_point=jobs_container_entrypoint,
          container_arguments=container_args,
          tags=[],
      )
  
      mock_training_step.assert_called_once_with(
          name="-".join(["pipeline_name", "feature-processor"]),
          estimator=mock_estimator(),
          inputs={
              RUNTIME_SCRIPTS_CHANNEL_NAME: mock_training_input(
                  s3_data=mock_script_upload.return_value,
                  s3_data_type="S3Prefix",
              ),
              REMOTE_FUNCTION_WORKSPACE: mock_training_input(
                  s3_data=f"{S3_URI}/pipeline_name/sm_rf_user_ws",
                  s3_data_type="S3Prefix",
              ),
              SPARK_CONF_CHANNEL_NAME: mock_training_input(
                  s3_data="path_d", s3_data_type="S3Prefix"
              ),
          },
          retry_policies=[
              StepRetryPolicy(
                  exception_types=[
                      StepExceptionTypeEnum.SERVICE_FAULT,
                      StepExceptionTypeEnum.THROTTLING,
                  ],
                  max_attempts=1,
              ),
              SageMakerJobStepRetryPolicy(
                  exception_types=[
                      SageMakerJobExceptionTypeEnum.INTERNAL_ERROR,
                      SageMakerJobExceptionTypeEnum.CAPACITY_ERROR,
                      SageMakerJobExceptionTypeEnum.RESOURCE_LIMIT,
                  ],
                  max_attempts=1,
              ),
          ],
      )
  
      pipeline.assert_called_once_with(
          name="pipeline_name",
          steps=[mock_training_step()],
          sagemaker_session=session,
          parameters=[
              Parameter(
                  name="scheduled_time",
                  parameter_type=ParameterTypeEnum.STRING,
              )
          ],
      )
  
      pipeline().upsert.assert_has_calls(
          [
              call(
                  role_arn=EXECUTION_ROLE_ARN,
                  tags=[
                      dict(
                          Key=FEATURE_PROCESSOR_TAG_KEY,
                          Value=FEATURE_PROCESSOR_TAG_VALUE,
                      ),
                      dict(Key="tag_key_1", Value="tag_value_1"),
                      dict(Key="tag_key_2", Value="tag_value_2"),
                  ],
              ),
              call(
                  role_arn=EXECUTION_ROLE_ARN,
                  tags=[
                      {
                          "Key": PIPELINE_CONTEXT_NAME_TAG_KEY,
                          "Value": "pipeline-context-name",
                      },
                      {
                          "Key": PIPELINE_VERSION_CONTEXT_NAME_TAG_KEY,
                          "Value": "pipeline-version-context-name",
                      },
                  ],
              ),
          ]
      )
  
imports: from __future__ import absolute_import
  from datetime import datetime
  from typing import Callable
  import pytest
  import json
  from botocore.exceptions import ClientError
  from mock import Mock, patch, call
  from sagemaker.feature_store.feature_processor.feature_scheduler import FeatureProcessorLineageHandler
  from sagemaker.feature_store.feature_processor import FeatureProcessorPipelineEvents, FeatureProcessorPipelineExecutionStatus
  from sagemaker.lineage.context import Context
  from sagemaker.remote_function.spark_config import SparkConfig
  from sagemaker import Session
  from sagemaker.feature_store.feature_processor._enums import FeatureProcessorMode
  from sagemaker.feature_store.feature_processor._constants import FEATURE_PROCESSOR_TAG_KEY, FEATURE_PROCESSOR_TAG_VALUE, EXECUTION_TIME_PIPELINE_PARAMETER_FORMAT, PIPELINE_NAME_MAXIMUM_LENGTH
  from sagemaker.feature_store.feature_processor.feature_scheduler import schedule, to_pipeline, execute, delete_schedule, describe, list_pipelines, put_trigger, enable_trigger, disable_trigger, delete_trigger, _validate_fg_lineage_resources, _validate_pipeline_lineage_resources
  from sagemaker.remote_function.job import _JobSettings, SPARK_APP_SCRIPT_PATH, RUNTIME_SCRIPTS_CHANNEL_NAME, REMOTE_FUNCTION_WORKSPACE, ENTRYPOINT_SCRIPT_NAME, SPARK_CONF_CHANNEL_NAME
  from sagemaker.workflow.parameters import Parameter, ParameterTypeEnum
  from sagemaker.workflow.retry import StepRetryPolicy, StepExceptionTypeEnum, SageMakerJobStepRetryPolicy, SageMakerJobExceptionTypeEnum
  import test_data_helpers
label_desc: Use image node_id 8 for calling a global function with example usage: test_to_pipeline(get_execution_role, session, mock_get_pipeline_lineage_names, mock_create_lineage, mock_upload_callable, mock_runtime_manager, mock_script_upload, mock_dependency_upload, mock_spark_dependency_upload, mock_estimator, mock_training_step, mock_training_input, mock_spark_image, pipeline, lineage_validator) without return types