code: def test_batch_transform_data_quality_step_pipeline_happycase(
      sagemaker_session,
      role,
      pipeline_name,
      check_job_config,
      supplied_baseline_statistics_uri_param,
      supplied_baseline_constraints_uri_param,
      data_quality_check_config,
      data_quality_supplied_baseline_statistics,
      transform_args,
      step_model_create,
      pipeline_session,
  ):
      data_quality_supplied_baseline_constraints = Constraints.from_file_path(
          constraints_file_path=os.path.join(
              DATA_DIR,
              "pipeline/quality_check_step/data_quality/good_cases/constraints.json",
          ),
          sagemaker_session=sagemaker_session,
      ).file_s3_uri
      data_quality_check_step = MonitorBatchTransformStep(
          name="MonitorBatchTransformStep",
          transform_step_args=transform_args,
          monitor_configuration=data_quality_check_config,
          check_job_configuration=check_job_config,
          supplied_baseline_statistics=supplied_baseline_statistics_uri_param,
          supplied_baseline_constraints=supplied_baseline_constraints_uri_param,
          fail_on_violation=True,
      )
      pipeline = Pipeline(
          name=pipeline_name,
          steps=[step_model_create, data_quality_check_step],
          parameters=[
              supplied_baseline_statistics_uri_param,
              supplied_baseline_constraints_uri_param,
          ],
          sagemaker_session=pipeline_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": data_quality_supplied_baseline_statistics,
                      "SuppliedBaselineConstraintsUri": data_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) == 3
              for execution_step in execution_steps:
                  failure_reason = execution_step.get(
                      "FailureReason", ""
                  )
                  if failure_reason != "":
                      logging.error(
                          f"Pipeline execution failed with error: {failure_reason}. Retrying.."
                      )
                      continue
  
              for execution_step in execution_steps:
                  assert execution_step["StepStatus"] == "Succeeded"
                  if (
                      execution_step["StepName"]
                      == "MonitorBatchTransformStep-transform"
                  ):
                      transform_execution_end_time = execution_step[
                          "EndTime"
                      ]
                  if (
                      execution_step["StepName"]
                      == "MonitorBatchTransformStep-monitoring"
                  ):
                      monitoring_execution_start_time = execution_step[
                          "StartTime"
                      ]
                      data_qual_metadata = execution_step["Metadata"][
                          "QualityCheck"
                      ]
                      assert not data_qual_metadata["SkipCheck"]
                      assert not data_qual_metadata[
                          "RegisterNewBaseline"
                      ]
                      assert not data_qual_metadata.get(
                          "ViolationReport", ""
                      )
                      assert (
                          data_qual_metadata[
                              "BaselineUsedForDriftCheckConstraints"
                          ]
                          == data_quality_supplied_baseline_constraints
                      )
                      assert (
                          data_qual_metadata[
                              "BaselineUsedForDriftCheckStatistics"
                          ]
                          == data_quality_supplied_baseline_statistics
                      )
                      assert (
                          data_qual_metadata[
                              "BaselineUsedForDriftCheckConstraints"
                          ]
                          != data_qual_metadata[
                              "CalculatedBaselineConstraints"
                          ]
                      )
                      assert (
                          data_qual_metadata[
                              "BaselineUsedForDriftCheckStatistics"
                          ]
                          != data_qual_metadata[
                              "CalculatedBaselineStatistics"
                          ]
                      )
              assert (
                  monitoring_execution_start_time
                  > transform_execution_end_time
              )
              break
      finally:
          try:
              pipeline.delete()
          except Exception:
              pass
  
imports: from __future__ import absolute_import
  import json
  import logging
  import os
  import pytest
  from tests.integ.sagemaker.workflow.helpers import wait_pipeline_execution
  from sagemaker.clarify import BiasConfig, DataConfig
  from sagemaker.s3 import S3Uploader, S3Downloader
  from sagemaker.workflow.clarify_check_step import DataBiasCheckConfig
  from sagemaker.workflow.monitor_batch_transform_step import MonitorBatchTransformStep
  from sagemaker.workflow.model_step import ModelStep
  from sagemaker.workflow.parameters import ParameterString
  from tests.integ import DATA_DIR
  from sagemaker.model_monitor import DatasetFormat, Statistics, Constraints
  from sagemaker.xgboost import XGBoostModel
  from sagemaker.transformer import Transformer
  from sagemaker import get_execution_role, utils
  from sagemaker.workflow.pipeline import Pipeline
  from sagemaker.workflow.check_job_config import CheckJobConfig
  from tests.integ.retry import retries
  from sagemaker.workflow.quality_check_step import DataQualityCheckConfig, ModelQualityCheckConfig
label_desc: Use image node_id 21 for calling a global function with example usage: test_batch_transform_data_quality_step_pipeline_happycase(sagemaker_session, role, pipeline_name, check_job_config, supplied_baseline_statistics_uri_param, supplied_baseline_constraints_uri_param, data_quality_check_config, data_quality_supplied_baseline_statistics, transform_args, step_model_create, pipeline_session) without return types