EC2 - Error when deleting snapshots in use (#4721)

moto/ec2/exceptions.py

@@ -275,6 +275,14 @@ class InvalidSnapshotIdError(EC2ClientError):
         )  # Note: AWS returns empty message for this, as of 2014.08.22.
 
 
+class InvalidSnapshotInUse(EC2ClientError):
+    def __init__(self, snapshot_id, ami_id):
+        super().__init__(
+            "InvalidSnapshot.InUse",
+            f"The snapshot {snapshot_id} is currently in use by {ami_id}",
+        )
+
+
 class InvalidVolumeIdError(EC2ClientError):
     def __init__(self, volume_id):
         super().__init__(

moto/ec2/models.py

@@ -84,6 +84,7 @@ from .exceptions import (
     InvalidSecurityGroupDuplicateError,
     InvalidSecurityGroupNotFoundError,
     InvalidSnapshotIdError,
+    InvalidSnapshotInUse,
     InvalidSubnetConflictError,
     InvalidSubnetIdError,
     InvalidSubnetRangeError,
@@ -1668,6 +1669,7 @@ class Ami(TaggedEC2Resource):
         root_device_name="/dev/sda1",
         sriov="simple",
         region_name="us-east-1a",
+        snapshot_description=None,
     ):
         self.ec2_backend = ec2_backend
         self.id = ami_id
@@ -1721,8 +1723,11 @@ class Ami(TaggedEC2Resource):
 
         # AWS auto-creates these, we should reflect the same.
         volume = self.ec2_backend.create_volume(size=15, zone_name=region_name)
+        snapshot_description = (
+            snapshot_description or "Auto-created snapshot for AMI %s" % self.id
+        )
         self.ebs_snapshot = self.ec2_backend.create_snapshot(
-            volume.id, "Auto-created snapshot for AMI %s" % self.id, owner_id
+            volume.id, snapshot_description, owner_id, from_ami=ami_id
         )
         self.ec2_backend.delete_volume(volume.id)
 
@@ -1802,6 +1807,7 @@ class AmiBackend(object):
             name=name,
             description=description,
             owner_id=OWNER_ID,
+            snapshot_description=f"Created by CreateImage({instance_id}) for {ami_id}",
         )
         for tag in tags:
             ami.add_tag(tag["Key"], tag["Value"])
@@ -3470,6 +3476,7 @@ class Snapshot(TaggedEC2Resource):
         description,
         encrypted=False,
         owner_id=OWNER_ID,
+        from_ami=None,
     ):
         self.id = snapshot_id
         self.volume = volume
@@ -3481,6 +3488,7 @@ class Snapshot(TaggedEC2Resource):
         self.status = "completed"
         self.encrypted = encrypted
         self.owner_id = owner_id
+        self.from_ami = from_ami
 
     def get_filter_value(self, filter_name):
         if filter_name == "description":
@@ -3609,12 +3617,14 @@ class EBSBackend(object):
         volume.attachment = None
         return old_attachment
 
-    def create_snapshot(self, volume_id, description, owner_id=None):
+    def create_snapshot(self, volume_id, description, owner_id=None, from_ami=None):
         snapshot_id = random_snapshot_id()
         volume = self.get_volume(volume_id)
         params = [self, snapshot_id, volume, description, volume.encrypted]
         if owner_id:
             params.append(owner_id)
+        if from_ami:
+            params.append(from_ami)
         snapshot = Snapshot(*params)
         self.snapshots[snapshot_id] = snapshot
         return snapshot
@@ -3653,6 +3663,9 @@ class EBSBackend(object):
 
     def delete_snapshot(self, snapshot_id):
         if snapshot_id in self.snapshots:
+            snapshot = self.snapshots[snapshot_id]
+            if snapshot.from_ami and snapshot.from_ami in self.amis:
+                raise InvalidSnapshotInUse(snapshot_id, snapshot.from_ami)
             return self.snapshots.pop(snapshot_id)
         raise InvalidSnapshotIdError(snapshot_id)
 

tests/test_ec2/test_amis.py

@@ -64,9 +64,7 @@ def test_ami_create_and_delete():
     )
     [s.id for s in snapshots].should.contain(retrieved_image_snapshot_id)
     snapshot = [s for s in snapshots if s.id == retrieved_image_snapshot_id][0]
-    snapshot.description.should.equal(
-        "Auto-created snapshot for AMI {0}".format(retrieved_image.id)
-    )
+    snapshot.description.should.match("Created by CreateImage")
 
     # root device should be in AMI's block device mappings
     root_mapping = retrieved_image.block_device_mapping.get(
@@ -167,8 +165,9 @@ def test_ami_create_and_delete_boto3():
     snapshot = [s for s in snapshots if s["SnapshotId"] == retrieved_image_snapshot_id][
         0
     ]
+    image_id = retrieved_image["ImageId"]
     snapshot["Description"].should.equal(
-        "Auto-created snapshot for AMI {0}".format(retrieved_image["ImageId"])
+        f"Created by CreateImage({instance_id}) for {image_id}"
     )
 
     # root device should be in AMI's block device mappings
@@ -1794,3 +1793,39 @@ def test_describe_images_dryrun():
     ex.value.response["Error"]["Message"].should.equal(
         "An error occurred (DryRunOperation) when calling the DescribeImages operation: Request would have succeeded, but DryRun flag is set"
     )
+
+
+@mock_ec2
+def test_delete_snapshot_from_create_image():
+    ec2_client = boto3.client("ec2", region_name="us-east-1")
+    resp = ec2_client.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
+    instance_id = resp["Instances"][0]["InstanceId"]
+    ami = ec2_client.create_image(InstanceId=instance_id, Name="test")
+    ami_id = ami["ImageId"]
+
+    snapshots = ec2_client.describe_snapshots(
+        Filters=[
+            {
+                "Name": "description",
+                "Values": ["Created by CreateImage(" + instance_id + "*"],
+            }
+        ]
+    )["Snapshots"]
+    snapshot_id = snapshots[0]["SnapshotId"]
+    with pytest.raises(ClientError) as exc:
+        ec2_client.delete_snapshot(SnapshotId=snapshot_id)
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("InvalidSnapshot.InUse")
+    err["Message"].should.equal(
+        f"The snapshot {snapshot_id} is currently in use by {ami_id}"
+    )
+
+    # Deregister the Ami first
+    ec2_client.deregister_image(ImageId=ami_id)
+
+    # Now we can delete the snapshot without problems
+    ec2_client.delete_snapshot(SnapshotId=snapshot_id)
+
+    with pytest.raises(ClientError) as exc:
+        ec2_client.describe_snapshots(SnapshotIds=[snapshot_id])
+    exc.value.response["Error"]["Code"].should.equal("InvalidSnapshot.NotFound")