RDS: Check if new instance ID already exists (#8475)

moto/rds/exceptions.py

@@ -27,6 +27,11 @@ def __init__(self, database_identifier: str):
         )
 
 
+class DBInstanceAlreadyExists(RDSClientError):
+    def __init__(self) -> None:
+        super().__init__("DBInstanceAlreadyExists", "DB instance already exists")
+
+
 class DBSnapshotNotFoundError(RDSClientError):
     def __init__(self, snapshot_identifier: str):
         super().__init__(

moto/rds/models.py

@@ -23,6 +23,7 @@
     DBClusterSnapshotAlreadyExistsError,
     DBClusterSnapshotNotFoundError,
     DBClusterToBeDeletedHasActiveMembers,
+    DBInstanceAlreadyExists,
     DBInstanceNotFoundError,
     DBParameterGroupNotFoundError,
     DBProxyAlreadyExistsFault,
@@ -1822,6 +1823,8 @@ def db_cluster_options(self, engine) -> List[Dict[str, Any]]:  # type: ignore
 
     def create_db_instance(self, db_kwargs: Dict[str, Any]) -> DBInstance:
         database_id = db_kwargs["db_instance_identifier"]
+        if database_id in self.databases:
+            raise DBInstanceAlreadyExists()
         self._validate_db_identifier(database_id)
         database = DBInstance(self, **db_kwargs)
 
@@ -2040,6 +2043,10 @@ def restore_db_instance_from_db_snapshot(
             db_instance_identifier=None, db_snapshot_identifier=from_snapshot_id
         )[0]
         original_database = snapshot.database
+
+        if overrides["db_instance_identifier"] in self.databases:
+            raise DBInstanceAlreadyExists()
+
         new_instance_props = {}
         for key, value in original_database.__dict__.items():
             if key != "backend":

tests/test_rds/test_rds.py

@@ -71,6 +71,15 @@ def test_create_database(client):
     assert db_instance["DbInstancePort"] == 1234
 
 
+@mock_aws
+def test_create_database_already_exists():
+    create_db_instance()
+    with pytest.raises(ClientError) as exc:
+        create_db_instance()
+    err = exc.value.response["Error"]
+    assert err["Message"] == "DB instance already exists"
+
+
 @mock_aws
 def test_database_with_deletion_protection_cannot_be_deleted():
     db_instance = create_db_instance(DeletionProtection=True)
@@ -985,9 +994,6 @@ def test_restore_db_instance_from_db_snapshot(
         db_subnet_group_name = create_db_subnet_group()
 
     # restore
-    new_instance = client.restore_db_instance_from_db_snapshot(
-        DBInstanceIdentifier="db-restore-1", DBSnapshotIdentifier=db_snapshot_identifier
-    )["DBInstance"]
     kwargs = {
         "DBInstanceIdentifier": "db-restore-1",
         "DBSnapshotIdentifier": db_snapshot_identifier,
@@ -1024,6 +1030,23 @@ def test_restore_db_instance_from_db_snapshot(
     )
 
 
+@mock_aws
+def test_restore_db_instance_from_db_snapshot_called_twice(client):
+    create_db_instance(DBInstanceIdentifier="db-primary-1")
+    client.create_db_snapshot(
+        DBInstanceIdentifier="db-primary-1", DBSnapshotIdentifier="snapshot"
+    )
+    client.restore_db_instance_from_db_snapshot(
+        DBInstanceIdentifier="db-restore-1", DBSnapshotIdentifier="snapshot"
+    )
+    with pytest.raises(ClientError) as exc:
+        client.restore_db_instance_from_db_snapshot(
+            DBInstanceIdentifier="db-restore-1", DBSnapshotIdentifier="snapshot"
+        )
+    err = exc.value.response["Error"]
+    assert err["Message"] == "DB instance already exists"
+
+
 @pytest.mark.parametrize(
     "custom_db_subnet_group", [True, False], ids=("custom_subnet", "default_subnet")
 )