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

EC2 - create_snapshots (#4767)

Browse Source
This commit is contained in:
Bert Blommers 2022-01-25 09:27:02 -01:00 committed by GitHub
parent 44e01a298e
commit d53dd23390
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 213 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
moto/core/responses.py
View File

@ -643,6 +643,12 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
# reset parent
obj = []
parent[keylist[i - 1]] = obj
elif isinstance(obj, dict):
# initialize dict
obj[key] = {}
# step into
parent = obj
obj = obj[key]
elif key.isdigit():
index = int(key) - 1
if len(obj) <= index:
@ -651,12 +657,6 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
# step into
parent = obj
obj = obj[index]
else:
# initialize dict
obj[key] = {}
# step into
parent = obj
obj = obj[key]
if isinstance(obj, list):
obj.append(value)
else:

24
moto/ec2/models.py
View File

@ -741,6 +741,7 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel):
self.platform = ami.platform if ami else None
self.virtualization_type = ami.virtualization_type if ami else "paravirtual"
self.architecture = ami.architecture if ami else "x86_64"
self.root_device_name = ami.root_device_name if ami else None
# handle weird bug around user_data -- something grabs the repr(), so
# it must be clean
@ -3625,6 +3626,29 @@ class EBSBackend(object):
self.snapshots[snapshot_id] = snapshot
return snapshot
def create_snapshots(self, instance_spec, description, tags):
"""
The CopyTagsFromSource-parameter is not yet implemented.
"""
instance = self.get_instance(instance_spec["InstanceId"])
block_device_mappings = instance.block_device_mapping
if str(instance_spec.get("ExcludeBootVolume", False)).lower() == "true":
volumes = [
m.volume_id
for k, m in block_device_mappings.items()
if k != instance.root_device_name
]
else:
volumes = [m.volume_id for m in block_device_mappings.values()]
snapshots = [
self.create_snapshot(v_id, description=description) for v_id in volumes
]
for snapshot in snapshots:
snapshot.add_tags(tags)
return snapshots
def describe_snapshots(self, snapshot_ids=None, filters=None):
matches = self.snapshots.copy().values()
if snapshot_ids:

43
moto/ec2/responses/elastic_block_store.py
View File

