Repository: getmoto/moto
Pull request #7925 (2024-08-02 20:33:04)

Consolidate `Neptune` and `RDS` backends
Despite being documented as two distinct services (with two distinct boto3 clients), the Neptune and RDS services share a single AWS backend (RDS).  This is reflected in the botocore service definition for Neptune, which lists `rds` as its endpoint.  Calling the `DescribeDBClusters` API method with either the `neptune` or `rds` boto3 client results in the exact same list of clusters--no filtering is done on the AWS backend.  Neptune clusters are commingled with Aurora clusters.

@bblommers I think this makes more sense than trying to separate these services (and it prevents a lot of code duplication), but I want to get your opinion since you originally implemented/split out the Neptune service. (As I continue to work on improving the RDS backend overall, I do plan to handle some of the `if self.engine == "neptune` conditionals in a more elegant way.)

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/7925?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 94.39%. Comparing base [(`bbcfc4c`)](https://app.codecov.io/gh/getmoto/moto/commit/bbcfc4cba320812bb83acded86da7e36d39b76cc?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`dad257e`)](https://app.codecov.io/gh/getmoto/moto/commit/dad257e6fe87943186a2b30661bd7361048d867c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).
> Report is 8 commits behind head on master.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #7925      +/-   ##
==========================================
+ Coverage   94.38%   94.39%   +0.01%     
==========================================
  Files        1114     1109       -5     
  Lines       95224    95082     -142     
==========================================
- Hits        89875    89757     -118     
+ Misses       5349     5325      -24     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/7925/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/7925/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `29.13% <29.03%> (-0.03%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/7925/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `94.37% <100.00%> (+0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/7925?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

bblommers: @bpandola It is much cleaner like this, with RDS handling everything. If I remember correctly, I did try this approach first, but ran into some problems differentiating when a call came from Neptune and when from RDS. I don't remember exactly what went wrong, but based on this PR, it does work, so I'm more than happy to go ahead with this approach.

Patch:
diff --git a/moto/backends.py b/moto/backends.py
index 1f8fa4a50..b05d30a79 100644
--- a/moto/backends.py
+++ b/moto/backends.py
@@ -94,7 +94,6 @@ if TYPE_CHECKING:
     from moto.meteringmarketplace.models import MeteringMarketplaceBackend
     from moto.moto_api._internal.models import MotoAPIBackend
     from moto.mq.models import MQBackend
-    from moto.neptune.models import NeptuneBackend
     from moto.networkmanager.models import NetworkManagerBackend
     from moto.opensearch.models import OpenSearchServiceBackend
     from moto.opensearchserverless.models import OpenSearchServiceServerlessBackend
@@ -147,13 +146,18 @@ if TYPE_CHECKING:
     from moto.xray.models import XRayBackend
 
 
-ALT_SERVICE_NAMES = {"lambda": "awslambda", "moto_api": "moto_api._internal"}
+ALT_SERVICE_NAMES = {
+    "lambda": "awslambda",
+    "moto_api": "moto_api._internal",
+    "neptune": "rds",
+}
 ALT_BACKEND_NAMES = {
     "moto_api._internal": "moto_api",
     "awslambda": "lambda",
     "awslambda_simple": "lambda_simple",
     "dynamodb_v20111205": "dynamodb",
     "elasticbeanstalk": "eb",
+    "neptune": "rds",
 }
 
 
@@ -556,7 +560,7 @@ def get_backend(name: "Literal['moto_api']") -> "BackendDict[MotoAPIBackend]": .
 @overload
 def get_backend(name: "Literal['mq']") -> "BackendDict[MQBackend]": ...
 @overload
-def get_backend(name: "Literal['neptune']") -> "BackendDict[NeptuneBackend]": ...
+def get_backend(name: "Literal['neptune']") -> "BackendDict[RDSBackend]": ...
 @overload
 def get_backend(
     name: "Literal['networkmanager']",
diff --git a/moto/neptune/README.md b/moto/neptune/README.md
new file mode 100644
index 000000000..abedabe61
--- /dev/null
+++ b/moto/neptune/README.md
@@ -0,0 +1,3 @@
+Neptune service is handled by the RDS service backend located [here](../rds).
+
+Neptune service has a dedicated test suite located [here](../../tests/test_neptune).
\ No newline at end of file
diff --git a/moto/neptune/__init__.py b/moto/neptune/__init__.py
deleted file mode 100644
index e7353bb83..000000000
--- a/moto/neptune/__init__.py
+++ /dev/null
@@ -1,9 +0,0 @@
-"""
-Neptune is a bit of an odd duck.
-It shares almost everything with RDS: the endpoint URL, and the features. Only the parameters to these features can be different.
-
-Because the endpoint URL is the same (rds.amazonaws.com), every request is intercepted by the RDS service.
-RDS then has to determine whether any incoming call was meant for RDS, or for neptune.
-"""
-
-from .models import neptune_backends  # noqa: F401
diff --git a/moto/neptune/exceptions.py b/moto/neptune/exceptions.py
deleted file mode 100644
index c2958b0e8..000000000
--- a/moto/neptune/exceptions.py
+++ /dev/null
@@ -1,27 +0,0 @@
-from jinja2 import Template
-
-from moto.core.exceptions import RESTError
-
-
-class NeptuneClientError(RESTError):
-    def __init__(self, code: str, message: str):
-        super().__init__(error_type=code, message=message)
-        template = Template(
-            """
-        <ErrorResponse>
-            <Error>
-              <Code>{{ code }}</Code>
-              <Message>{{ message }}</Message>
-              <Type>Sender</Type>
-            </Error>
-            <RequestId>6876f774-7273-11e4-85dc-39e55ca848d1</RequestId>
-        </ErrorResponse>"""
-        )
-        self.description = template.render(code=code, message=message)
-
-
-class DBClusterNotFoundError(NeptuneClientError):
-    def __init__(self, cluster_identifier: str):
-        super().__init__(
-            "DBClusterNotFoundFault", f"DBCluster {cluster_identifier} not found."
-        )
diff --git a/moto/neptune/models.py b/moto/neptune/models.py
deleted file mode 100644
index ee11a5686..000000000
--- a/moto/neptune/models.py
+++ /dev/null
@@ -1,377 +0,0 @@
-import copy
-import string
-from typing import Any, Dict, List, Optional
-
-from jinja2 import Template
-
-from moto.core.base_backend import BackendDict, BaseBackend
-from moto.core.common_models import BaseModel
-from moto.core.utils import iso_8601_datetime_with_milliseconds
-from moto.moto_api._internal import mock_random as random
-from moto.utilities.utils import get_partition, load_resource
-
-from .exceptions import DBClusterNotFoundError
-
-
-class GlobalCluster(BaseModel):
-    def __init__(
-        self,
-        account_id: str,
-        region_name: str,
-        global_cluster_identifier: str,
-        engine: Optional[str],
-        engine_version: Optional[str],
-        storage_encrypted: Optional[str],
-        deletion_protection: Optional[str],
-    ):
-        self.global_cluster_identifier = global_cluster_identifier
-        self.global_cluster_resource_id = "cluster-" + random.get_random_hex(8)
-        self.global_cluster_arn = f"arn:{get_partition(region_name)}:rds::{account_id}:global-cluster:{global_cluster_identifier}"
-        self.engine = engine or "neptune"
-        self.engine_version = engine_version or "1.2.0.0"
-        self.storage_encrypted = (
-            storage_encrypted and storage_encrypted.lower() == "true"
-        )
-        self.deletion_protection = (
-            deletion_protection and deletion_protection.lower() == "true"
-        )
-
-    def to_xml(self) -> str:
-        template = Template(
-            """
-          <GlobalClusterIdentifier>{{ cluster.global_cluster_identifier }}</GlobalClusterIdentifier>
-          <GlobalClusterResourceId>{{ cluster.global_cluster_resource_id }}</GlobalClusterResourceId>
-          <GlobalClusterArn>{{ cluster.global_cluster_arn }}</GlobalClusterArn>
-          <Engine>{{ cluster.engine }}</Engine>
-          <Status>available</Status>
-          <EngineVersion>{{ cluster.engine_version }}</EngineVersion>
-          <StorageEncrypted>{{ 'true' if cluster.storage_encrypted else 'false' }}</StorageEncrypted>
-          <DeletionProtection>{{ 'true' if cluster.deletion_protection else 'false' }}</DeletionProtection>"""
-        )
-        return template.render(cluster=self)
-
-
-class DBCluster(BaseModel):
-    def __init__(
-        self,
-        account_id: str,
-        region_name: str,
-        db_cluster_identifier: str,
-        database_name: Optional[str],
-        tags: List[Dict[str, str]],
-        storage_encrypted: str,
-        parameter_group_name: str,
-        engine: str,
-        engine_version: str,
-        kms_key_id: Optional[str],
-        preferred_maintenance_window: Optional[str],
-        preferred_backup_window: Optional[str],
-        backup_retention_period: Optional[int],
-        port: Optional[int],
-        serverless_v2_scaling_configuration: Optional[Dict[str, int]],
-    ):
-        self.account_id = account_id
-        self.region_name = region_name
-        self.db_cluster_identifier = db_cluster_identifier
-        self.resource_id = "cluster-" + random.get_random_hex(8)
-        self.tags = tags
-        self.storage_encrypted = storage_encrypted.lower() != "false"
-        self.db_cluster_parameter_group_name = parameter_group_name
-        self.engine = engine
-        self.engine_version = engine_version
-        self.database_name = database_name
-        self.db_subnet_group = "default"
-        self.status = "available"
-        self.backup_retention_period = backup_retention_period
-        self.cluster_create_time = iso_8601_datetime_with_milliseconds()
-        self.url_identifier = "".join(
-            random.choice(string.ascii_lowercase + string.digits) for _ in range(12)
-        )
-        self.endpoint = f"{self.db_cluster_identifier}.cluster-{self.url_identifier}.{self.region_name}.neptune.amazonaws.com"
-        self.reader_endpoint = f"{self.db_cluster_identifier}.cluster-ro-{self.url_identifier}.{self.region_name}.neptune.amazonaws.com"
-        self.resource_id = "cluster-" + "".join(
-            random.choice(string.ascii_uppercase + string.digits) for _ in range(26)
-        )
-        self.hosted_zone_id = "".join(
-            random.choice(string.ascii_uppercase + string.digits) for _ in range(14)
-        )
-        self.kms_key_id = kms_key_id or (
-            "default_kms_key_id" if self.storage_encrypted else None
-        )
-        self.preferred_maintenance_window = preferred_maintenance_window
-        self.preferred_backup_window = preferred_backup_window
-        self.port = port
-        self.availability_zones = [
-            f"{self.region_name}a",
-            f"{self.region_name}b",
-            f"{self.region_name}c",
-        ]
-        self.serverless_v2_scaling_configuration = serverless_v2_scaling_configuration
-
-    @property
-    def db_cluster_arn(self) -> str:
-        return f"arn:{get_partition(self.region_name)}:rds:{self.region_name}:{self.account_id}:cluster:{self.db_cluster_identifier}"
-
-    def get_tags(self) -> List[Dict[str, str]]:
-        return self.tags
-
-    def add_tags(self, tags: List[Dict[str, str]]) -> List[Dict[str, str]]:
-        new_keys = [tag_set["Key"] for tag_set in tags]
-        self.tags = [tag_set for tag_set in self.tags if tag_set["Key"] not in new_keys]
-        self.tags.extend(tags)
-        return self.tags
-
-    def remove_tags(self, tag_keys: List[str]) -> None:
-        self.tags = [tag_set for tag_set in self.tags if tag_set["Key"] not in tag_keys]
-
-    def to_xml(self) -> str:
-        template = Template(
-            """<DBCluster>
-            {% if cluster.allocated_storage %}
-        <AllocatedStorage>{{ cluster.allocated_storage }}</AllocatedStorage>
-        {% endif %}
-        <AvailabilityZones>
-          {% for zone in cluster.availability_zones %}
-          <AvailabilityZone>{{ zone }}</AvailabilityZone>
-          {% endfor %}
-        </AvailabilityZones>
-        {% if cluster.backup_retention_period %}
-        <BackupRetentionPeriod>{{ cluster.backup_retention_period }}</BackupRetentionPeriod>
-        {% endif %}
-        {% if cluster.character_set_name %}
-        <CharacterSetName>{{ cluster.character_set_name }}</CharacterSetName>
-        {% endif %}
-        {% if cluster.database_name %}
-        <DatabaseName>{{ cluster.database_name }}</DatabaseName>
-        {% endif %}
-        <DBClusterIdentifier>{{ cluster.db_cluster_identifier }}</DBClusterIdentifier>
-        <DBClusterParameterGroup>{{ cluster.db_cluster_parameter_group_name }}</DBClusterParameterGroup>
-        <DBSubnetGroup>{{ cluster.db_subnet_group }}</DBSubnetGroup>
-        <Status>{{ cluster.status }}</Status>
-        <PercentProgress>{{ cluster.percent_progress }}</PercentProgress>
-        {% if cluster.earliest_restorable_time %}
-        <EarliestRestorableTime>{{ cluster.earliest_restorable_time }}</EarliestRestorableTime>
-        {% endif %}
-        <Endpoint>{{ cluster.endpoint }}</Endpoint>
-        <ReaderEndpoint>{{ cluster.reader_endpoint }}</ReaderEndpoint>
-        <MultiAZ>false</MultiAZ>
-        <Engine>{{ cluster.engine }}</Engine>
-        <EngineVersion>{{ cluster.engine_version }}</EngineVersion>
-        {% if cluster.latest_restorable_time %}
-        <LatestRestorableTime>{{ cluster.latest_restorable_time }}</LatestRestorableTime>
-        {% endif %}
-        {% if cluster.port %}
-        <Port>{{ cluster.port }}</Port>
-        {% endif %}
-        <MasterUsername>{{ cluster.master_username }}</MasterUsername>
-        <DBClusterOptionGroupMemberships>
-{% for dbclusteroptiongroupmembership in cluster.dbclusteroptiongroupmemberships %}
-          <member>
-            <DBClusterOptionGroupName>{{ dbclusteroptiongroupmembership.db_cluster_option_group_name }}</DBClusterOptionGroupName>
-            <Status>{{ dbclusteroptiongroupmembership.status }}</Status>
-          </member>
-{% endfor %}
-        </DBClusterOptionGroupMemberships>
-        <PreferredBackupWindow>{{ cluster.preferred_backup_window }}</PreferredBackupWindow>
-        <PreferredMaintenanceWindow>{{ cluster.preferred_maintenance_window }}</PreferredMaintenanceWindow>
-        <ReplicationSourceIdentifier>{{ cluster.replication_source_identifier }}</ReplicationSourceIdentifier>
-        <ReadReplicaIdentifiers>
-{% for readreplicaidentifier in cluster.readreplicaidentifiers %}
-          <member/>
-{% endfor %}
-        </ReadReplicaIdentifiers>
-        <DBClusterMembers>
-{% for dbclustermember in cluster.dbclustermembers %}
-          <member>
-            <DBInstanceIdentifier>{{ dbclustermember.db_instance_identifier }}</DBInstanceIdentifier>
-            <IsClusterWriter>{{ dbclustermember.is_cluster_writer }}</IsClusterWriter>
-            <DBClusterParameterGroupStatus>{{ dbclustermember.db_cluster_parameter_group_status }}</DBClusterParameterGroupStatus>
-            <PromotionTier>{{ dbclustermember.promotion_tier }}</PromotionTier>
-          </member>
-{% endfor %}
-        </DBClusterMembers>
-        <VpcSecurityGroups>
-{% for vpcsecuritygroup in cluster.vpcsecuritygroups %}
-          <member>
-            <VpcSecurityGroupId>{{ vpcsecuritygroup.vpc_security_group_id }}</VpcSecurityGroupId>
-            <Status>{{ vpcsecuritygroup.status }}</Status>
-          </member>
-{% endfor %}
-        </VpcSecurityGroups>
-        <HostedZoneId>{{ cluster.hosted_zone_id }}</HostedZoneId>
-        <StorageEncrypted>{{ 'true' if cluster.storage_encrypted else 'false'}}</StorageEncrypted>
-        <KmsKeyId>{{ cluster.kms_key_id }}</KmsKeyId>
-        <DbClusterResourceId>{{ cluster.resource_id }}</DbClusterResourceId>
-        <DBClusterArn>{{ cluster.db_cluster_arn }}</DBClusterArn>
-        <AssociatedRoles>
-{% for associatedrole in cluster.associatedroles %}
-          <member>
-            <RoleArn>{{ associatedrole.role_arn }}</RoleArn>
-            <Status>{{ associatedrole.status }}</Status>
-            <FeatureName>{{ associatedrole.feature_name }}</FeatureName>
-          </member>
-{% endfor %}
-        </AssociatedRoles>
-        <IAMDatabaseAuthenticationEnabled>false</IAMDatabaseAuthenticationEnabled>
-        <CloneGroupId>{{ cluster.clone_group_id }}</CloneGroupId>
-        <ClusterCreateTime>{{ cluster.cluster_create_time }}</ClusterCreateTime>
-        <CopyTagsToSnapshot>false</CopyTagsToSnapshot>
-        <EnabledCloudwatchLogsExports>
-{% for enabledcloudwatchlogsexport in cluster.enabledcloudwatchlogsexports %}
-          <member/>db_cluster_arn
-{% endfor %}
-        </EnabledCloudwatchLogsExports>
-        <DeletionProtection>false</DeletionProtection>
-        <CrossAccountClone>false</CrossAccountClone>
-        {% if cluster.automatic_restart_time %}
-        <AutomaticRestartTime>{{ cluster.automatic_restart_time }}</AutomaticRestartTime>
-        {% endif %}
-        {% if cluster.serverless_v2_scaling_configuration %}
-        <ServerlessV2ScalingConfiguration>
-            <MinCapacity>{{ cluster.serverless_v2_scaling_configuration["MinCapacity"] }}</MinCapacity>
-            <MaxCapacity>{{ cluster.serverless_v2_scaling_configuration["MaxCapacity"] }}</MaxCapacity>
-        </ServerlessV2ScalingConfiguration>
-        {% endif %}
-        </DBCluster>"""
-        )
-        return template.render(cluster=self)
-
-
-class NeptuneBackend(BaseBackend):
-    """Implementation of Neptune APIs."""
-
-    def __init__(self, region_name: str, account_id: str):
-        super().__init__(region_name, account_id)
-        self.clusters: Dict[str, DBCluster] = dict()
-        self.global_clusters: Dict[str, GlobalCluster] = dict()
-        self._db_cluster_options: Optional[List[Dict[str, Any]]] = None
-
-    @property
-    def global_backend(self) -> "NeptuneBackend":
-        return neptune_backends[self.account_id]["us-east-1"]
-
-    @property
-    def db_cluster_options(self) -> List[Dict[str, Any]]:  # type: ignore[misc]
-        if self._db_cluster_options is None:
-            from moto.rds.utils import decode_orderable_db_instance
-
-            decoded_options: List[Dict[str, Any]] = load_resource(
-                __name__, "../rds/resources/cluster_options/neptune.json"
-            )
-            self._db_cluster_options = [
-                decode_orderable_db_instance(option) for option in decoded_options
-            ]
-        return self._db_cluster_options
-
-    def create_db_cluster(self, **kwargs: Any) -> DBCluster:
-        cluster = DBCluster(
-            account_id=self.account_id,
-            region_name=self.region_name,
-            db_cluster_identifier=kwargs["db_cluster_identifier"],
-            database_name=kwargs.get("database_name"),
-            storage_encrypted=kwargs.get("storage_encrypted", True),
-            parameter_group_name=kwargs.get("db_cluster_parameter_group_name") or "",
-            tags=kwargs.get("tags", []),
-            engine=kwargs.get("engine", "neptune"),
-            engine_version=kwargs.get("engine_version") or "1.2.0.2",
-            kms_key_id=kwargs.get("kms_key_id"),
-            preferred_maintenance_window=kwargs.get("preferred_maintenance_window")
-            or "none",
-            preferred_backup_window=kwargs.get("preferred_backup_window"),
-            backup_retention_period=kwargs.get("backup_retention_period") or 1,
-            port=kwargs.get("port") or 8192,
-            serverless_v2_scaling_configuration=kwargs.get(
-                "serverless_v2_scaling_configuration"
-            ),
-        )
-        self.clusters[cluster.db_cluster_identifier] = cluster
-        return cluster
-
-    def create_global_cluster(
-        self,
-        global_cluster_identifier: str,
-        engine: Optional[str],
-        engine_version: Optional[str],
-        storage_encrypted: Optional[str],
-        deletion_protection: Optional[str],
-    ) -> GlobalCluster:
-        cluster = GlobalCluster(
-            account_id=self.account_id,
-            region_name=self.region_name,
-            global_cluster_identifier=global_cluster_identifier,
-            engine=engine,
-            engine_version=engine_version,
-            storage_encrypted=storage_encrypted,
-            deletion_protection=deletion_protection,
-        )
-        self.global_backend.global_clusters[global_cluster_identifier] = cluster
-        return cluster
-
-    def delete_global_cluster(self, global_cluster_identifier: str) -> GlobalCluster:
-        return self.global_backend.global_clusters.pop(global_cluster_identifier)
-
-    def describe_global_clusters(self) -> List[GlobalCluster]:
-        return list(self.global_backend.global_clusters.values())
-
-    def describe_db_clusters(self, db_cluster_identifier: str) -> List[DBCluster]:
-        """
-        Pagination and the Filters-argument is not yet implemented
-        """
-        if db_cluster_identifier:
-            if db_cluster_identifier not in self.clusters:
-                raise DBClusterNotFoundError(db_cluster_identifier)
-            return [self.clusters[db_cluster_identifier]]
-        return list(self.clusters.values())
-
-    def delete_db_cluster(self, cluster_identifier: str) -> DBCluster:
-        """
-        The parameters SkipFinalSnapshot and FinalDBSnapshotIdentifier are not yet implemented.
-        The DeletionProtection-attribute is not yet enforced
-        """
-        if cluster_identifier in self.clusters:
-            return self.clusters.pop(cluster_identifier)
-        raise DBClusterNotFoundError(cluster_identifier)
-
-    def modify_db_cluster(self, kwargs: Any) -> DBCluster:
-        cluster_id = kwargs["db_cluster_identifier"]
-
-        cluster = self.clusters[cluster_id]
-        del self.clusters[cluster_id]
-
-        kwargs["db_cluster_identifier"] = kwargs.pop("new_db_cluster_identifier")
-        for k, v in kwargs.items():
-            if v is not None:
-                setattr(cluster, k, v)
-
-        cluster_id = kwargs.get("new_db_cluster_identifier", cluster_id)
-        self.clusters[cluster_id] = cluster
-
-        initial_state = copy.deepcopy(cluster)  # Return status=creating
-        cluster.status = "available"  # Already set the final status in the background
-        return initial_state
-
-    def start_db_cluster(self, cluster_identifier: str) -> DBCluster:
-        if cluster_identifier not in self.clusters:
-            raise DBClusterNotFoundError(cluster_identifier)
-        cluster = self.clusters[cluster_identifier]
-        temp_state = copy.deepcopy(cluster)
-        temp_state.status = "started"
-        cluster.status = "available"  # This is the final status - already setting it in the background
-        return temp_state
-
-    def describe_orderable_db_instance_options(
-        self, engine_version: Optional[str]
-    ) -> List[Dict[str, Any]]:
-        """
-        Only the EngineVersion-parameter is currently implemented.
-        """
-        if engine_version:
-            return [
-                option
-                for option in self.db_cluster_options
-                if option["EngineVersion"] == engine_version
-            ]
-        return self.db_cluster_options
-
-
-neptune_backends = BackendDict(NeptuneBackend, "neptune")
diff --git a/moto/neptune/responses.py b/moto/neptune/responses.py
deleted file mode 100644
index d9b57532a..000000000
--- a/moto/neptune/responses.py
+++ /dev/null
@@ -1,194 +0,0 @@
-from moto.core.responses import BaseResponse
-
-from .models import NeptuneBackend, neptune_backends
-
-
-class NeptuneResponse(BaseResponse):
-    """Handler for Neptune requests and responses."""
-
-    def __init__(self) -> None:
-        super().__init__(service_name="neptune")
-
-    @property
-    def neptune_backend(self) -> NeptuneBackend:
-        """Return backend instance specific for this region."""
-        return neptune_backends[self.current_account][self.region]
-
-    @property
-    def global_backend(self) -> NeptuneBackend:
-        """Return backend instance of the region that stores Global Clusters"""
-        return neptune_backends[self.current_account]["us-east-1"]
-
-    def create_db_cluster(self) -> str:
-        params = self._get_params()
-        availability_zones = params.get("AvailabilityZones")
-        backup_retention_period = params.get("BackupRetentionPeriod")
-        character_set_name = params.get("CharacterSetName")
-        copy_tags_to_snapshot = params.get("CopyTagsToSnapshot")
-        database_name = params.get("DatabaseName")
-        db_cluster_identifier = params.get("DBClusterIdentifier")
-        db_cluster_parameter_group_name = params.get("DBClusterParameterGroupName")
-        vpc_security_group_ids = params.get("VpcSecurityGroupIds")
-        db_subnet_group_name = params.get("DBSubnetGroupName")
-        engine = params.get("Engine")
-        engine_version = params.get("EngineVersion")
-        port = params.get("Port")
-        master_username = params.get("MasterUsername")
-        master_user_password = params.get("MasterUserPassword")
-        option_group_name = params.get("OptionGroupName")
-        preferred_backup_window = params.get("PreferredBackupWindow")
-        preferred_maintenance_window = params.get("PreferredMaintenanceWindow")
-        replication_source_identifier = params.get("ReplicationSourceIdentifier")
-        tags = (self._get_multi_param_dict("Tags") or {}).get("Tag", [])
-        storage_encrypted = params.get("StorageEncrypted", "")
-        kms_key_id = params.get("KmsKeyId")
-        pre_signed_url = params.get("PreSignedUrl")
-        enable_iam_database_authentication = params.get(
-            "EnableIAMDatabaseAuthentication"
-        )
-        enable_cloudwatch_logs_exports = params.get("EnableCloudwatchLogsExports")
-        deletion_protection = params.get("DeletionProtection")
-        serverless_v2_scaling_configuration = params.get(
-            "ServerlessV2ScalingConfiguration"
-        )
-        global_cluster_identifier = params.get("GlobalClusterIdentifier")
-        source_region = params.get("SourceRegion")
-        db_cluster = self.neptune_backend.create_db_cluster(
-            availability_zones=availability_zones,
-            backup_retention_period=backup_retention_period,
-            character_set_name=character_set_name,
-            copy_tags_to_snapshot=copy_tags_to_snapshot,
-            database_name=database_name,
-            db_cluster_identifier=db_cluster_identifier,
-            db_cluster_parameter_group_name=db_cluster_parameter_group_name,
-            vpc_security_group_ids=vpc_security_group_ids,
-            db_subnet_group_name=db_subnet_group_name,
-            engine=engine,
-            engine_version=engine_version,
-            port=port,
-            master_username=master_username,
-            master_user_password=master_user_password,
-            option_group_name=option_group_name,
-            preferred_backup_window=preferred_backup_window,
-            preferred_maintenance_window=preferred_maintenance_window,
-            replication_source_identifier=replication_source_identifier,
-            tags=tags,
-            storage_encrypted=storage_encrypted,
-            kms_key_id=kms_key_id,
-            pre_signed_url=pre_signed_url,
-            enable_iam_database_authentication=enable_iam_database_authentication,
-            enable_cloudwatch_logs_exports=enable_cloudwatch_logs_exports,
-            deletion_protection=deletion_protection,
-            serverless_v2_scaling_configuration=serverless_v2_scaling_configuration,
-            global_cluster_identifier=global_cluster_identifier,
-            source_region=source_region,
-        )
-        template = self.response_template(CREATE_DB_CLUSTER_TEMPLATE)
-        return template.render(cluster=db_cluster)
-
-    def describe_db_clusters(self) -> str:
-        params = self._get_params()
-        db_cluster_identifier = params["DBClusterIdentifier"]
-        db_clusters = self.neptune_backend.describe_db_clusters(
-            db_cluster_identifier=db_cluster_identifier
-        )
-        template = self.response_template(DESCRIBE_DB_CLUSTERS_TEMPLATE)
-        return template.render(db_clusters=db_clusters)
-
-    def describe_global_clusters(self) -> str:
-        clusters = self.global_backend.describe_global_clusters()
-        template = self.response_template(DESCRIBE_GLOBAL_CLUSTERS_TEMPLATE)
-        return template.render(clusters=clusters)
-
-    def create_global_cluster(self) -> str:
-        params = self._get_params()
-        cluster = self.global_backend.create_global_cluster(
-            global_cluster_identifier=params["GlobalClusterIdentifier"],
-            engine=params.get("Engine"),
-            engine_version=params.get("EngineVersion"),
-            storage_encrypted=params.get("StorageEncrypted"),
-            deletion_protection=params.get("DeletionProtection"),
-        )
-        template = self.response_template(CREATE_GLOBAL_CLUSTER_TEMPLATE)
-        return template.render(cluster=cluster)
-
-    def delete_global_cluster(self) -> str:
-        params = self._get_params()
-        cluster = self.global_backend.delete_global_cluster(
-            global_cluster_identifier=params["GlobalClusterIdentifier"],
-        )
-        template = self.response_template(DELETE_GLOBAL_CLUSTER_TEMPLATE)
-        return template.render(cluster=cluster)
-
-
-CREATE_DB_CLUSTER_TEMPLATE = """<CreateDBClusterResponse xmlns="http://rds.amazonaws.com/doc/2014-10-31/">
-  <ResponseMetadata>
-    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
-  </ResponseMetadata>
-  <CreateDBClusterResult>
-        {{ cluster.to_xml() }}
-  </CreateDBClusterResult>
-</CreateDBClusterResponse>"""
-
-DESCRIBE_DB_CLUSTERS_TEMPLATE = """<DescribeDBClustersResponse xmlns="http://rds.amazonaws.com/doc/2014-10-31/">
-  <ResponseMetadata>
-    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
-  </ResponseMetadata>
-  <DescribeDBClustersResult>
-    <DBClusters>
-{% for cluster in db_clusters %}
-        {{ cluster.to_xml() }}
-{% endfor %}
-    </DBClusters>
-  </DescribeDBClustersResult>
-</DescribeDBClustersResponse>"""
-
-CREATE_GLOBAL_CLUSTER_TEMPLATE = """<CreateGlobalClusterResponse xmlns="http://rds.amazonaws.com/doc/2014-10-31/">
-  <ResponseMetadata>
-    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
-  </ResponseMetadata>
-  <CreateGlobalClusterResult>
-  <GlobalCluster>
-    {{ cluster.to_xml() }}
-  </GlobalCluster>
-  </CreateGlobalClusterResult>
-</CreateGlobalClusterResponse>"""
-
-DELETE_GLOBAL_CLUSTER_TEMPLATE = """<DeleteGlobalClusterResponse xmlns="http://rds.amazonaws.com/doc/2014-10-31/">
-  <ResponseMetadata>
-    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
-  </ResponseMetadata>
-  <DeleteGlobalClusterResult>
-  <GlobalCluster>
-    {{ cluster.to_xml() }}
-  </GlobalCluster>
-  </DeleteGlobalClusterResult>
-</DeleteGlobalClusterResponse>"""
-
-DESCRIBE_GLOBAL_CLUSTERS_TEMPLATE = """<DescribeGlobalClustersResponse xmlns="http://rds.amazonaws.com/doc/2014-10-31/">
-  <ResponseMetadata>
-    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
-  </ResponseMetadata>
-  <DescribeGlobalClustersResult>
-    <GlobalClusters>
-{% for cluster in clusters %}
-    <GlobalClusterMember>
-        {{ cluster.to_xml() }}
-        </GlobalClusterMember>
-{% endfor %}
-    </GlobalClusters>
-  </DescribeGlobalClustersResult>
-</DescribeGlobalClustersResponse>"""
-
-REMOVE_FROM_GLOBAL_CLUSTER_TEMPLATE = """<RemoveFromGlobalClusterResponse xmlns="http://rds.amazonaws.com/doc/2014-10-31/">
-  <ResponseMetadata>
-    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
-  </ResponseMetadata>
-  <RemoveFromGlobalClusterResult>
-  {% if cluster %}
-  <GlobalCluster>
-    {{ cluster.to_xml() }}
-  </GlobalCluster>
-  {% endif %}
-  </RemoveFromGlobalClusterResult>
-</RemoveFromGlobalClusterResponse>"""
diff --git a/moto/neptune/urls.py b/moto/neptune/urls.py
deleted file mode 100644
index bcdbcb294..000000000
--- a/moto/neptune/urls.py
+++ /dev/null
@@ -1,8 +0,0 @@
-"""
-All calls to this service are intercepted by RDS
-"""
-
-url_bases = []  # type: ignore[var-annotated]
-
-
-url_paths = {}  # type: ignore[var-annotated]
diff --git a/moto/rds/models.py b/moto/rds/models.py
index d22b188c9..4e21129a9 100644
--- a/moto/rds/models.py
+++ b/moto/rds/models.py
@@ -3,6 +3,7 @@ import os
 import re
 import string
 from collections import OrderedDict, defaultdict
+from functools import lru_cache
 from re import compile as re_compile
 from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Tuple, Union
 
@@ -13,7 +14,6 @@ from moto.core.common_models import BaseModel, CloudFormationModel
 from moto.core.utils import iso_8601_datetime_with_milliseconds
 from moto.ec2.models import ec2_backends
 from moto.moto_api._internal import mock_random as random
-from moto.neptune.models import NeptuneBackend, neptune_backends
 from moto.utilities.utils import ARN_PARTITION_REGEX, load_resource
 
 from .exceptions import (
@@ -137,7 +137,9 @@ class GlobalCluster(RDSBaseModel):
         self.global_cluster_identifier = global_cluster_identifier
         self.global_cluster_resource_id = "cluster-" + random.get_random_hex(8)
         self.engine = engine
-        self.engine_version = engine_version or "5.7.mysql_aurora.2.11.2"
+        self.engine_version = engine_version or DBCluster.default_engine_version(
+            self.engine
+        )
         self.storage_encrypted = (
             storage_encrypted and storage_encrypted.lower() == "true"
         )
@@ -249,7 +251,11 @@ class DBCluster(RDSBaseModel):
             )
         self.master_username = kwargs.get("master_username")
         self.global_cluster_identifier = kwargs.get("global_cluster_identifier")
-        if not self.master_username and self.global_cluster_identifier:
+        if (
+            not self.master_username
+            and self.global_cluster_identifier
+            or self.engine == "neptune"
+        ):
             pass
         elif not self.master_username:
             raise InvalidParameterValue(
@@ -265,7 +271,10 @@ class DBCluster(RDSBaseModel):
                 f"{self.region}b",
                 f"{self.region}c",
             ]
-        self.parameter_group = kwargs.get("parameter_group") or "default.aurora8.0"
+        default_pg = (
+            "default.neptune1.3" if self.engine == "neptune" else "default.aurora8.0"
+        )
+        self.parameter_group = kwargs.get("parameter_group") or default_pg
         self.subnet_group = kwargs.get("db_subnet_group_name") or "default"
         self.url_identifier = "".join(
             random.choice(string.ascii_lowercase + string.digits) for _ in range(12)
@@ -539,6 +548,7 @@ class DBCluster(RDSBaseModel):
             "aurora-mysql": "5.7.mysql_aurora.2.07.2",
             "aurora-postgresql": "12.7",
             "mysql": "8.0.23",
+            "neptune": "1.3.2.1",
             "postgres": "13.4",
         }[engine]
 
@@ -549,6 +559,7 @@ class DBCluster(RDSBaseModel):
             "aurora-mysql": 3306,
             "aurora-postgresql": 5432,
             "mysql": 3306,
+            "neptune": 8182,
             "postgres": 5432,
         }[engine]
 
@@ -566,6 +577,7 @@ class DBCluster(RDSBaseModel):
             "aurora-mysql": {"gp2": 20, "io1": 100, "standard": 10},
             "aurora-postgresql": {"gp2": 20, "io1": 100, "standard": 10},
             "mysql": {"gp2": 20, "io1": 100, "standard": 5},
+            "neptune": {"gp2": 0, "io1": 0, "standard": 0},
             "postgres": {"gp2": 20, "io1": 100, "standard": 5},
         }[engine][storage_type]
 
@@ -1651,24 +1663,18 @@ class RDSBackend(BaseBackend):
         self.db_proxies: Dict[str, DBProxy] = OrderedDict()
 
     def reset(self) -> None:
-        self.neptune.reset()
         super().reset()
 
-    @property
-    def neptune(self) -> NeptuneBackend:
-        return neptune_backends[self.account_id][self.region_name]
+    @lru_cache()
+    def db_cluster_options(self, engine) -> List[Dict[str, Any]]:  # type: ignore
+        from moto.rds.utils import decode_orderable_db_instance
 
-    @property
-    def db_cluster_options(self) -> List[Dict[str, Any]]:  # type: ignore
-        if self._db_cluster_options is None:
-            from moto.rds.utils import decode_orderable_db_instance
-
-            decoded_options = load_resource(
-                __name__, "resources/cluster_options/aurora-postgresql.json"
-            )
-            self._db_cluster_options = [
-                decode_orderable_db_instance(option) for option in decoded_options
-            ]
+        decoded_options = load_resource(
+            __name__, f"resources/cluster_options/{engine}.json"
+        )
+        self._db_cluster_options = [
+            decode_orderable_db_instance(option) for option in decoded_options
+        ]
         return self._db_cluster_options
 
     def create_db_instance(self, db_kwargs: Dict[str, Any]) -> DBInstance:
@@ -2290,23 +2296,7 @@ class RDSBackend(BaseBackend):
         cluster = DBCluster(self, **kwargs)
         self.clusters[cluster_id] = cluster
 
-        if (
-            cluster.global_cluster_identifier
-            and cluster.global_cluster_identifier in self.global_clusters
-        ):
-            global_cluster = self.global_clusters[cluster.global_cluster_identifier]
-
-            # Main DB cluster, does RW on global cluster
-            setattr(cluster, "is_writer", True)
-            # self.clusters[cluster_id] = cluster
-            global_cluster.members.append(cluster)
-
-        # search all backend to check if global cluster named global_cluster_identifier exists
-        # anywhere else
-        if (
-            cluster.global_cluster_identifier
-            and cluster.global_cluster_identifier not in self.global_clusters
-        ):
+        if cluster.global_cluster_identifier:
             for regional_backend in rds_backends[self.account_id]:
                 if (
                     cluster.global_cluster_identifier
@@ -2316,6 +2306,12 @@ class RDSBackend(BaseBackend):
                         regional_backend
                     ].global_clusters[cluster.global_cluster_identifier]
                     global_cluster.members.append(cluster)
+                    if len(global_cluster.members) == 1:
+                        # primary cluster
+                        setattr(cluster, "is_writer", True)
+                    else:
+                        # secondary cluster(s)
+                        setattr(cluster, "is_writer", False)
 
         if cluster.replication_source_identifier:
             cluster_identifier = cluster.replication_source_identifier
@@ -2329,14 +2325,13 @@ class RDSBackend(BaseBackend):
     def modify_db_cluster(self, kwargs: Dict[str, Any]) -> DBCluster:
         cluster_id = kwargs["db_cluster_identifier"]
 
-        if cluster_id in self.neptune.clusters:
-            return self.neptune.modify_db_cluster(kwargs)  # type: ignore
-
         cluster = self.clusters[cluster_id]
         del self.clusters[cluster_id]
 
         kwargs["db_cluster_identifier"] = kwargs.pop("new_db_cluster_identifier")
         for k, v in kwargs.items():
+            if k == "db_cluster_parameter_group_name":
+                k = "parameter_group"
             if v is not None:
                 setattr(cluster, k, v)
 
@@ -2426,17 +2421,13 @@ class RDSBackend(BaseBackend):
         self, cluster_identifier: Optional[str] = None, filters: Any = None
     ) -> List[DBCluster]:
         clusters = self.clusters
