code: def test_pytorch_training_model_registration_and_creation_with_custom_inference(
      pipeline_session,
      role,
      pipeline_name,
  ):
      kms_key = get_or_create_kms_key(pipeline_session, role)
      base_dir = os.path.join(DATA_DIR, "pytorch_mnist")
      entry_point = os.path.join(base_dir, "mnist.py")
      input_path = pipeline_session.upload_data(
          path=os.path.join(base_dir, "training"),
          key_prefix="integ-test-data/pytorch_mnist/training",
      )
      inputs = TrainingInput(s3_data=input_path)
  
      instance_count = ParameterInteger(
          name="InstanceCount", default_value=1
      )
  
      # If image_uri is not provided, the instance_type should not be a pipeline variable
      # since instance_type is used to retrieve image_uri in compile time (PySDK)
      pytorch_estimator = PyTorch(
          entry_point=entry_point,
          role=role,
          framework_version="1.5.0",
          py_version="py3",
          instance_count=instance_count,
          instance_type="ml.m5.xlarge",
          sagemaker_session=pipeline_session,
          output_kms_key=kms_key,
      )
      train_step_args = pytorch_estimator.fit(inputs=inputs)
      step_train = TrainingStep(
          name="pytorch-train",
          step_args=train_step_args,
      )
      model = Model(
          name="MyModel",
          image_uri=pytorch_estimator.training_image_uri(),
          model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
          sagemaker_session=pipeline_session,
          role=role,
          entry_point=entry_point,
          source_dir=base_dir,
          model_kms_key=kms_key,
      )
      # register model with runtime repack
      regis_model_step_args = model.register(
          content_types=["text/csv"],
          response_types=["text/csv"],
          inference_instances=["ml.t2.medium", "ml.m5.large"],
          transform_instances=["ml.m5.large"],
          description="test-description",
          model_package_group_name=f"{pipeline_name}TestModelPackageGroup",
      )
      step_model_regis = ModelStep(
          name="pytorch-register-model",
          step_args=regis_model_step_args,
      )
      create_model_step_args = model.create(
          instance_type="ml.m5.large",
          accelerator_type="ml.eia1.medium",
      )
      step_model_create = ModelStep(
          name="pytorch-model",
          step_args=create_model_step_args,
      )
      # Use FailStep error_message to reference model step properties
      step_fail = FailStep(
          name="fail-step",
          error_message=Join(
              on=", ",
              values=[
                  "Fail the execution on purpose to check model step properties",
                  "register model",
                  step_model_regis.properties.ModelApprovalStatus,
                  "create model",
                  step_model_create.properties.ModelName,
              ],
          ),
      )
      pipeline = Pipeline(
          name=pipeline_name,
          parameters=[instance_count],
          steps=[
              step_train,
              step_model_regis,
              step_model_create,
              step_fail,
          ],
          sagemaker_session=pipeline_session,
      )
  
      try:
          pipeline.create(role)
  
          for _ in retries(
              max_retry_count=5,
              exception_message_prefix="Waiting for a successful execution of pipeline",
              seconds_to_sleep=10,
          ):
              execution = pipeline.start(parameters={})
              wait_pipeline_execution(execution=execution)
              execution_steps = execution.list_steps()
              is_execution_fail = False
              for step in execution_steps:
                  if step["StepName"] == "fail-step":
                      assert step["StepStatus"] == "Failed"
                      assert (
                          "PendingManualApproval"
                          in step["FailureReason"]
                      )
                      assert "pytorch-model" in step["FailureReason"]
                      continue
                  failure_reason = step.get("FailureReason", "")
                  if failure_reason != "":
                      logging.error(
                          f"Pipeline execution failed with error: {failure_reason}."
                          " Retrying.."
                      )
                      is_execution_fail = True
                      break
                  assert step["StepStatus"] == "Succeeded"
                  if _REGISTER_MODEL_NAME_BASE in step["StepName"]:
                      assert step["Metadata"][_REGISTER_MODEL_TYPE]
                  if _CREATE_MODEL_NAME_BASE in step["StepName"]:
                      assert step["Metadata"][_CREATE_MODEL_TYPE]
              if is_execution_fail:
                  continue
              assert len(execution_steps) == 6
              break
      finally:
          try:
              pipeline.delete()
          except Exception as error:
              logging.error(error)
  
imports: from __future__ import absolute_import
  import logging
  import os
  import pytest
  from tests.integ.sagemaker.workflow.helpers import wait_pipeline_execution
  from sagemaker.workflow.fail_step import FailStep
  from sagemaker.workflow.functions import Join
  from tests.integ.timeout import timeout_and_delete_endpoint_by_name
  from sagemaker.tensorflow import TensorFlow, TensorFlowModel, TensorFlowPredictor
  from sagemaker.utils import unique_name_from_base
  from sagemaker.workflow.model_step import ModelStep, _REGISTER_MODEL_NAME_BASE, _CREATE_MODEL_NAME_BASE
  from tests.integ.retry import retries
  from sagemaker.drift_check_baselines import DriftCheckBaselines
  from sagemaker import PipelineModel, TrainingInput, Model, ModelMetrics, MetricsSource, get_execution_role, ModelPackage
  from sagemaker import FileSource, utils
  from sagemaker.pytorch import PyTorch
  from sagemaker.s3 import S3Uploader
  from sagemaker.mxnet.model import MXNetModel
  from sagemaker.workflow.parameters import ParameterInteger, ParameterString
  from sagemaker.workflow.pipeline import Pipeline
  from sagemaker.workflow.steps import TrainingStep
  from sagemaker.xgboost import XGBoost, XGBoostModel, XGBoostPredictor
  from tests.integ.kms_utils import get_or_create_kms_key
  from tests.integ import DATA_DIR
label_desc: Use image node_id 4 for calling a global function with example usage: test_pytorch_training_model_registration_and_creation_with_custom_inference(pipeline_session, role, pipeline_name) without return types