@ -39,6 +39,20 @@ class ElasticBlockStore(BaseResponse):
template = self.response_template(CREATE_SNAPSHOT_RESPONSE)
return template.render(snapshot=snapshot)
def create_snapshots(self):
params = self._get_params()
instance_spec = params.get("InstanceSpecification")
description = params.get("Description", "")
tags = self._parse_tag_specification("TagSpecification")
snapshot_tags = tags.get("snapshot", {})
if self.is_not_dryrun("CreateSnapshots"):
snapshots = self.ec2_backend.create_snapshots(
instance_spec, description, snapshot_tags
)
template = self.response_template(CREATE_SNAPSHOTS_RESPONSE)
return template.render(snapshots=snapshots)
def create_volume(self):
size = self._get_param("Size")
zone = self._get_param("AvailabilityZone")
@ -286,6 +300,35 @@ CREATE_SNAPSHOT_RESPONSE = """<CreateSnapshotResponse xmlns="http://ec2.amazonaw
</tagSet>
</CreateSnapshotResponse>"""
CREATE_SNAPSHOTS_RESPONSE = """<CreateSnapshotsResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<snapshotSet>
{% for snapshot in snapshots %}
<item>
<snapshotId>{{ snapshot.id }}</snapshotId>
<volumeId>{{ snapshot.volume.id }}</volumeId>
<status>pending</status>
<startTime>{{ snapshot.start_time}}</startTime>
<progress>60%</progress>
<ownerId>{{ snapshot.owner_id }}</ownerId>
<volumeSize>{{ snapshot.volume.size }}</volumeSize>
<description>{{ snapshot.description }}</description>
<encrypted>{{ 'true' if snapshot.encrypted else 'false' }}</encrypted>
<tagSet>
{% for tag in snapshot.get_tags() %}
<item>
<resourceId>{{ tag.resource_id }}</resourceId>
<resourceType>{{ tag.resource_type }}</resourceType>
<key>{{ tag.key }}</key>
<value>{{ tag.value }}</value>
</item>
{% endfor %}
</tagSet>
</item>
{% endfor %}
</snapshotSet>
</CreateSnapshotsResponse>"""
COPY_SNAPSHOT_RESPONSE = """<CopySnapshotResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<snapshotId>{{ snapshot.id }}</snapshotId>

140
tests/test_ec2/test_elastic_block_store.py
View File

@ -909,3 +909,143 @@ def test_create_volume_with_non_standard_type(volume_type):
volume = ec2.describe_volumes(VolumeIds=[volume["VolumeId"]])["Volumes"][0]
volume["VolumeType"].should.equal(volume_type)
@mock_ec2
def test_create_snapshots_dryrun():
client = boto3.client("ec2", region_name="us-east-1")
with pytest.raises(ClientError) as ex:
client.create_snapshots(
InstanceSpecification={"InstanceId": "asf"}, DryRun=True
)
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(412)
ex.value.response["Error"]["Code"].should.equal("DryRunOperation")
ex.value.response["Error"]["Message"].should.equal(
"An error occurred (DryRunOperation) when calling the CreateSnapshots operation: Request would have succeeded, but DryRun flag is set"
)
@mock_ec2
def test_create_snapshots_with_tagspecification():
client = boto3.client("ec2", region_name="us-east-1")
reservation = client.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
instance = reservation["Instances"][0]
resp = client.create_snapshots(
Description="my tagged snapshots",
InstanceSpecification={"InstanceId": instance["InstanceId"]},
TagSpecifications=[
{
"ResourceType": "snapshot",
"Tags": [
{"Key": "key1", "Value": "val1"},
{"Key": "key2", "Value": "val2"},
],
}
],
)
snapshots = resp["Snapshots"]
snapshots.should.have.length_of(1)
snapshots[0].should.have.key("Description").equals("my tagged snapshots")
snapshots[0].should.have.key("Tags").equals(
[{"Key": "key1", "Value": "val1"}, {"Key": "key2", "Value": "val2"}]
)
@mock_ec2
def test_create_snapshots_single_volume():
client = boto3.client("ec2", region_name="us-east-1")
reservation = client.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
instance = reservation["Instances"][0]
instance = client.describe_instances(InstanceIds=[instance["InstanceId"]])[
"Reservations"
][0]["Instances"][0]
boot_volume = instance["BlockDeviceMappings"][0]["Ebs"]
snapshots = client.create_snapshots(
InstanceSpecification={"InstanceId": instance["InstanceId"]}
)["Snapshots"]
snapshots.should.have.length_of(1)
snapshots[0].should.have.key("Encrypted").equals(False)
snapshots[0].should.have.key("VolumeId").equals(boot_volume["VolumeId"])
snapshots[0].should.have.key("VolumeSize").equals(8)
snapshots[0].should.have.key("SnapshotId")
snapshots[0].should.have.key("Description").equals("")
snapshots[0].should.have.key("Tags").equals([])
@mock_ec2
def test_create_snapshots_multiple_volumes():
client = boto3.client("ec2", region_name="us-east-1")
ec2 = boto3.resource("ec2", region_name="us-east-1")
reservation = client.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
instance = reservation["Instances"][0]
instance = client.describe_instances(InstanceIds=[instance["InstanceId"]])[
"Reservations"
][0]["Instances"][0]
boot_volume = instance["BlockDeviceMappings"][0]["Ebs"]
volume1 = ec2.create_volume(Size=80, AvailabilityZone="us-east-1a")
volume1.attach_to_instance(InstanceId=instance["InstanceId"], Device="/dev/sdh")
volume2 = ec2.create_volume(Size=100, AvailabilityZone="us-east-1b")
volume2.attach_to_instance(InstanceId=instance["InstanceId"], Device="/dev/sdg")
snapshots = client.create_snapshots(
InstanceSpecification={"InstanceId": instance["InstanceId"]}
)["Snapshots"]
# 3 Snapshots ; 1 boot, two additional volumes
snapshots.should.have.length_of(3)
# 3 unique snapshot IDs
set([s["SnapshotId"] for s in snapshots]).should.have.length_of(3)
boot_snapshot = next(
s for s in snapshots if s["VolumeId"] == boot_volume["VolumeId"]
)
boot_snapshot.should.have.key("VolumeSize").equals(8)
snapshot1 = next(s for s in snapshots if s["VolumeId"] == volume1.volume_id)
snapshot1.should.have.key("VolumeSize").equals(80)
snapshot2 = next(s for s in snapshots if s["VolumeId"] == volume2.volume_id)
snapshot2.should.have.key("VolumeSize").equals(100)
@mock_ec2
def test_create_snapshots_multiple_volumes_without_boot():
client = boto3.client("ec2", region_name="us-east-1")
ec2 = boto3.resource("ec2", region_name="us-east-1")
reservation = client.run_instances(ImageId=EXAMPLE_AMI_ID, MinCount=1, MaxCount=1)
instance = reservation["Instances"][0]
volume1 = ec2.create_volume(Size=80, AvailabilityZone="us-east-1a")
volume1.attach_to_instance(InstanceId=instance["InstanceId"], Device="/dev/sdh")
volume2 = ec2.create_volume(Size=100, AvailabilityZone="us-east-1b")
volume2.attach_to_instance(InstanceId=instance["InstanceId"], Device="/dev/sdg")
snapshots = client.create_snapshots(
InstanceSpecification={
"InstanceId": instance["InstanceId"],
"ExcludeBootVolume": True,
}
)["Snapshots"]
# 1 Snapshots ; Only the additional volumes are returned
snapshots.should.have.length_of(2)
snapshot1 = next(s for s in snapshots if s["VolumeId"] == volume1.volume_id)
snapshot1.should.have.key("VolumeSize").equals(80)
snapshot2 = next(s for s in snapshots if s["VolumeId"] == volume2.volume_id)
snapshot2.should.have.key("VolumeSize").equals(100)