Repository: getmoto/moto
Pull request #7893 (2024-07-25 13:57:29)

Feature: Sagemaker Model Bias Job Definitions
Adds support to Sagemaker Model Bias Job Definitions (create, describe, list, delete)


Patch:
diff --git a/moto/applicationautoscaling/models.py b/moto/applicationautoscaling/models.py
index 426069f8c..2ed839ab8 100644
--- a/moto/applicationautoscaling/models.py
+++ b/moto/applicationautoscaling/models.py
@@ -308,6 +308,7 @@ class ApplicationAutoscalingBackend(BaseBackend):
                 a
                 for a in self.scheduled_actions
                 if a.service_namespace == service_namespace
+                and a.scheduled_action_name == scheduled_action_name
                 and a.resource_id == resource_id
                 and a.scalable_dimension == scalable_dimension
             ),
@@ -640,7 +641,7 @@ class FakeScheduledAction(BaseModel):
         account_id: str,
         region: str,
     ) -> None:
-        self.arn = f"arn:{get_partition(region)}:autoscaling:{region}:{account_id}:scheduledAction:{service_namespace}:scheduledActionName/{scheduled_action_name}"
+        self.arn = f"arn:{get_partition(region)}:autoscaling:{region}:{account_id}:scheduledAction:{service_namespace}/{resource_id}:scheduledActionName/{scheduled_action_name}"
         self.service_namespace = service_namespace
         self.schedule = schedule
         self.timezone = timezone
diff --git a/moto/sagemaker/models.py b/moto/sagemaker/models.py
index 5b97304ed..e0bdbb939 100644
--- a/moto/sagemaker/models.py
+++ b/moto/sagemaker/models.py
@@ -148,6 +148,12 @@ PAGINATION_MODEL = {
         "limit_default": 100,
         "unique_attribute": "model_card_arn",
     },
+    "list_model_bias_job_definitions": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "unique_attribute": "arn",
+    },
 }
 
 METRIC_INFO_TYPE = Dict[str, Union[str, int, float, datetime]]
@@ -2864,6 +2870,7 @@ class SageMakerModelBackend(BaseBackend):
         self.model_package_name_mapping: Dict[str, str] = {}
         self.feature_groups: Dict[str, FeatureGroup] = {}
         self.clusters: Dict[str, Cluster] = {}
+        self.model_bias_job_definitions: Dict[str, FakeModelBiasJobDefinition] = {}
         self.auto_ml_jobs: Dict[str, AutoMLJob] = {}
         self.compilation_jobs: Dict[str, CompilationJob] = {}
         self.domains: Dict[str, Domain] = {}
@@ -3002,6 +3009,7 @@ class SageMakerModelBackend(BaseBackend):
             "pipeline": self.pipelines,
             "model-package-group": self.model_package_groups,
             "cluster": self.clusters,
+            "model-bias-job-definition": self.model_bias_job_definitions,
             "automl-job": self.auto_ml_jobs,
             "compilation-job": self.compilation_jobs,
             "domain": self.domains,
@@ -4848,6 +4856,57 @@ class SageMakerModelBackend(BaseBackend):
             )
         return nodes_list
 
+    def create_model_bias_job_definition(
+        self,
+        account_id: str,
+        job_definition_name: str,
+        tags: List[Dict[str, str]] = [],
+        role_arn: str = "",
+        job_resources: Optional[Dict[str, Any]] = None,
+        stopping_condition: Optional[Dict[str, Any]] = None,
+        environment: Optional[Dict[str, str]] = None,
+        network_config: Optional[Dict[str, Any]] = None,
+        model_bias_baseline_config: Optional[Dict[str, Any]] = None,
+        model_bias_app_specification: Optional[Dict[str, Any]] = None,
+        model_bias_job_input: Optional[Dict[str, Any]] = None,
+        model_bias_job_output_config: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, str]:
+        job_definition = FakeModelBiasJobDefinition(
+            account_id=account_id,
+            region_name=self.region_name,
+            job_definition_name=job_definition_name,
+            tags=tags,
+            role_arn=role_arn,
+            job_resources=job_resources,
+            stopping_condition=stopping_condition,
+            environment=environment,
+            network_config=network_config,
+            model_bias_baseline_config=model_bias_baseline_config,
+            model_bias_app_specification=model_bias_app_specification,
+            model_bias_job_input=model_bias_job_input,
+            model_bias_job_output_config=model_bias_job_output_config,
+        )
+        self.model_bias_job_definitions[job_definition_name] = job_definition
+        return job_definition.response_create
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_model_bias_job_definitions(self) -> List[Dict[str, str]]:
+        return [job.summary_object for job in self.model_bias_job_definitions.values()]
+
+    def describe_model_bias_job_definition(
+        self, job_definition_name: str
+    ) -> Dict[str, Any]:
+        job_definition = self.model_bias_job_definitions.get(job_definition_name)
+        if job_definition is None:
+            raise ResourceNotFound(f"Job definition {job_definition_name} not found")
+        return job_definition.response_object
+
+    def delete_model_bias_job_definition(self, job_definition_name: str) -> None:
+        if job_definition_name in self.model_bias_job_definitions:
+            del self.model_bias_job_definitions[job_definition_name]
+        else:
+            raise ResourceNotFound(f"Job definition {job_definition_name} not found")
+
     def create_auto_ml_job_v2(
         self,
         auto_ml_job_name: str,
@@ -5820,4 +5879,68 @@ class FakeTrialComponent(BaseObject):
         )
 
 