-        clusters_neptune = self.neptune.clusters
         if cluster_identifier:
             filters = merge_filters(filters, {"db-cluster-id": [cluster_identifier]})
         if filters:
             clusters = self._filter_resources(clusters, filters, DBCluster)
-            clusters_neptune = self._filter_resources(
-                clusters_neptune, filters, DBCluster
-            )
-        if cluster_identifier and not (clusters or clusters_neptune):
+        if cluster_identifier and not clusters:
             raise DBClusterNotFoundError(cluster_identifier)
-        return list(clusters.values()) + list(clusters_neptune.values())  # type: ignore
+        return list(clusters.values())  # type: ignore
 
     def describe_db_cluster_snapshots(
         self,
@@ -2475,13 +2466,10 @@ class RDSBackend(BaseBackend):
             if snapshot_name:
                 self.create_auto_cluster_snapshot(cluster_identifier, snapshot_name)
             return self.clusters.pop(cluster_identifier)
-        if cluster_identifier in self.neptune.clusters:
-            return self.neptune.delete_db_cluster(cluster_identifier)  # type: ignore
         raise DBClusterNotFoundError(cluster_identifier)
 
     def start_db_cluster(self, cluster_identifier: str) -> DBCluster:
         if cluster_identifier not in self.clusters:
-            return self.neptune.start_db_cluster(cluster_identifier)  # type: ignore
             raise DBClusterNotFoundError(cluster_identifier)
         cluster = self.clusters[cluster_identifier]
         if cluster.status != "stopped":
@@ -2605,8 +2593,6 @@ class RDSBackend(BaseBackend):
             elif resource_type == "cluster":  # Cluster
                 if resource_name in self.clusters:
                     return self.clusters[resource_name].get_tags()
-                if resource_name in self.neptune.clusters:
-                    return self.neptune.clusters[resource_name].get_tags()
             elif resource_type == "es":  # Event Subscription
                 if resource_name in self.event_subscriptions:
                     return self.event_subscriptions[resource_name].get_tags()
@@ -2669,8 +2655,6 @@ class RDSBackend(BaseBackend):
             elif resource_type == "cluster":
                 if resource_name in self.clusters:
                     self.clusters[resource_name].remove_tags(tag_keys)
-                if resource_name in self.neptune.clusters:
-                    self.neptune.clusters[resource_name].remove_tags(tag_keys)
             elif resource_type == "cluster-snapshot":  # DB Cluster Snapshot
                 if resource_name in self.cluster_snapshots:
                     self.cluster_snapshots[resource_name].remove_tags(tag_keys)
@@ -2715,8 +2699,6 @@ class RDSBackend(BaseBackend):
             elif resource_type == "cluster":
                 if resource_name in self.clusters:
                     return self.clusters[resource_name].add_tags(tags)
-                if resource_name in self.neptune.clusters:
-                    return self.neptune.clusters[resource_name].add_tags(tags)
             elif resource_type == "cluster-snapshot":  # DB Cluster Snapshot
                 if resource_name in self.cluster_snapshots:
                     return self.cluster_snapshots[resource_name].add_tags(tags)
@@ -2771,16 +2753,14 @@ class RDSBackend(BaseBackend):
         """
         Only the Aurora-Postgresql and Neptune-engine is currently implemented
         """
-        if engine == "neptune":
-            return self.neptune.describe_orderable_db_instance_options(engine_version)
-        if engine == "aurora-postgresql":
+        if engine in ["aurora-postgresql", "neptune"]:
             if engine_version:
                 return [
                     option
-                    for option in self.db_cluster_options
+                    for option in self.db_cluster_options(engine)
                     if option["EngineVersion"] == engine_version
                 ]
-            return self.db_cluster_options
+            return self.db_cluster_options(engine)
         return []
 
     def create_db_cluster_parameter_group(
@@ -2853,16 +2833,9 @@ class RDSBackend(BaseBackend):
         return global_cluster
 
     def describe_global_clusters(self) -> List[GlobalCluster]:
-        return (
-            list(self.global_clusters.values())
-            + self.neptune.describe_global_clusters()  # type: ignore
-        )
+        return list(self.global_clusters.values())
 
     def delete_global_cluster(self, global_cluster_identifier: str) -> GlobalCluster:
-        try:
-            return self.neptune.delete_global_cluster(global_cluster_identifier)  # type: ignore
-        except:  # noqa: E722 Do not use bare except
-            pass  # It's not a Neptune Global Cluster - assume it's an RDS cluster instead
         global_cluster = self.global_clusters[global_cluster_identifier]
         if global_cluster.members:
             raise InvalidGlobalClusterStateFault(global_cluster.global_cluster_arn)
diff --git a/moto/rds/responses.py b/moto/rds/responses.py
index 635851c89..b39bd61cd 100644
--- a/moto/rds/responses.py
+++ b/moto/rds/responses.py
@@ -1,42 +1,22 @@
 from collections import defaultdict
 from typing import Any, Dict, Iterable, List
 
-from moto.core.common_types import TYPE_RESPONSE
 from moto.core.responses import BaseResponse
 from moto.ec2.models import ec2_backends
-from moto.neptune.responses import (
-    CREATE_GLOBAL_CLUSTER_TEMPLATE,
-    DELETE_GLOBAL_CLUSTER_TEMPLATE,
-    DESCRIBE_GLOBAL_CLUSTERS_TEMPLATE,
-    REMOVE_FROM_GLOBAL_CLUSTER_TEMPLATE,
-    NeptuneResponse,
-)
 
 from .exceptions import DBParameterGroupNotFoundError
 from .models import RDSBackend, rds_backends
 
 
 class RDSResponse(BaseResponse):
-    def __init__(self) -> None:
-        super().__init__(service_name="rds")
-        # Neptune and RDS share a HTTP endpoint RDS is the lucky guy that catches all requests
-        # So we have to determine whether we can handle an incoming request here, or whether it needs redirecting to Neptune
-        self.neptune = NeptuneResponse()
-
     @property
     def backend(self) -> RDSBackend:
         return rds_backends[self.current_account][self.region]
 
-    def _dispatch(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE:
-        # Because some requests are send through to Neptune, we have to prepare the NeptuneResponse-class
-        self.neptune.setup_class(request, full_url, headers)
-        return super()._dispatch(request, full_url, headers)
-
-    def __getattribute__(self, name: str) -> Any:
-        if name in ["create_db_cluster", "create_global_cluster"]:
-            if self._get_param("Engine") == "neptune":
-                return object.__getattribute__(self.neptune, name)
-        return object.__getattribute__(self, name)
+    @property
+    def global_backend(self) -> RDSBackend:
+        """Return backend instance of the region that stores Global Clusters"""
+        return rds_backends[self.current_account]["us-east-1"]
 
     def _get_db_kwargs(self) -> Dict[str, Any]:
         args = {
@@ -756,13 +736,13 @@ class RDSResponse(BaseResponse):
         return template.render(options=options, marker=None)
 
     def describe_global_clusters(self) -> str:
-        clusters = self.backend.describe_global_clusters()
+        clusters = self.global_backend.describe_global_clusters()
         template = self.response_template(DESCRIBE_GLOBAL_CLUSTERS_TEMPLATE)
         return template.render(clusters=clusters)
 
     def create_global_cluster(self) -> str:
         params = self._get_params()
-        cluster = self.backend.create_global_cluster(
+        cluster = self.global_backend.create_global_cluster(
             global_cluster_identifier=params["GlobalClusterIdentifier"],
             source_db_cluster_identifier=params.get("SourceDBClusterIdentifier"),
             engine=params.get("Engine"),
@@ -775,7 +755,7 @@ class RDSResponse(BaseResponse):
 
     def delete_global_cluster(self) -> str:
         params = self._get_params()
-        cluster = self.backend.delete_global_cluster(
+        cluster = self.global_backend.delete_global_cluster(
             global_cluster_identifier=params["GlobalClusterIdentifier"],
         )
         template = self.response_template(DELETE_GLOBAL_CLUSTER_TEMPLATE)
@@ -1699,3 +1679,53 @@ DESCRIBE_DB_PROXIES_TEMPLATE = """<DescribeDBProxiesResponse xmlns="http://rds.a
     </ResponseMetadata>
 </DescribeDBProxiesResponse>
 """
+
+CREATE_GLOBAL_CLUSTER_TEMPLATE = """<CreateGlobalClusterResponse xmlns="http://rds.amazonaws.com/doc/2014-10-31/">
+  <ResponseMetadata>
+    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
+  </ResponseMetadata>
+  <CreateGlobalClusterResult>
+  <GlobalCluster>
+    {{ cluster.to_xml() }}
+  </GlobalCluster>
+  </CreateGlobalClusterResult>
+</CreateGlobalClusterResponse>"""
+
+DELETE_GLOBAL_CLUSTER_TEMPLATE = """<DeleteGlobalClusterResponse xmlns="http://rds.amazonaws.com/doc/2014-10-31/">
+  <ResponseMetadata>
+    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
+  </ResponseMetadata>
+  <DeleteGlobalClusterResult>
+  <GlobalCluster>
+    {{ cluster.to_xml() }}
+  </GlobalCluster>
+  </DeleteGlobalClusterResult>
+</DeleteGlobalClusterResponse>"""
+
+DESCRIBE_GLOBAL_CLUSTERS_TEMPLATE = """<DescribeGlobalClustersResponse xmlns="http://rds.amazonaws.com/doc/2014-10-31/">
+  <ResponseMetadata>
+    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
+  </ResponseMetadata>
+  <DescribeGlobalClustersResult>
+    <GlobalClusters>
+{% for cluster in clusters %}
+    <GlobalClusterMember>
+        {{ cluster.to_xml() }}
+        </GlobalClusterMember>
+{% endfor %}
+    </GlobalClusters>
+  </DescribeGlobalClustersResult>
+</DescribeGlobalClustersResponse>"""
+
+REMOVE_FROM_GLOBAL_CLUSTER_TEMPLATE = """<RemoveFromGlobalClusterResponse xmlns="http://rds.amazonaws.com/doc/2014-10-31/">
+  <ResponseMetadata>
+    <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId>
+  </ResponseMetadata>
+  <RemoveFromGlobalClusterResult>
+  {% if cluster %}
+  <GlobalCluster>
+    {{ cluster.to_xml() }}
+  </GlobalCluster>
+  {% endif %}
+  </RemoveFromGlobalClusterResult>
+</RemoveFromGlobalClusterResponse>"""
diff --git a/moto/rds/utils.py b/moto/rds/utils.py
index 41dabc435..ebb75d898 100644
--- a/moto/rds/utils.py
+++ b/moto/rds/utils.py
@@ -51,6 +51,7 @@ class DbInstanceEngine(str, Enum):
 class ClusterEngine(str, Enum):
     AURORA_POSTGRESQL = "aurora-postgresql"
     AURORA_MYSQL = "aurora-mysql"
+    NEPTUNE = "neptune"
     RDS_POSTGRESQL = "postgres"
     RDS_MYSQL = "mysql"
 
diff --git a/moto/sagemaker/models.py b/moto/sagemaker/models.py
index 13c9e6b95..1025fabfe 100644
--- a/moto/sagemaker/models.py
+++ b/moto/sagemaker/models.py
@@ -127,6 +127,12 @@ PAGINATION_MODEL = {
         "limit_default": 100,
         "unique_attribute": "arn",
     },
+    "list_model_quality_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]]
@@ -2315,6 +2321,76 @@ class HyperParameterTuningJob(BaseObject):
         }
 
 
+class ModelQualityJobDefinition(BaseObject):
+    def __init__(
+        self,
+        job_definition_name: str,
+        model_quality_baseline_config: Optional[Dict[str, Any]],
+        model_quality_app_specification: Dict[str, Any],
+        model_quality_job_input: Dict[str, Any],
+        model_quality_job_output_config: Dict[str, Any],
+        job_resources: Dict[str, Any],
+        network_config: Optional[Dict[str, Any]],
+        role_arn: str,
+        stopping_condition: Optional[Dict[str, Any]],
+        tags: Optional[List[Dict[str, str]]],
+        region_name: str,
+        account_id: str,
+    ):
+        self.region_name = region_name
+        self.account_id = account_id
+        self.job_definition_name = job_definition_name
+        if (
+            job_definition_name
+            in sagemaker_backends[account_id][region_name].model_quality_job_definitions
+        ):
+            raise ResourceInUseException(
+                message=f"Resource Already Exists: Model Quality Job Definition with name {job_definition_name} already exists. Choose a different name."
+            )
+        self.model_quality_baseline_config = model_quality_baseline_config
+        self.model_quality_app_specification = model_quality_app_specification
+        self.model_quality_job_input = model_quality_job_input
+        self.model_quality_job_output_config = model_quality_job_output_config
+        self.job_resources = job_resources
+        self.network_config = network_config
+        self.role_arn = role_arn
+        self.stopping_condition = stopping_condition
+        self.tags = tags or []
+        self.arn = arn_formatter(
+            "model-quality-job-definition",
+            self.job_definition_name,
+            account_id,
+            region_name,
+        )
+        self.creation_time = str(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+        self.endpoint_name = self.model_quality_job_input["EndpointInput"][
+            "EndpointName"
+        ]
+
+    def describe(self) -> Dict[str, Any]:
+        return {
+            "JobDefinitionArn": self.arn,
+            "JobDefinitionName": self.job_definition_name,
+            "CreationTime": self.creation_time,
+            "ModelQualityBaselineConfig": self.model_quality_baseline_config,
+            "ModelQualityAppSpecification": self.model_quality_app_specification,
+            "ModelQualityJobInput": self.model_quality_job_input,
+            "ModelQualityJobOutputConfig": self.model_quality_job_output_config,
+            "JobResources": self.job_resources,
+            "NetworkConfig": self.network_config,
+            "RoleArn": self.role_arn,
+            "StoppingCondition": self.stopping_condition,
+        }
+
+    def summary(self) -> Dict[str, Any]:
+        return {
+            "MonitoringJobDefinitionName": self.job_definition_name,
+            "MonitoringJobDefinitionArn": self.arn,
+            "CreationTime": self.creation_time,
+            "EndpointName": self.endpoint_name,
+        }
+
+
 class VpcConfig(BaseObject):
     def __init__(self, security_group_ids: List[str], subnets: List[str]):
         self.security_group_ids = security_group_ids
@@ -2716,6 +2792,7 @@ class SageMakerModelBackend(BaseBackend):
             str, ModelExplainabilityJobDefinition
         ] = {}
         self.hyper_parameter_tuning_jobs: Dict[str, HyperParameterTuningJob] = {}
+        self.model_quality_job_definitions: Dict[str, ModelQualityJobDefinition] = {}
 
     @staticmethod
     def default_vpc_endpoint_service(
@@ -2851,6 +2928,7 @@ class SageMakerModelBackend(BaseBackend):
             "domain": self.domains,
             "model-explainability-job-definition": self.model_explainability_job_definitions,
             "hyper-parameter-tuning-job": self.hyper_parameter_tuning_jobs,
+            "model-quality-job-definition": self.model_quality_job_definitions,
         }
         target_resource, target_name = arn.split(":")[-1].split("/")
         try:
@@ -5258,6 +5336,107 @@ class SageMakerModelBackend(BaseBackend):
         del self.hyper_parameter_tuning_jobs[hyper_parameter_tuning_job_name]
         return
 
+    def create_model_quality_job_definition(
+        self,
+        job_definition_name: str,
+        model_quality_baseline_config: Optional[Dict[str, Any]],
+        model_quality_app_specification: Dict[str, Any],
+        model_quality_job_input: Dict[str, Any],
+        model_quality_job_output_config: Dict[str, Any],
+        job_resources: Dict[str, Any],
+        network_config: Optional[Dict[str, Any]],
+        role_arn: str,
+        stopping_condition: Optional[Dict[str, Any]],
+        tags: Optional[List[Dict[str, str]]],
+    ) -> str:
+        model_quality_job_definition = ModelQualityJobDefinition(
+            job_definition_name=job_definition_name,
+            model_quality_baseline_config=model_quality_baseline_config,
+            model_quality_app_specification=model_quality_app_specification,
+            model_quality_job_input=model_quality_job_input,
+            model_quality_job_output_config=model_quality_job_output_config,
+            job_resources=job_resources,
+            network_config=network_config,
+            role_arn=role_arn,
+            stopping_condition=stopping_condition,
+            region_name=self.region_name,
+            account_id=self.account_id,
+            tags=tags,
+        )
+        self.model_quality_job_definitions[job_definition_name] = (
+            model_quality_job_definition
+        )
+        return model_quality_job_definition.arn
+
+    def describe_model_quality_job_definition(
+        self, job_definition_name: str
+    ) -> Dict[str, Any]:
+        if job_definition_name not in self.model_quality_job_definitions:
+            raise ResourceNotFound(
+                message=f"Could not find model quality job definition '{job_definition_name}'."
+            )
+        return self.model_quality_job_definitions[job_definition_name].describe()
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_model_quality_job_definitions(
+        self,
+        endpoint_name: Optional[str],
+        sort_by: Optional[str],
+        sort_order: Optional[str],
+        name_contains: Optional[str],
+        creation_time_before: Optional[str],
+        creation_time_after: Optional[str],
+    ) -> List[ModelQualityJobDefinition]:
+        model_quality_job_definitions = list(
+            self.model_quality_job_definitions.values()
+        )
+        if endpoint_name:
+            model_quality_job_definitions = [
+                i
+                for i in model_quality_job_definitions
+                if endpoint_name == i.endpoint_name
+            ]
+        if name_contains:
+            model_quality_job_definitions = [
+                i
+                for i in model_quality_job_definitions
+                if name_contains in i.job_definition_name
+            ]
+        if creation_time_before:
+            model_quality_job_definitions = [
+                i
+                for i in model_quality_job_definitions
+                if i.creation_time < str(creation_time_before)
+            ]
+        if creation_time_after:
+            model_quality_job_definitions = [
+                i
+                for i in model_quality_job_definitions
+                if i.creation_time > str(creation_time_after)
+            ]
+        reverse = sort_order == "Descending"
+        if sort_by == "Name":
+            model_quality_job_definitions = sorted(
+                model_quality_job_definitions,
+                key=lambda x: x.job_definition_name,
+                reverse=reverse,
+            )
+        if sort_by == "CreationTime" or sort_by is None:
+            model_quality_job_definitions = sorted(
+                model_quality_job_definitions,
+                key=lambda x: x.creation_time,
+                reverse=reverse,
+            )
+        return model_quality_job_definitions
+
+    def delete_model_quality_job_definition(self, job_definition_name: str) -> None:
+        if job_definition_name not in self.model_quality_job_definitions:
+            raise ResourceNotFound(
+                message=f"Could not find model quality job definition '{job_definition_name}'."
+            )
+        del self.model_quality_job_definitions[job_definition_name]
+        return
+
 
 class FakeExperiment(BaseObject):
     def __init__(
diff --git a/moto/sagemaker/responses.py b/moto/sagemaker/responses.py
index c123d55a2..225d64761 100644
--- a/moto/sagemaker/responses.py
+++ b/moto/sagemaker/responses.py
@@ -1409,3 +1409,70 @@ class SageMakerResponse(BaseResponse):
             hyper_parameter_tuning_job_name=hyper_parameter_tuning_job_name,
         )
         return json.dumps(dict())
+
+    def create_model_quality_job_definition(self) -> str:
+        job_definition_name = self._get_param("JobDefinitionName")
+        model_quality_baseline_config = self._get_param("ModelQualityBaselineConfig")
+        model_quality_app_specification = self._get_param(
+            "ModelQualityAppSpecification"
+        )
+        model_quality_job_input = self._get_param("ModelQualityJobInput")
+        model_quality_job_output_config = self._get_param("ModelQualityJobOutputConfig")
+        job_resources = self._get_param("JobResources")
+        network_config = self._get_param("NetworkConfig")
+        role_arn = self._get_param("RoleArn")
+        stopping_condition = self._get_param("StoppingCondition")
+        tags = self._get_param("Tags")
+        job_definition_arn = self.sagemaker_backend.create_model_quality_job_definition(
+            job_definition_name=job_definition_name,
+            model_quality_baseline_config=model_quality_baseline_config,
+            model_quality_app_specification=model_quality_app_specification,
+            model_quality_job_input=model_quality_job_input,
+            model_quality_job_output_config=model_quality_job_output_config,
+            job_resources=job_resources,
+            network_config=network_config,
+            role_arn=role_arn,
+            stopping_condition=stopping_condition,
+            tags=tags,
+        )
+        return json.dumps(dict(JobDefinitionArn=job_definition_arn))
+
+    def describe_model_quality_job_definition(self) -> str:
+        job_definition_name = self._get_param("JobDefinitionName")
+        description = self.sagemaker_backend.describe_model_quality_job_definition(
+            job_definition_name=job_definition_name,
+        )
+        return json.dumps(description)
+
+    def list_model_quality_job_definitions(self) -> str:
+        endpoint_name = self._get_param("EndpointName")
+        sort_by = self._get_param("SortBy")
+        sort_order = self._get_param("SortOrder")
+        next_token = self._get_param("NextToken")
+        max_results = self._get_param("MaxResults")
+        name_contains = self._get_param("NameContains")
+        creation_time_before = self._get_param("CreationTimeBefore")
+        creation_time_after = self._get_param("CreationTimeAfter")
+        job_definitions, next_token = (
+            self.sagemaker_backend.list_model_quality_job_definitions(
+                endpoint_name=endpoint_name,
+                sort_by=sort_by,
+                sort_order=sort_order,
+                next_token=next_token,
+                max_results=max_results,
+                name_contains=name_contains,
+                creation_time_before=creation_time_before,
+                creation_time_after=creation_time_after,
+            )
+        )
+        job_definition_summaries = [x.summary() for x in job_definitions]
+        return json.dumps(
+            dict(JobDefinitionSummaries=job_definition_summaries, NextToken=next_token)
+        )
+
+    def delete_model_quality_job_definition(self) -> str:
+        job_definition_name = self._get_param("JobDefinitionName")
+        self.sagemaker_backend.delete_model_quality_job_definition(
+            job_definition_name=job_definition_name,
+        )
+        return json.dumps(dict())


Test patch:
diff --git a/tests/test_neptune/test_clusters.py b/tests/test_neptune/test_clusters.py
index c1dfdd343..2d9376de1 100644
--- a/tests/test_neptune/test_clusters.py
+++ b/tests/test_neptune/test_clusters.py
@@ -18,9 +18,9 @@ def test_create_db_cluster():
     assert "DbClusterResourceId" in resp
     assert "DBClusterArn" in resp
     assert resp["Engine"] == "neptune"
-    assert resp["EngineVersion"] == "1.2.0.2"
-    assert resp["StorageEncrypted"] is True
-    assert resp["DBClusterParameterGroup"] == ""
+    assert "EngineVersion" in resp
+    assert resp["StorageEncrypted"] is False
+    assert resp["DBClusterParameterGroup"].startswith("default.neptune")
     assert "Endpoint" in resp
     assert "cluster-" in resp["DbClusterResourceId"]
     assert resp["AvailabilityZones"] == ["us-east-2a", "us-east-2b", "us-east-2c"]
@@ -108,9 +108,7 @@ def test_modify_db_cluster():
 @mock_aws
 def test_start_db_cluster():
     client = boto3.client("neptune", region_name="us-east-2")
-    client.create_db_cluster(DBClusterIdentifier="cluster-id", Engine="neptune")[
-        "DBCluster"
-    ]
-
+    client.create_db_cluster(DBClusterIdentifier="cluster-id", Engine="neptune")
+    client.stop_db_cluster(DBClusterIdentifier="cluster-id")
     cluster = client.start_db_cluster(DBClusterIdentifier="cluster-id")["DBCluster"]
     assert cluster["Status"] == "started"
diff --git a/tests/test_neptune/test_global_clusters.py b/tests/test_neptune/test_global_clusters.py
index edf207ad1..f8c8aa468 100644
--- a/tests/test_neptune/test_global_clusters.py
+++ b/tests/test_neptune/test_global_clusters.py
@@ -19,7 +19,7 @@ def test_create_global_cluster():
     assert "GlobalClusterResourceId" in resp
     assert "GlobalClusterArn" in resp
     assert resp["Engine"] == "neptune"
-    assert resp["EngineVersion"] == "1.2.0.0"
+    assert "EngineVersion" in resp
     assert resp["StorageEncrypted"] is False
     assert resp["DeletionProtection"] is False
 
diff --git a/tests/test_rds/test_global_clusters.py b/tests/test_rds/test_global_clusters.py
index be84baa71..d61208b5e 100644
--- a/tests/test_rds/test_global_clusters.py
+++ b/tests/test_rds/test_global_clusters.py
@@ -39,7 +39,7 @@ def test_global_cluster_members():
     )
     assert global_cluster["Status"] == "available"
     assert global_cluster["Engine"] == "aurora-mysql"
-    assert global_cluster["EngineVersion"] == "5.7.mysql_aurora.2.11.2"
+    assert "mysql_aurora" in global_cluster["EngineVersion"]
     assert global_cluster["StorageEncrypted"] is False
     assert global_cluster["DeletionProtection"] is False
     assert global_cluster["GlobalClusterMembers"] == []
diff --git a/tests/test_sagemaker/test_sagemaker_model_quality.py b/tests/test_sagemaker/test_sagemaker_model_quality.py
new file mode 100644
index 000000000..641e54ac4
--- /dev/null
+++ b/tests/test_sagemaker/test_sagemaker_model_quality.py
@@ -0,0 +1,584 @@
+"""Unit tests for sagemaker-supported APIs."""
+
+import datetime
+
+import boto3
+
+from moto import mock_aws
+from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
+
+# See our Development Tips on writing tests for hints on how to write good tests:
+# http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
+
+
+@mock_aws
+def test_create_model_quality_job_definition():
+    client = boto3.client("sagemaker", region_name="eu-west-1")
+    arn = client.create_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition",
+        ModelQualityBaselineConfig={
+            "BaseliningJobName": "testbaseliningjobname",
+            "ConstraintsResource": {"S3Uri": "s3://constrain/resource"},
+        },
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+            "Environment": {"test": "test"},
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "testendpointname",
+                "LocalPath": "testlocalpath",
+                "S3DataDistributionType": "FullyReplicated",
+                "S3InputMode": "File",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                        "S3UploadMode": "Continuous",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        NetworkConfig={"EnableInterContainerTrafficEncryption": True},
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+        StoppingCondition={"MaxRuntimeInSeconds": 123},
+        Tags=[{"Key": "testkey", "Value": "testvalue"}],
+    )
+    assert (
+        arn["JobDefinitionArn"]
+        == f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition"
+    )
+
+
+@mock_aws
+def test_describe_model_quality_job_definition():
+    client = boto3.client("sagemaker", region_name="eu-west-1")
+    client.create_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition",
+        ModelQualityBaselineConfig={
+            "BaseliningJobName": "testbaseliningjobname",
+            "ConstraintsResource": {"S3Uri": "s3://constrain/resource"},
+        },
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+            "Environment": {"test": "test"},
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "testendpointname",
+                "LocalPath": "testlocalpath",
+                "S3DataDistributionType": "FullyReplicated",
+                "S3InputMode": "File",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                        "S3UploadMode": "Continuous",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        NetworkConfig={"EnableInterContainerTrafficEncryption": True},
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+        StoppingCondition={"MaxRuntimeInSeconds": 123},
+        Tags=[{"Key": "testkey", "Value": "testvalue"}],
+    )
+    resp = client.describe_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition"
+    )
+    assert (
+        resp["JobDefinitionArn"]
+        == f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition"
+    )
+    assert resp["JobDefinitionName"] == "testmodelqualityjobdefinition"
+    assert resp["ModelQualityBaselineConfig"] == {
+        "BaseliningJobName": "testbaseliningjobname",
+        "ConstraintsResource": {"S3Uri": "s3://constrain/resource"},
+    }
+    assert resp["ModelQualityAppSpecification"] == {
+        "ImageUri": "testimageuri",
+        "Environment": {"test": "test"},
+    }
+    assert resp["ModelQualityJobInput"] == {
+        "EndpointInput": {
+            "EndpointName": "testendpointname",
+            "LocalPath": "testlocalpath",
+            "S3DataDistributionType": "FullyReplicated",
+            "S3InputMode": "File",
+        },
+        "GroundTruthS3Input": {
+            "S3Uri": "s3://groundtruth/test",
+        },
+    }
+    assert resp["ModelQualityJobOutputConfig"] == {
+        "MonitoringOutputs": [
+            {
+                "S3Output": {
+                    "S3Uri": "s3://output/uri",
+                    "LocalPath": "testlocalpath",
+                    "S3UploadMode": "Continuous",
+                }
+            },
+        ]
+    }
+    assert resp["JobResources"] == {
+        "ClusterConfig": {
+            "InstanceCount": 1,
+            "InstanceType": "ml.m4.xlarge",
+            "VolumeSizeInGB": 4,
+        },
+    }
+    assert resp["NetworkConfig"] == {"EnableInterContainerTrafficEncryption": True}
+    assert resp["RoleArn"] == f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole"
+
+
+@mock_aws
+def test_describe_model_quality_job_definition_defaults():
+    client = boto3.client("sagemaker", region_name="eu-west-1")
+    client.create_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition",
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "testendpointname",
+                "LocalPath": "testlocalpath",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+    )
+    resp = client.describe_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition"
+    )
+    assert (
+        resp["JobDefinitionArn"]
+        == f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition"
+    )
+    assert resp["JobDefinitionName"] == "testmodelqualityjobdefinition"
+    assert resp["ModelQualityAppSpecification"] == {
+        "ImageUri": "testimageuri",
+    }
+    assert resp["ModelQualityJobInput"] == {
+        "EndpointInput": {
+            "EndpointName": "testendpointname",
+            "LocalPath": "testlocalpath",
+        },
+        "GroundTruthS3Input": {
+            "S3Uri": "s3://groundtruth/test",
+        },
+    }
+    assert resp["ModelQualityJobOutputConfig"] == {
+        "MonitoringOutputs": [
+            {
+                "S3Output": {
+                    "S3Uri": "s3://output/uri",
+                    "LocalPath": "testlocalpath",
+                }
+            },
+        ]
+    }
+    assert resp["JobResources"] == {
+        "ClusterConfig": {
+            "InstanceCount": 1,
+            "InstanceType": "ml.m4.xlarge",
+            "VolumeSizeInGB": 4,
+        },
+    }
+    assert resp["RoleArn"] == f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole"
+
+
+@mock_aws
+def test_list_model_quality_job_definitions():
+    client = boto3.client("sagemaker", region_name="eu-west-1")
+    client.create_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition",
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "testendpointname",
+                "LocalPath": "testlocalpath",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+    )
+    client.create_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition2",
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "testendpointname2",
+                "LocalPath": "testlocalpath",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+    )
+    resp = client.list_model_quality_job_definitions()["JobDefinitionSummaries"]
+    assert resp[0]["MonitoringJobDefinitionName"] == "testmodelqualityjobdefinition"
+    assert (
+        resp[0]["MonitoringJobDefinitionArn"]
+        == f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition"
+    )
+    assert isinstance(resp[0]["CreationTime"], datetime.datetime)
+    assert resp[0]["EndpointName"] == "testendpointname"
+    assert resp[1]["MonitoringJobDefinitionName"] == "testmodelqualityjobdefinition2"
+    assert (
+        resp[1]["MonitoringJobDefinitionArn"]
+        == f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition2"
+    )
+    assert isinstance(resp[1]["CreationTime"], datetime.datetime)
+    assert resp[1]["EndpointName"] == "testendpointname2"
+
+
+@mock_aws
+def test_list_model_quality_job_definitions_filters():
+    client = boto3.client("sagemaker", region_name="eu-west-1")
+    client.create_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition",
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "testendpointname",
+                "LocalPath": "testlocalpath",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+    )
+    client.create_model_quality_job_definition(
+        JobDefinitionName="test2modelqualityjobdefinition2",
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "testendpointname",
+                "LocalPath": "testlocalpath",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+    )
+    client.create_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition3",
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "test3endpointname3",
+                "LocalPath": "testlocalpath",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+    )
+    resp = client.list_model_quality_job_definitions(
+        EndpointName="testendpointname",
+        SortBy="Name",
+        SortOrder="Ascending",
+        NameContains="testmodelqualityjobdefinition",
+        CreationTimeBefore=str(datetime.datetime(2099, 1, 1)),
+        CreationTimeAfter=str(datetime.datetime(2021, 1, 1)),
+    )["JobDefinitionSummaries"]
+    assert len(resp) == 1
+    assert resp[0]["MonitoringJobDefinitionName"] == "testmodelqualityjobdefinition"
+    assert (
+        resp[0]["MonitoringJobDefinitionArn"]
+        == f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition"
+    )
+    assert isinstance(resp[0]["CreationTime"], datetime.datetime)
+    assert resp[0]["EndpointName"] == "testendpointname"
+
+
+@mock_aws
+def test_delete_model_quality_job_definition():
+    client = boto3.client("sagemaker", region_name="eu-west-1")
+    client.create_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition",
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "testendpointname",
+                "LocalPath": "testlocalpath",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+    )
+    client.create_model_quality_job_definition(
+        JobDefinitionName="test2modelqualityjobdefinition2",
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "testendpointname",
+                "LocalPath": "testlocalpath",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+    )
+    client.delete_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition"
+    )
+    resp = client.list_model_quality_job_definitions()
+    assert len(resp["JobDefinitionSummaries"]) == 1
+    assert (
+        resp["JobDefinitionSummaries"][0]["MonitoringJobDefinitionName"]
+        == "test2modelqualityjobdefinition2"
+    )
+
+
+@mock_aws
+def test_tag_model_quality_job_definition():
+    client = boto3.client("sagemaker", region_name="eu-west-1")
+    client.create_model_quality_job_definition(
+        JobDefinitionName="testmodelqualityjobdefinition",
+        ModelQualityAppSpecification={
+            "ImageUri": "testimageuri",
+        },
+        ModelQualityJobInput={
+            "EndpointInput": {
+                "EndpointName": "testendpointname",
+                "LocalPath": "testlocalpath",
+            },
+            "GroundTruthS3Input": {
+                "S3Uri": "s3://groundtruth/test",
+            },
+        },
+        ModelQualityJobOutputConfig={
+            "MonitoringOutputs": [
+                {
+                    "S3Output": {
+                        "S3Uri": "s3://output/uri",
+                        "LocalPath": "testlocalpath",
+                    }
+                },
+            ]
+        },
+        JobResources={
+            "ClusterConfig": {
+                "InstanceCount": 1,
+                "InstanceType": "ml.m4.xlarge",
+                "VolumeSizeInGB": 4,
+            },
+        },
+        RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole",
+        Tags=[{"Key": "testkey", "Value": "testvalue"}],
+    )
+    resp1 = client.list_tags(
+        ResourceArn=f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition"
+    )
+    assert resp1["Tags"] == [{"Key": "testkey", "Value": "testvalue"}]
+    client.add_tags(
+        ResourceArn=f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition",
+        Tags=[{"Key": "testkey2", "Value": "testvalue2"}],
+    )
+    resp2 = client.list_tags(
+        ResourceArn=f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition"
+    )
+    assert resp2["Tags"] == [
+        {"Key": "testkey", "Value": "testvalue"},
+        {"Key": "testkey2", "Value": "testvalue2"},
+    ]
+    client.delete_tags(
+        ResourceArn=f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition",
+        TagKeys=["testkey"],
+    )
+    resp3 = client.list_tags(
+        ResourceArn=f"arn:aws:sagemaker:eu-west-1:{ACCOUNT_ID}:model-quality-job-definition/testmodelqualityjobdefinition"
+    )
+    assert resp3["Tags"] == [{"Key": "testkey2", "Value": "testvalue2"}]