code: def test_one_step_model_quality_pipeline_constraint_violation(
      sagemaker_session,
      role,
      pipeline_name,
      check_job_config,
      supplied_baseline_statistics_uri_param,
      supplied_baseline_constraints_uri_param,
      model_quality_check_config,
      model_quality_supplied_baseline_statistics,
  ):
      model_quality_supplied_baseline_constraints = Constraints.from_file_path(
          constraints_file_path=os.path.join(
              DATA_DIR,
              "pipeline/quality_check_step/model_quality/bad_cases/constraints.json",
          ),
          sagemaker_session=sagemaker_session,
      ).file_s3_uri
      model_quality_check_step = QualityCheckStep(
          name="ModelQualityCheckStep",
          register_new_baseline=False,
          skip_check=False,
          quality_check_config=model_quality_check_config,
          check_job_config=check_job_config,
          supplied_baseline_statistics=supplied_baseline_statistics_uri_param,
          supplied_baseline_constraints=supplied_baseline_constraints_uri_param,
      )
      pipeline = Pipeline(
          name=pipeline_name,
          steps=[model_quality_check_step],
          parameters=[
              supplied_baseline_statistics_uri_param,
              supplied_baseline_constraints_uri_param,
          ],
          sagemaker_session=sagemaker_session,
      )
  
      try:
          response = pipeline.create(role)
          create_arn = response["PipelineArn"]
  
          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={
                      "SuppliedBaselineStatisticsUri": model_quality_supplied_baseline_statistics,
                      "SuppliedBaselineConstraintsUri": model_quality_supplied_baseline_constraints,
                  }
              )
              response = execution.describe()
  
              assert response["PipelineArn"] == create_arn
  
              wait_pipeline_execution(execution=execution)
              execution_steps = execution.list_steps()
  
              assert len(execution_steps) == 1
              failure_reason = execution_steps[0].get(
                  "FailureReason", ""
              )
              if _CHECK_FAIL_ERROR_MSG not in failure_reason:
                  logging.error(
                      f"Pipeline execution failed with error: {failure_reason}. Retrying.."
                  )
                  continue
              assert (
                  execution_steps[0]["StepName"]
                  == "ModelQualityCheckStep"
              )
              assert execution_steps[0]["StepStatus"] == "Failed"
              break
      finally:
          try:
              pipeline.delete()
          except Exception:
              pass
  
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.parameters import ParameterString
  from tests.integ import DATA_DIR
  from sagemaker import get_execution_role, utils
  from sagemaker.model_monitor import DatasetFormat, Statistics, Constraints
  from sagemaker.workflow.pipeline import Pipeline
  from sagemaker.workflow.quality_check_step import QualityCheckStep, DataQualityCheckConfig, ModelQualityCheckConfig
  from sagemaker.workflow.check_job_config import CheckJobConfig
  from tests.integ.retry import retries
label_desc: Use image node_id 15 for calling a global function with example usage: test_one_step_model_quality_pipeline_constraint_violation(sagemaker_session, role, pipeline_name, check_job_config, supplied_baseline_statistics_uri_param, supplied_baseline_constraints_uri_param, model_quality_check_config, model_quality_supplied_baseline_statistics) without return types