+class FakeModelBiasJobDefinition(BaseObject):
+    def __init__(
+        self,
+        account_id: str,
+        region_name: str,
+        job_definition_name: str,
+        tags: List[Dict[str, str]] = [],
+        role_arn: str = "",
+        job_resources: Optional[Dict[str, Any]] = None,
+        stopping_condition: Optional[Dict[str, Any]] = None,
+        environment: Optional[Dict[str, str]] = None,
+        network_config: Optional[Dict[str, Any]] = None,
+        model_bias_baseline_config: Optional[Dict[str, Any]] = None,
+        model_bias_app_specification: Optional[Dict[str, Any]] = None,
+        model_bias_job_input: Optional[Dict[str, Any]] = None,
+        model_bias_job_output_config: Optional[Dict[str, Any]] = None,
+    ):
+        self.job_definition_name = job_definition_name
+        self.job_definition_arn = FakeModelBiasJobDefinition.arn_formatter(
+            job_definition_name, account_id, region_name
+        )
+        self.tags = tags
+        self.role_arn = role_arn
+        self.job_resources = job_resources or {}
+        self.stopping_condition = stopping_condition or {}
+        self.environment = environment or {}
+        self.network_config = network_config or {}
+        self.model_bias_baseline_config = model_bias_baseline_config or {}
+        self.model_bias_app_specification = model_bias_app_specification or {}
+        self.model_bias_job_input = model_bias_job_input or {}
+        self.model_bias_job_output_config = model_bias_job_output_config or {}
+        self.creation_time = self.last_modified_time = datetime.now().strftime(
+            "%Y-%m-%d %H:%M:%S"
+        )
+
+    @property
+    def response_object(self) -> Dict[str, str]:
+        response_object = self.gen_response_object()
+        return {
+            k: v for k, v in response_object.items() if v is not None and v != [None]
+        }
+
+    @property
+    def response_create(self) -> Dict[str, str]:
+        return {"JobDefinitionArn": self.job_definition_arn}
+
+    @staticmethod
+    def arn_formatter(name: str, account_id: str, region: str) -> str:
+        return (
+            f"arn:aws:sagemaker:{region}:{account_id}:model-bias-job-definition/{name}"
+        )
+
+    @property
+    def summary_object(self) -> Dict[str, str]:
+        return {
+            "MonitoringJobDefinitionName": self.job_definition_name,
+            "MonitoringJobDefinitionArn": self.job_definition_arn,
+            "CreationTime": self.creation_time,
+            "EndpointName": self.model_bias_job_input.get("EndpointInput", {}).get(
+                "EndpointName", "EndpointName"
+            ),
+        }
+
+
 sagemaker_backends = BackendDict(SageMakerModelBackend, "sagemaker")
diff --git a/moto/sagemaker/responses.py b/moto/sagemaker/responses.py
index df44a2ce6..0226ce4ce 100644
--- a/moto/sagemaker/responses.py
+++ b/moto/sagemaker/responses.py
@@ -1024,6 +1024,52 @@ class SageMakerResponse(BaseResponse):
             dict(NextToken=next_token, ClusterNodeSummaries=cluster_node_summaries)
         )
 
+    def create_model_bias_job_definition(self) -> str:
+        account_id = self.current_account
+        job_definition_name = self._get_param("JobDefinitionName")
+        tags = self._get_param("Tags", [])
+        role_arn = self._get_param("RoleArn")
+        job_resources = self._get_param("JobResources")
+        stopping_condition = self._get_param("StoppingCondition")
+        environment = self._get_param("Environment", {})
+        network_config = self._get_param("NetworkConfig", {})
+        model_bias_baseline_config = self._get_param("ModelBiasBaselineConfig", {})
+        model_bias_app_specification = self._get_param("ModelBiasAppSpecification")
+        model_bias_job_input = self._get_param("ModelBiasJobInput")
+        model_bias_job_output_config = self._get_param("ModelBiasJobOutputConfig")
+
+        response = self.sagemaker_backend.create_model_bias_job_definition(
+            account_id=account_id,
+            job_definition_name=job_definition_name,
+            tags=tags,
+            role_arn=role_arn,
+            job_resources=job_resources,
+            stopping_condition=stopping_condition,
+            environment=environment,
+            network_config=network_config,
+            model_bias_baseline_config=model_bias_baseline_config,
+            model_bias_app_specification=model_bias_app_specification,
+            model_bias_job_input=model_bias_job_input,
+            model_bias_job_output_config=model_bias_job_output_config,
+        )
+        return json.dumps(response)
+
+    def list_model_bias_job_definitions(self) -> str:
+        result, next_token = self.sagemaker_backend.list_model_bias_job_definitions()
+        return json.dumps({"JobDefinitionSummaries": result, "NextToken": next_token})
+
+    def describe_model_bias_job_definition(self) -> str:
+        job_definition_name = self._get_param("JobDefinitionName")
+        job_definition = self.sagemaker_backend.describe_model_bias_job_definition(
+            job_definition_name
+        )
+        return json.dumps(job_definition)
+
+    def delete_model_bias_job_definition(self) -> str:
+        job_definition_name = self._get_param("JobDefinitionName")
+        self.sagemaker_backend.delete_model_bias_job_definition(job_definition_name)
+        return json.dumps({})
+
     def create_auto_ml_job_v2(self) -> str:
         auto_ml_job_name = self._get_param("AutoMLJobName")
         auto_ml_job_input_data_config = self._get_param("AutoMLJobInputDataConfig")


Test patch:
diff --git a/tests/__init__.py b/tests/__init__.py
index ab4f2a49f..8a1644152 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -1,6 +1,8 @@
 import logging
 import os
+from functools import wraps
 
