Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

rds - support final snapshot when delete a cluster (#5023)

Browse Source
This commit is contained in:
Kent Ou 2022-04-15 00:58:41 +10:00 committed by GitHub
parent 28d708c4a0
commit d46987ec29
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 27 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
moto/rds/models.py
View File

@ -1801,12 +1801,14 @@ class RDSBackend(BaseBackend):
raise DBClusterSnapshotNotFoundError(db_snapshot_identifier)
return list(snapshots.values())
def delete_db_cluster(self, cluster_identifier):
def delete_db_cluster(self, cluster_identifier, snapshot_name=None):
if cluster_identifier in self.clusters:
if self.clusters[cluster_identifier].deletion_protection:
raise InvalidParameterValue(
"Can't delete Cluster with protection enabled"
)
if snapshot_name:
self.create_db_cluster_snapshot(cluster_identifier, snapshot_name)
return self.clusters.pop(cluster_identifier)
raise DBClusterNotFoundError(cluster_identifier)

5
moto/rds/responses.py
View File

@ -507,7 +507,10 @@ class RDSResponse(BaseResponse):
def delete_db_cluster(self):
_id = self._get_param("DBClusterIdentifier")
cluster = self.backend.delete_db_cluster(cluster_identifier=_id)
snapshot_name = self._get_param("FinalDBSnapshotIdentifier")
cluster = self.backend.delete_db_cluster(
cluster_identifier=_id, snapshot_name=snapshot_name
)
template = self.response_template(DELETE_CLUSTER_TEMPLATE)
return template.render(cluster=cluster)

20
tests/test_rds/test_rds_clusters.py
View File

@ -210,6 +210,26 @@ def test_delete_db_cluster():
client.describe_db_clusters()["DBClusters"].should.have.length_of(0)
@mock_rds
def test_delete_db_cluster_do_snapshot():
client = boto3.client("rds", region_name="eu-north-1")
client.create_db_cluster(
DBClusterIdentifier="cluster-id",
Engine="aurora",
MasterUsername="root",
MasterUserPassword="hunter2_",
)
client.delete_db_cluster(
DBClusterIdentifier="cluster-id", FinalDBSnapshotIdentifier="final-snapshot"
)
client.describe_db_clusters()["DBClusters"].should.have.length_of(0)
snapshots = client.describe_db_cluster_snapshots()["DBClusterSnapshots"]
snapshots[0]["DBClusterIdentifier"].should.equal("cluster-id")
snapshots[0]["DBClusterSnapshotIdentifier"].should.equal("final-snapshot")
@mock_rds
def test_delete_db_cluster_that_is_protected():
client = boto3.client("rds", region_name="eu-north-1")