code: def test_training_job_with_debugger_and_profiler(
      sagemaker_session,
      pipeline_name,
      role,
      pytorch_training_latest_version,
      pytorch_training_latest_py_version,
  ):
      instance_count = ParameterInteger(
          name="InstanceCount", default_value=1
      )
      instance_type = "ml.m5.xlarge"
  
      rules = [
          Rule.sagemaker(rule_configs.vanishing_gradient()),
          Rule.sagemaker(
              base_config=rule_configs.all_zero(),
              rule_parameters={"tensor_regex": ".*"},
          ),
          Rule.sagemaker(rule_configs.loss_not_decreasing()),
      ]
      debugger_hook_config = DebuggerHookConfig(
          s3_output_path=f"s3://{sagemaker_session.default_bucket()}/{uuid.uuid4()}/tensors"
      )
  
      base_dir = os.path.join(DATA_DIR, "pytorch_mnist")
      entry_point = "mnist.py"
      source_dir = sagemaker_session.upload_data(
          path=os.path.join(
              base_dir, "pytorch_mnist_source_code.tar.gz"
          ),
          key_prefix="integ-test-data/pytorch_mnist/training",
      )
      entry_point_param = ParameterString(name="EntryPoint")
      source_dir_param = ParameterString(name="SourceDir")
      input_path = sagemaker_session.upload_data(
          path=os.path.join(base_dir, "training"),
          key_prefix="integ-test-data/pytorch_mnist/training",
      )
      inputs = TrainingInput(s3_data=input_path)
  
      # 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_param,
          source_dir=source_dir_param,
          role="SageMakerRole",
          framework_version=pytorch_training_latest_version,
          py_version=pytorch_training_latest_py_version,
          instance_count=instance_count,
          instance_type=instance_type,
          sagemaker_session=sagemaker_session,
          rules=rules,
          debugger_hook_config=debugger_hook_config,
          # TODO: remove base_job_name once we merge
          # https://github.com/aws/sagemaker-python-sdk/pull/3158/files
          base_job_name="TestJob",
      )
  
      step_train = TrainingStep(
          name="pytorch-train",
          estimator=pytorch_estimator,
          inputs=inputs,
      )
  
      pipeline = Pipeline(
          name=pipeline_name,
          parameters=[
              instance_count,
              entry_point_param,
              source_dir_param,
          ],
          steps=[step_train],
          sagemaker_session=sagemaker_session,
      )
  
      try:
          pipeline.create(role)
          execution_steps = _start_and_verify_execution_with_retry(
              pipeline=pipeline,
              parameters={
                  "EntryPoint": entry_point,
                  "SourceDir": source_dir,
              },
          )
          training_job_arn = execution_steps[0]["Metadata"][
              "TrainingJob"
          ]["Arn"]
          job_description = (
              sagemaker_session.sagemaker_client.describe_training_job(
                  TrainingJobName=training_job_arn.split("/")[1]
              )
          )
  
          for index, rule in enumerate(rules):
              config = job_description["DebugRuleConfigurations"][index]
              assert config["RuleConfigurationName"] == rule.name
              assert config["RuleEvaluatorImage"] == rule.image_uri
              assert config["VolumeSizeInGB"] == 0
              assert (
                  config["RuleParameters"]["rule_to_invoke"]
                  == rule.rule_parameters["rule_to_invoke"]
              )
          assert (
              job_description["DebugHookConfig"]
              == debugger_hook_config._to_request_dict()
          )
          assert job_description["ProfilingStatus"] == "Enabled"
          assert (
              job_description["ProfilerConfig"][
                  "ProfilingIntervalInMilliseconds"
              ]
              == 500
          )
      finally:
          try:
              pipeline.delete()
          except Exception as error:
              logging.error(error)
  
imports: from __future__ import absolute_import
  import os
  import uuid
  import logging
  import pytest
  from tests.integ.sagemaker.workflow.helpers import wait_pipeline_execution
  from sagemaker import TrainingInput, get_execution_role, utils, image_uris
  from sagemaker.debugger import DebuggerHookConfig, Rule, rule_configs
  from sagemaker.estimator import Estimator
  from sagemaker.pytorch.estimator import PyTorch
  from sagemaker.tensorflow import TensorFlow
  from sagemaker.utils import sagemaker_timestamp
  from sagemaker.workflow.functions import Join
  from sagemaker.workflow.parameters import ParameterInteger, ParameterString
  from sagemaker.workflow.pipeline import Pipeline
  from sagemaker.workflow.steps import TrainingStep
  from tests.integ.retry import retries
  from tests.integ import DATA_DIR
label_desc: Use image node_id 4 for calling a global function with example usage: test_training_job_with_debugger_and_profiler(sagemaker_session, pipeline_name, role, pytorch_training_latest_version, pytorch_training_latest_py_version) without return types