+from moto import mock_aws
 from moto.core.config import default_user_config
 
 # Disable extra logging for tests
@@ -22,3 +24,27 @@ os.environ["MOTO_EC2_LOAD_DEFAULT_AMIS"] = "false"
 
 # Don't reset boto3 Session, as it's unnecessarily slow
 default_user_config["core"]["reset_boto3_session"] = False
+
+
+def allow_aws_request() -> bool:
+    return os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
+
+
+def aws_verified(func):
+    """
+    Function that is verified to work against AWS.
+    Can be run against AWS at any time by setting:
+      MOTO_TEST_ALLOW_AWS_REQUEST=true
+
+    If this environment variable is not set, the function runs in a `mock_aws` context.
+    """
+
+    @wraps(func)
+    def pagination_wrapper(**kwargs):
+        if allow_aws_request():
+            return func(**kwargs)
+        else:
+            with mock_aws():
+                return func(**kwargs)
+
+    return pagination_wrapper
diff --git a/tests/test_applicationautoscaling/test_applicationautoscaling.py b/tests/test_applicationautoscaling/test_applicationautoscaling.py
index a068dfa8d..6c30a0c93 100644
--- a/tests/test_applicationautoscaling/test_applicationautoscaling.py
+++ b/tests/test_applicationautoscaling/test_applicationautoscaling.py
@@ -5,6 +5,7 @@ import pytest
 
 from moto import mock_aws
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
+from tests.test_dynamodb import dynamodb_aws_verified
 
 DEFAULT_REGION = "us-east-1"
 DEFAULT_ECS_CLUSTER = "default"
@@ -50,6 +51,33 @@ def _create_ecs_defaults(ecs, create_service=True):
         )
 
 
+def delete_scheduled_actions():
+    service_namespaces = ["custom-resource", "dynamodb", "ecs"]
+    client = boto3.client("application-autoscaling", region_name=DEFAULT_REGION)
+
+    for namespace in service_namespaces:
+        actions = client.describe_scheduled_actions(ServiceNamespace=namespace)[
+            "ScheduledActions"
+        ]
+        for a in actions:
+            client.delete_scheduled_action(
+                ServiceNamespace=namespace,
+                ScheduledActionName=a["ScheduledActionName"],
+                ResourceId=a["ResourceId"],
+                ScalableDimension=a["ScalableDimension"],
+            )
+
+        targets = client.describe_scalable_targets(ServiceNamespace=namespace)[
+            "ScalableTargets"
+        ]
+        for t in targets:
+            client.deregister_scalable_target(
+                ServiceNamespace=namespace,
+                ResourceId=t["ResourceId"],
+                ScalableDimension=t["ScalableDimension"],
+            )
+
+
 @mock_aws
 def test_describe_scalable_targets_one_basic_ecs_success():
     ecs = boto3.client("ecs", region_name=DEFAULT_REGION)
@@ -535,30 +563,71 @@ def test_describe_scheduled_actions():
     assert len(resp["ScheduledActions"]) == 0
 
 
-@mock_aws
-def test_put_scheduled_action():
-    client = boto3.client("application-autoscaling", region_name="ap-southeast-1")
-    client.put_scheduled_action(
-        ServiceNamespace="ecs",
-        ScheduledActionName="action_name",
-        ResourceId="ecs:cluster:x",
-        ScalableDimension="ecs:service:DesiredCount",
-    )
+@dynamodb_aws_verified()
+@pytest.mark.aws_verified
+def test_put_scheduled_action(table_name=None):
+    client = boto3.client("application-autoscaling", region_name="us-east-1")
+    try:
+        resource_id = f"table/{table_name}"
+        client.register_scalable_target(
+            ServiceNamespace="dynamodb",
+            ResourceId=resource_id,
+            ScalableDimension="dynamodb:table:ReadCapacityUnits",
+            MinCapacity=1,
+            MaxCapacity=100,
+        )
+        client.put_scheduled_action(
+            ServiceNamespace="dynamodb",
+            Schedule="rate(2 minutes)",
+            ScheduledActionName="action_name",
+            ResourceId=resource_id,
+            ScalableDimension="dynamodb:table:ReadCapacityUnits",
+            ScalableTargetAction={"MinCapacity": 1, "MaxCapacity": 5},
+        )
 
-    resp = client.describe_scheduled_actions(ServiceNamespace="ecs")
-    assert len(resp["ScheduledActions"]) == 1
+        resp = client.describe_scheduled_actions(ServiceNamespace="dynamodb")
+        assert len(resp["ScheduledActions"]) == 1
 
-    action = resp["ScheduledActions"][0]
-    assert action["ScheduledActionName"] == "action_name"
-    assert (
-        action["ScheduledActionARN"]
-        == f"arn:aws:autoscaling:ap-southeast-1:{ACCOUNT_ID}:scheduledAction:ecs:scheduledActionName/action_name"
-    )
-    assert action["ServiceNamespace"] == "ecs"
-    assert action["ResourceId"] == "ecs:cluster:x"
-    assert action["ScalableDimension"] == "ecs:service:DesiredCount"
-    assert "CreationTime" in action
-    assert "ScalableTargetAction" not in action
+        action = resp["ScheduledActions"][0]
+        assert action["ScheduledActionName"] == "action_name"
+        assert action["ServiceNamespace"] == "dynamodb"
+        assert action["ResourceId"] == resource_id
+        assert action["ScalableDimension"] == "dynamodb:table:ReadCapacityUnits"
+        assert "CreationTime" in action
+
+        # Verify it's possible to create a second action against the same Dimension
+        client.put_scheduled_action(
+            ServiceNamespace="dynamodb",
+            Schedule="rate(1 hour)",
+            ScheduledActionName="action_name2",
+            ResourceId=resource_id,
+            ScalableDimension="dynamodb:table:ReadCapacityUnits",
+            ScalableTargetAction={"MinCapacity": 1, "MaxCapacity": 5},
+        )
+        actions = client.describe_scheduled_actions(ServiceNamespace="dynamodb")[
+            "ScheduledActions"
+        ]
+        assert len(actions) == 2
+
+        action1 = [a for a in actions if a["ScheduledActionName"] == "action_name"][0]
+        action2 = [a for a in actions if a["ScheduledActionName"] == "action_name2"][0]
+
+        assert action1["ResourceId"] == resource_id
+        assert action2["ResourceId"] == resource_id
+
+        # AWS includes a random ID
+        # arn:aws:autoscaling:us-east-1:{ACCOUNT_ID}:scheduledAction:2e767478-3c94-463b-8d0c-5a5fb3ee04cf:resource/dynamodb/table/{table_name}:scheduledActionName/action_name
+        assert action1["ScheduledActionARN"].endswith(
+            f"dynamodb/table/{table_name}:scheduledActionName/action_name"
+        )
+        assert action2["ScheduledActionARN"].endswith(
+            f"dynamodb/table/{table_name}:scheduledActionName/action_name2"
+        )
+
+        assert action1["Schedule"] == "rate(2 minutes)"
+        assert action2["Schedule"] == "rate(1 hour)"
+    finally:
+        delete_scheduled_actions()
 
 
 @mock_aws
@@ -572,7 +641,7 @@ def test_put_scheduled_action__use_update():
     )
     client.put_scheduled_action(
         ServiceNamespace="ecs",
-        ScheduledActionName="action_name_updated",
+        ScheduledActionName="action_name",
         ResourceId="ecs:cluster:x",
         ScalableDimension="ecs:service:DesiredCount",
         ScalableTargetAction={
@@ -585,10 +654,10 @@ def test_put_scheduled_action__use_update():
     assert len(resp["ScheduledActions"]) == 1
 
     action = resp["ScheduledActions"][0]
-    assert action["ScheduledActionName"] == "action_name_updated"
+    assert action["ScheduledActionName"] == "action_name"
     assert (
         action["ScheduledActionARN"]
-        == f"arn:aws:autoscaling:ap-southeast-1:{ACCOUNT_ID}:scheduledAction:ecs:scheduledActionName/action_name"
+        == f"arn:aws:autoscaling:ap-southeast-1:{ACCOUNT_ID}:scheduledAction:ecs/ecs:cluster:x:scheduledActionName/action_name"
     )
     assert action["ServiceNamespace"] == "ecs"
     assert action["ResourceId"] == "ecs:cluster:x"
diff --git a/tests/test_athena/__init__.py b/tests/test_athena/__init__.py
index 1822c621a..e69de29bb 100644
--- a/tests/test_athena/__init__.py
+++ b/tests/test_athena/__init__.py
@@ -1,59 +0,0 @@
-import os
-from functools import wraps
-from uuid import uuid4
-
-import boto3
-
-from moto import mock_aws
-from moto.s3.responses import DEFAULT_REGION_NAME
-from tests.test_s3 import empty_bucket
-
-
-def athena_aws_verified(func):
-    """
-    Function that is verified to work against AWS.
-    Can be run against AWS at any time by setting:
-      MOTO_TEST_ALLOW_AWS_REQUEST=true
-
-    If this environment variable is not set, the function runs in a `mock_aws` context.
-
-    This decorator will:
-      - Create a bucket
-      - Run the test and pass the bucket_name as an argument
-      - Delete the objects and the bucket itself
-    """
-
-    @wraps(func)
-    def pagination_wrapper():
-        bucket_name = str(uuid4())
-
-        allow_aws_request = (
-            os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
-        )
-
-        if allow_aws_request:
-            resp = create_bucket_and_test(bucket_name, is_aws=True)
-        else:
-            with mock_aws():
-                resp = create_bucket_and_test(bucket_name, is_aws=False)
-        return resp
-
-    def create_bucket_and_test(bucket_name: str, is_aws: bool):
-        client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
-
-        client.create_bucket(Bucket=bucket_name)
-        client.put_bucket_tagging(
-            Bucket=bucket_name,
-            Tagging={"TagSet": [{"Key": "environment", "Value": "moto_tests"}]},
-        )
-        try:
-            resp = func(bucket_name, is_aws)
-        finally:
-            ### CLEANUP ###
-
-            empty_bucket(client, bucket_name)
-            client.delete_bucket(Bucket=bucket_name)
-
-        return resp
-
-    return pagination_wrapper
diff --git a/tests/test_athena/test_athena_integration.py b/tests/test_athena/test_athena_integration.py
index 6af2ffca8..b075d1494 100644
--- a/tests/test_athena/test_athena_integration.py
+++ b/tests/test_athena/test_athena_integration.py
@@ -7,7 +7,8 @@ import pytest
 import requests
 
 from moto import settings
-from tests.test_athena import athena_aws_verified
+from tests import allow_aws_request
+from tests.test_s3 import s3_aws_verified
 
 DATA = ""
 DATA += json.dumps({"k1": "r1_v1", "k2": "r1_v2", "k3": 1}) + "\n"
@@ -72,9 +73,9 @@ COLUMN_INFO = [
 ]
 
 
-@athena_aws_verified
+@s3_aws_verified
 @pytest.mark.aws_verified
-def test_athena_csv_result(bucket_name=None, is_aws: bool = False):
+def test_athena_csv_result(bucket_name=None):
     if not settings.TEST_DECORATOR_MODE:
         raise SkipTest("No point in testing this outside of decorators")
 
@@ -102,7 +103,7 @@ OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat'
 LOCATION 's3://{bucket_name}/input/'; """
     get_query_results(CREATE_TABLE, athena, config, context)
 
-    if not is_aws:
+    if not allow_aws_request():
         athena_result = {
             "results": [
                 {
diff --git a/tests/test_cloudformation/__init__.py b/tests/test_cloudformation/__init__.py
index 2dc371035..e69de29bb 100644
--- a/tests/test_cloudformation/__init__.py
+++ b/tests/test_cloudformation/__init__.py
@@ -1,31 +0,0 @@
-import os
-from functools import wraps
-
-from moto import mock_aws
-
-
-def cloudformation_aws_verified():
-    """
-    Function that is verified to work against AWS.
-    Can be run against AWS at any time by setting:
-      MOTO_TEST_ALLOW_AWS_REQUEST=true
-
-    If this environment variable is not set, the function runs in a `mock_aws` context.
-    """
-
-    def inner(func):
-        @wraps(func)
-        def pagination_wrapper():
-            allow_aws_request = (
-                os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
-            )
-
-            if allow_aws_request:
-                return func()
-            else:
-                with mock_aws():
-                    return func()
-
-        return pagination_wrapper
-
-    return inner
diff --git a/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py b/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
index cc9fa776e..4320b1126 100644
--- a/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
+++ b/tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
@@ -15,8 +15,7 @@ from moto import mock_aws, settings
 from moto.cloudformation import cloudformation_backends
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
 from moto.utilities.distutils_version import LooseVersion
-from tests import EXAMPLE_AMI_ID
-from tests.test_cloudformation import cloudformation_aws_verified
+from tests import EXAMPLE_AMI_ID, aws_verified
 
 TEST_STACK_NAME = "test_stack"
 REGION_NAME = "us-east-1"
@@ -1373,7 +1372,7 @@ def test_update_stack_deleted_resources_can_reference_deleted_resources():
 
 
 @pytest.mark.aws_verified
-@cloudformation_aws_verified()
+@aws_verified
 def test_update_stack_with_previous_value():
     name = f"update-stack-with-previous-value-{str(uuid.uuid4())[0:6]}"
     bucket_name = str(uuid.uuid4())
diff --git a/tests/test_cloudformation/test_conditions.py b/tests/test_cloudformation/test_conditions.py
index 343d3403c..3135ea55b 100644
--- a/tests/test_cloudformation/test_conditions.py
+++ b/tests/test_cloudformation/test_conditions.py
@@ -4,7 +4,7 @@ import boto3
 import pytest
 from botocore.exceptions import ClientError
 
-from . import cloudformation_aws_verified
+from tests import aws_verified
 
 conditional_output_conditional_value = """
 Outputs:
@@ -38,7 +38,7 @@ Resources:
 
 
 @pytest.mark.aws_verified
-@cloudformation_aws_verified()
+@aws_verified
 def test_conditional_output_conditional_value():
     cf = boto3.client("cloudformation", region_name="us-east-1")
     param = [{"ParameterKey": "roleName", "ParameterValue": "myrole"}]
@@ -104,7 +104,7 @@ Resources:
 
 
 @pytest.mark.aws_verified
-@cloudformation_aws_verified()
+@aws_verified
 def test_permanent_output_conditional_value():
     cf = boto3.client("cloudformation", region_name="us-east-1")
     param = [{"ParameterKey": "roleName", "ParameterValue": "myrole"}]
@@ -174,7 +174,7 @@ Resources:
 
 
 @pytest.mark.aws_verified
-@cloudformation_aws_verified()
+@aws_verified
 def test_conditional_output_of_permanent_value():
     cf = boto3.client("cloudformation", region_name="us-east-1")
     param = [{"ParameterKey": "roleName", "ParameterValue": "myrole"}]
diff --git a/tests/test_kms/__init__.py b/tests/test_kms/__init__.py
index 3e557fa3b..e69de29bb 100644
--- a/tests/test_kms/__init__.py
+++ b/tests/test_kms/__init__.py
@@ -1,28 +0,0 @@
-import os
-from functools import wraps
-
-from moto import mock_aws
-
-
-def kms_aws_verified(func):
-    """
-    Function that is verified to work against AWS.
-    Can be run against AWS at any time by setting:
-      MOTO_TEST_ALLOW_AWS_REQUEST=true
-
-    If this environment variable is not set, the function runs in a `mock_kms` context.
-    """
-
-    @wraps(func)
-    def pagination_wrapper():
-        allow_aws_request = (
-            os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
-        )
-
-        if allow_aws_request:
-            return func()
-        else:
-            with mock_aws():
-                return func()
-
-    return pagination_wrapper
diff --git a/tests/test_kms/test_kms_encrypt.py b/tests/test_kms/test_kms_encrypt.py
index 5994f0943..3cfc8b8cc 100644
--- a/tests/test_kms/test_kms_encrypt.py
+++ b/tests/test_kms/test_kms_encrypt.py
@@ -5,13 +5,13 @@ import pytest
 from botocore.exceptions import ClientError
 
 from moto import mock_aws
+from tests import aws_verified
 
-from . import kms_aws_verified
 from .test_kms_boto3 import PLAINTEXT_VECTORS, _get_encoded_value
 
 
 @pytest.mark.aws_verified
-@kms_aws_verified
+@aws_verified
 def test_encrypt_key_with_empty_content():
     client_kms = boto3.client("kms", region_name="ap-northeast-1")
     metadata = client_kms.create_key()["KeyMetadata"]
@@ -27,7 +27,7 @@ def test_encrypt_key_with_empty_content():
 
 
 @pytest.mark.aws_verified
-@kms_aws_verified
+@aws_verified
 def test_encrypt_key_with_large_content():
     client_kms = boto3.client("kms", region_name="ap-northeast-1")
     metadata = client_kms.create_key()["KeyMetadata"]
diff --git a/tests/test_sagemaker/test_sagemaker_model_bias_job_definitions.py b/tests/test_sagemaker/test_sagemaker_model_bias_job_definitions.py
new file mode 100644
index 000000000..013c34adc
--- /dev/null
+++ b/tests/test_sagemaker/test_sagemaker_model_bias_job_definitions.py
@@ -0,0 +1,252 @@
+"""Unit tests for sagemaker-supported APIs."""
+
+import boto3
+
+from moto import mock_aws
+from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
+
+
+@mock_aws
+def test_create_model_bias_job():
+    client = boto3.client("sagemaker", region_name="us-west-2")
+
+    response = client.create_model_bias_job_definition(
+        JobDefinitionName="test-bias-job",
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/SageMakerRole",
+        ModelBiasAppSpecification={
+            "ImageUri": f"{ACCOUNT_ID}.dkr.ecr.us-west-2.amazonaws.com/my-image:latest",
+            "ConfigUri": "s3://my-bucket/bias-config.json",
+        },
+        ModelBiasJobInput={
+            "EndpointInput": {
+                "EndpointName": "test-endpoint",
+                "LocalPath": "/opt/ml/processing/input/endpoint",
+                "S3InputMode": "File",
+                "S3DataDistributionType": "FullyReplicated",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://my-bucket/ground-truth-data",
+            },
+        },
+        ModelBiasJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://my-bucket/bias-output",
+                        "LocalPath": "/opt/ml/processing/output",
+                        "S3UploadMode": "EndOfJob",
+                    },
+                },
+            ],
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m5.xlarge",
+                "VolumeSizeInGB": 10,
+            },
+        },
+        NetworkConfig={
+            "EnableInterContainerTrafficEncryption": False,
+            "EnableNetworkIsolation": False,
+        },
+    )
+
+    expected_job_definition_arn = f"arn:aws:sagemaker:us-west-2:{ACCOUNT_ID}:model-bias-job-definition/test-bias-job"
+    assert response["JobDefinitionArn"] == expected_job_definition_arn
+
+
+@mock_aws
+def test_describe_model_bias_job():
+    client = boto3.client("sagemaker", region_name="us-east-1")
+
+    client.create_model_bias_job_definition(
+        JobDefinitionName="test-bias-job",
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/SageMakerRole",
+        ModelBiasAppSpecification={
+            "ImageUri": f"{ACCOUNT_ID}.dkr.ecr.us-west-2.amazonaws.com/my-image:latest",
+            "ConfigUri": "s3://my-bucket/bias-config.json",
+        },
+        ModelBiasJobInput={
+            "EndpointInput": {
+                "EndpointName": "test-endpoint",
+                "LocalPath": "/opt/ml/processing/input/endpoint",
+                "S3InputMode": "File",
+                "S3DataDistributionType": "FullyReplicated",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://my-bucket/ground-truth-data",
+            },
+        },
+        ModelBiasJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://my-bucket/bias-output",
+                        "LocalPath": "/opt/ml/processing/output",
+                        "S3UploadMode": "EndOfJob",
+                    },
+                },
+            ],
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m5.xlarge",
+                "VolumeSizeInGB": 10,
+            },
+        },
+        NetworkConfig={
+            "EnableInterContainerTrafficEncryption": False,
+            "EnableNetworkIsolation": False,
+        },
+    )
+    response = client.describe_model_bias_job_definition(
+        JobDefinitionName="test-bias-job"
+    )
+    assert response["JobDefinitionName"] == "test-bias-job"
+
+
+@mock_aws
+def test_list_model_bias_jobs():
+    client = boto3.client("sagemaker", region_name="us-east-1")
+    client.create_model_bias_job_definition(
+        JobDefinitionName="test-bias-job-1",
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/SageMakerRole",
+        ModelBiasAppSpecification={
+            "ImageUri": f"{ACCOUNT_ID}.dkr.ecr.us-west-2.amazonaws.com/my-image:latest",
+            "ConfigUri": "s3://my-bucket/bias-config.json",
+        },
+        ModelBiasJobInput={
+            "EndpointInput": {
+                "EndpointName": "test-endpoint",
+                "LocalPath": "/opt/ml/processing/input/endpoint",
+                "S3InputMode": "File",
+                "S3DataDistributionType": "FullyReplicated",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://my-bucket/ground-truth-data",
+            },
+        },
+        ModelBiasJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://my-bucket/bias-output",
+                        "LocalPath": "/opt/ml/processing/output",
+                        "S3UploadMode": "EndOfJob",
+                    },
+                },
+            ],
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m5.xlarge",
+                "VolumeSizeInGB": 10,
+            },
+        },
+        NetworkConfig={
+            "EnableInterContainerTrafficEncryption": False,
+            "EnableNetworkIsolation": False,
+        },
+    )
+    client.create_model_bias_job_definition(
+        JobDefinitionName="test-bias-job-2",
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/SageMakerRole",
+        ModelBiasAppSpecification={
+            "ImageUri": f"{ACCOUNT_ID}.dkr.ecr.us-west-2.amazonaws.com/my-image:latest",
+            "ConfigUri": "s3://my-bucket/bias-config.json",
+        },
+        ModelBiasJobInput={
+            "EndpointInput": {
+                "EndpointName": "test-endpoint",
+                "LocalPath": "/opt/ml/processing/input/endpoint",
+                "S3InputMode": "File",
+                "S3DataDistributionType": "FullyReplicated",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://my-bucket/ground-truth-data",
+            },
+        },
+        ModelBiasJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://my-bucket/bias-output",
+                        "LocalPath": "/opt/ml/processing/output",
+                        "S3UploadMode": "EndOfJob",
+                    },
+                },
+            ],
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m5.xlarge",
+                "VolumeSizeInGB": 10,
+            },
+        },
+        NetworkConfig={
+            "EnableInterContainerTrafficEncryption": False,
+            "EnableNetworkIsolation": False,
+        },
+    )
+    response = client.list_model_bias_job_definitions()
+
+    job_names = [
+        job["MonitoringJobDefinitionName"] for job in response["JobDefinitionSummaries"]
+    ]
+    assert "test-bias-job-1" in job_names
+    assert "test-bias-job-2" in job_names
+
+
+@mock_aws
+def test_delete_model_bias_job():
+    client = boto3.client("sagemaker", region_name="us-east-1")
+
+    client.create_model_bias_job_definition(
+        JobDefinitionName="test-bias-job",
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/SageMakerRole",
+        ModelBiasAppSpecification={
+            "ImageUri": f"{ACCOUNT_ID}.dkr.ecr.us-west-2.amazonaws.com/my-image:latest",
+            "ConfigUri": "s3://my-bucket/bias-config.json",
+        },
+        ModelBiasJobInput={
+            "EndpointInput": {
+                "EndpointName": "test-endpoint",
+                "LocalPath": "/opt/ml/processing/input/endpoint",
+                "S3InputMode": "File",
+                "S3DataDistributionType": "FullyReplicated",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://my-bucket/ground-truth-data",
+            },
+        },
+        ModelBiasJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://my-bucket/bias-output",
+                        "LocalPath": "/opt/ml/processing/output",
+                        "S3UploadMode": "EndOfJob",
+                    },
+                },
+            ],
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m5.xlarge",
+                "VolumeSizeInGB": 10,
+            },
+        },
+        NetworkConfig={
+            "EnableInterContainerTrafficEncryption": False,
+            "EnableNetworkIsolation": False,
+        },
+    )
+    client.delete_model_bias_job_definition(JobDefinitionName="test-bias-job")
+    response = client.list_model_bias_job_definitions()
+    job_names = [job["JobDefinitionName"] for job in response["JobDefinitionSummaries"]]
+    assert "test-bias-job" not in job_names
diff --git a/tests/test_ses/__init__.py b/tests/test_ses/__init__.py
index 208c44bab..e69de29bb 100644
--- a/tests/test_ses/__init__.py
+++ b/tests/test_ses/__init__.py
@@ -1,29 +0,0 @@
-import os
-from functools import wraps
-
-from moto import mock_aws
-
-
-def ses_aws_verified(func):
-    """
-    Function that is verified to work against AWS.
-    Can be run against AWS at any time by setting:
-      MOTO_TEST_ALLOW_AWS_REQUEST=true
-
-    If this environment variable is not set, the function runs in a `mock_aws` context.
-    """
-
-    @wraps(func)
-    def pagination_wrapper():
-        allow_aws_request = (
-            os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
-        )
-
-        if allow_aws_request:
-            resp = func()
-        else:
-            with mock_aws():
-                resp = func()
-        return resp
-
-    return pagination_wrapper
diff --git a/tests/test_ses/test_ses_boto3.py b/tests/test_ses/test_ses_boto3.py
index ffe0ab706..9c05ae908 100644
--- a/tests/test_ses/test_ses_boto3.py
+++ b/tests/test_ses/test_ses_boto3.py
@@ -7,8 +7,7 @@ import pytest
 from botocore.exceptions import ClientError, ParamValidationError
 
 from moto import mock_aws
-
-from . import ses_aws_verified
+from tests import aws_verified
 
 
 @mock_aws
@@ -1314,7 +1313,7 @@ def test_render_template():
 
 
 @pytest.mark.aws_verified
-@ses_aws_verified
+@aws_verified
 def test_render_template__advanced():
     conn = boto3.client("ses", region_name="us-east-1")
 
diff --git a/tests/test_sqs/__init__.py b/tests/test_sqs/__init__.py
index 5d04e1d17..e69de29bb 100644
--- a/tests/test_sqs/__init__.py
+++ b/tests/test_sqs/__init__.py
@@ -1,28 +0,0 @@
-import os
-from functools import wraps
-
-from moto import mock_aws
-
-
-def sqs_aws_verified(func):
-    """
-    Function that is verified to work against AWS.
-    Can be run against AWS at any time by setting:
-      MOTO_TEST_ALLOW_AWS_REQUEST=true
-
-    If this environment variable is not set, the function runs in a `mock_aws` context.
-    """
-
-    @wraps(func)
-    def pagination_wrapper(*args, **kwargs):
-        allow_aws_request = (
-            os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
-        )
-
-        if allow_aws_request:
-            return func(*args, **kwargs)
-        else:
-            with mock_aws():
-                return func(*args, **kwargs)
-
-    return pagination_wrapper
diff --git a/tests/test_sqs/test_sqs.py b/tests/test_sqs/test_sqs.py
index c57286b23..b6294ff82 100644
--- a/tests/test_sqs/test_sqs.py
+++ b/tests/test_sqs/test_sqs.py
@@ -22,8 +22,7 @@ from moto.sqs.models import (
     Queue,
 )
 from moto.utilities.distutils_version import LooseVersion
-
-from . import sqs_aws_verified
+from tests import aws_verified
 
 TEST_POLICY = """
 {
@@ -3337,7 +3336,7 @@ def test_fifo_dedupe_error_no_message_dedupe_id_batch():
     )
 
 
-@sqs_aws_verified
+@aws_verified
 @pytest.mark.aws_verified
 @pytest.mark.parametrize(
     "queue_config", [{"FifoQueue": "true"}, {"FifoQueue": "true", "DelaySeconds": "10"}]
diff --git a/tests/test_timestreamwrite/__init__.py b/tests/test_timestreamwrite/__init__.py
index 211bde487..e69de29bb 100644
--- a/tests/test_timestreamwrite/__init__.py
+++ b/tests/test_timestreamwrite/__init__.py
@@ -1,29 +0,0 @@
-import os
-from functools import wraps
-
-from moto import mock_aws
-
-
-def timestreamwrite_aws_verified(func):
-    """
-    Function that is verified to work against AWS.
-    Can be run against AWS at any time by setting:
-      MOTO_TEST_ALLOW_AWS_REQUEST=true
-
-    If this environment variable is not set, the function runs in a `mock_aws` context.
-
-    """
-
-    @wraps(func)
-    def pagination_wrapper():
-        allow_aws_request = (
-            os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
-        )
-
-        if allow_aws_request:
-            return func()
-        else:
-            with mock_aws():
-                return func()
-
-    return pagination_wrapper
diff --git a/tests/test_timestreamwrite/test_timestreamwrite_database.py b/tests/test_timestreamwrite/test_timestreamwrite_database.py
index a600f6dca..b29b5eebe 100644
--- a/tests/test_timestreamwrite/test_timestreamwrite_database.py
+++ b/tests/test_timestreamwrite/test_timestreamwrite_database.py
@@ -6,12 +6,11 @@ from botocore.exceptions import ClientError
 
 from moto import mock_aws
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
-
-from . import timestreamwrite_aws_verified
+from tests import aws_verified
 
 
 @pytest.mark.aws_verified
-@timestreamwrite_aws_verified
+@aws_verified
 def test_create_database_simple():
     ts = boto3.client("timestream-write", region_name="us-east-1")
     db_name = "db_" + str(uuid4())[0:6]
@@ -71,7 +70,7 @@ def test_describe_database():
 
 
 @pytest.mark.aws_verified
-@timestreamwrite_aws_verified
+@aws_verified
 def test_describe_unknown_database():
     ts = boto3.client("timestream-write", region_name="us-east-1")
     with pytest.raises(ClientError) as exc:
diff --git a/tests/test_timestreamwrite/test_timestreamwrite_table.py b/tests/test_timestreamwrite/test_timestreamwrite_table.py
index c94beab9b..38ddd40b8 100644
--- a/tests/test_timestreamwrite/test_timestreamwrite_table.py
+++ b/tests/test_timestreamwrite/test_timestreamwrite_table.py
@@ -7,12 +7,11 @@ from botocore.exceptions import ClientError
 
 from moto import mock_aws, settings
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
-
-from . import timestreamwrite_aws_verified
+from tests import aws_verified
 
 
 @pytest.mark.aws_verified
-@timestreamwrite_aws_verified
+@aws_verified
 def test_create_table():
     ts = boto3.client("timestream-write", region_name="us-east-1")
     db_name = "db_" + str(uuid4())[0:6]
@@ -49,7 +48,7 @@ def test_create_table():
 
 
 @pytest.mark.aws_verified
-@timestreamwrite_aws_verified
+@aws_verified
 def test_create_table__with_magnetic_store_write_properties():
     ts = boto3.client("timestream-write", region_name="us-east-1")
     db_name = "db_" + str(uuid4())[0:6]
@@ -94,7 +93,7 @@ def test_create_table__with_magnetic_store_write_properties():
 
 
 @pytest.mark.aws_verified
-@timestreamwrite_aws_verified
+@aws_verified
 def test_create_table_without_retention_properties():
     ts = boto3.client("timestream-write", region_name="us-east-1")
     db_name = "db_" + str(uuid4())[0:6]
@@ -150,7 +149,7 @@ def test_describe_table():
 
 
 @pytest.mark.aws_verified
-@timestreamwrite_aws_verified
+@aws_verified
 def test_describe_unknown_database():
     ts = boto3.client("timestream-write", region_name="us-east-1")
     db_name = "db_" + str(uuid4())[0:6]
diff --git a/tests/test_timestreamwrite/test_timestreamwrite_tagging.py b/tests/test_timestreamwrite/test_timestreamwrite_tagging.py
index f4f9c510c..5b1d01b83 100644
--- a/tests/test_timestreamwrite/test_timestreamwrite_tagging.py
+++ b/tests/test_timestreamwrite/test_timestreamwrite_tagging.py
@@ -4,8 +4,7 @@ import boto3
 import pytest
 
 from moto import mock_aws
-
-from . import timestreamwrite_aws_verified
+from tests import aws_verified
 
 
 @mock_aws
@@ -68,7 +67,7 @@ def test_list_tagging_for_database_with_tags():
 
 
 @pytest.mark.aws_verified
-@timestreamwrite_aws_verified
+@aws_verified
 def test_tag_and_untag_database():
     ts = boto3.client("timestream-write", region_name="us-east-1")
     db_name = "db_" + str(uuid4())[0:6]