EC2: integrate AMI responses with core serializer (#9108)

Author: bpandola

moto/ec2/models/amis.py

@@ -7,6 +7,7 @@
 from moto import settings
 from moto.utilities.utils import load_resource
 
+from ...core.utils import utcnow
 from ..exceptions import (
     InvalidAMIAttributeItemValueError,
     InvalidAMIIdError,
@@ -17,7 +18,6 @@
 from ..utils import (
     generic_filter,
     random_ami_id,
-    utc_date_and_time,
 )
 from .core import TaggedEC2Resource
 from .instances import Instance
@@ -75,7 +75,7 @@ def __init__(
         self.root_device_name = root_device_name
         self.root_device_type = root_device_type
         self.sriov = sriov
-        self.creation_date = creation_date or utc_date_and_time()
+        self.creation_date = creation_date or utcnow()
         self.product_codes = product_codes
         self.boot_mode = boot_mode
         self.instance_id: Optional[str] = None
@@ -126,8 +126,18 @@ def is_public(self) -> bool:
         return {"Group": "all"} in self.launch_permissions
 
     @property
-    def is_public_string(self) -> str:
-        return str(self.is_public).lower()
+    def block_device_mappings(self) -> List[Dict[str, Any]]:
+        return [
+            {
+                "DeviceName": self.root_device_name,
+                "Ebs": {
+                    "SnapshotId": self.ebs_snapshot.id,
+                    "VolumeSize": 15,
+                    "DeleteOnTermination": False,
+                    "VolumeType": "standard",
+                },
+            }
+        ]
 
     @property
     def source_instance_id(self) -> Optional[str]:
@@ -145,7 +155,7 @@ def get_filter_value(
         elif filter_name == "image-id":
             return self.id
         elif filter_name == "is-public":
-            return self.is_public_string
+            return self.is_public
         elif filter_name == "state":
             return self.state
         elif filter_name == "name":

moto/ec2/responses/amis.py

@@ -1,9 +1,11 @@
+from moto.core.responses import ActionResult, EmptyResult
+
 from ..exceptions import AuthFailureRestricted, InvalidRequest
 from ._base_response import EC2BaseResponse
 
 
 class AmisResponse(EC2BaseResponse):
-    def create_image(self) -> str:
+    def create_image(self) -> ActionResult:
         name = self.querystring.get("Name")[0]  # type: ignore[index]
         description = self._get_param("Description", if_none="")
         instance_id = self._get_param("InstanceId")
@@ -17,10 +19,10 @@ def create_image(self) -> str:
             description,
             tag_specifications=tag_specifications,
         )
-        template = self.response_template(CREATE_IMAGE_RESPONSE)
-        return template.render(image=image)
+        result = {"ImageId": image.id}
+        return ActionResult(result)
 
-    def copy_image(self) -> str:
+    def copy_image(self) -> ActionResult:
         source_image_id = self._get_param("SourceImageId")
         source_region = self._get_param("SourceRegion")
         name = self._get_param("Name")
@@ -31,19 +33,19 @@ def copy_image(self) -> str:
         image = self.ec2_backend.copy_image(
             source_image_id, source_region, name, description
         )
-        template = self.response_template(COPY_IMAGE_RESPONSE)
-        return template.render(image=image)
+        result = {"ImageId": image.id}
+        return ActionResult(result)
 
-    def deregister_image(self) -> str:
+    def deregister_image(self) -> ActionResult:
         ami_id = self._get_param("ImageId")
 
         self.error_on_dryrun()
 
         self.ec2_backend.deregister_image(ami_id)
-        template = self.response_template(DEREGISTER_IMAGE_RESPONSE)
-        return template.render(success="true")
+        result = {"Return": True}
+        return ActionResult(result)
 
-    def describe_images(self) -> str:
+    def describe_images(self) -> ActionResult:
         self.error_on_dryrun()
         ami_ids = self._get_multi_param("ImageId")
         filters = self._filters_from_querystring()
@@ -52,10 +54,10 @@ def describe_images(self) -> str:
         images = self.ec2_backend.describe_images(
             ami_ids=ami_ids, filters=filters, exec_users=exec_users, owners=owners
         )
-        template = self.response_template(DESCRIBE_IMAGES_RESPONSE)
-        return template.render(images=images)
+        result = {"Images": images}
+        return ActionResult(result)
 
-    def describe_image_attribute(self) -> str:
+    def describe_image_attribute(self) -> ActionResult:
         ami_id = self._get_param("ImageId")
         attribute_name = self._get_param("Attribute")
 
@@ -93,15 +95,20 @@ def describe_image_attribute(self) -> str:
                 ami_id, valid_attributes_list[attribute_name]
             )
 
-        template = self.response_template(DESCRIBE_IMAGE_ATTRIBUTES_RESPONSE)
-        return template.render(
-            ami_id=ami_id,
-            launch_permissions=launch_permissions,
-            attribute_name=attribute_name,
-            attribute_value=attribute_value,
-        )
+        result = {"ImageId": ami_id}
+        if attribute_name == "productCodes":
+            result["ProductCodes"] = [
+                {"ProductCodeId": code, "ProductCodeType": "marketplace"}
+                for code in attribute_value  # type: ignore[union-attr]
+            ]
+        elif attribute_name == "launchPermission":
+            result["LaunchPermissions"] = launch_permissions
+        else:
+            result[attribute_name] = {"Value": attribute_value}
 
-    def modify_image_attribute(self) -> str:
+        return ActionResult(result)
+
+    def modify_image_attribute(self) -> ActionResult:
         ami_id = self._get_param("ImageId")
         launch_permissions_to_add = list(
             self._get_params().get("LaunchPermission", {}).get("Add", {}).values()
@@ -139,141 +146,19 @@ def modify_image_attribute(self) -> str:
             launch_permissions_to_add=launch_permissions_to_add,
             launch_permissions_to_remove=launch_permissions_to_remove,
         )
-        return MODIFY_IMAGE_ATTRIBUTE_RESPONSE
+        return EmptyResult()
 
-    def register_image(self) -> str:
+    def register_image(self) -> ActionResult:
         name = self.querystring.get("Name")[0]  # type: ignore[index]
         description = self._get_param("Description", if_none="")
 
         self.error_on_dryrun()
 
         image = self.ec2_backend.register_image(name, description)
-        template = self.response_template(REGISTER_IMAGE_RESPONSE)
-        return template.render(image=image)
+        result = {"ImageId": image.id}
+        return ActionResult(result)
 
     def reset_image_attribute(self) -> str:
         self.error_on_dryrun()
 
         raise NotImplementedError("AMIs.reset_image_attribute is not yet implemented")
-
-
-CREATE_IMAGE_RESPONSE = """<CreateImageResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
-   <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
-   <imageId>{{ image.id }}</imageId>
-</CreateImageResponse>"""
-
-COPY_IMAGE_RESPONSE = """<CopyImageResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
-   <requestId>60bc441d-fa2c-494d-b155-5d6a3EXAMPLE</requestId>
-   <imageId>{{ image.id }}</imageId>
-</CopyImageResponse>"""
-
-# TODO almost all of these params should actually be templated based on
-# the ec2 image
-DESCRIBE_IMAGES_RESPONSE = """<DescribeImagesResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
-  <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
-  <imagesSet>
-    {% for image in images %}
-        <item>
-          <imageId>{{ image.id }}</imageId>
-          <imageLocation>{{ image.image_location }}</imageLocation>
-          <imageState>{{ image.state }}</imageState>
-          <imageOwnerId>{{ image.owner_id }}</imageOwnerId>
-          <isPublic>{{ image.is_public_string }}</isPublic>
-          <architecture>{{ image.architecture }}</architecture>
-          <imageType>{{ image.image_type }}</imageType>
-          <kernelId>{{ image.kernel_id }}</kernelId>
-          <ramdiskId>ari-1a2b3c4d</ramdiskId>
-          <imageOwnerAlias>amazon</imageOwnerAlias>
-          <creationDate>{{ image.creation_date }}</creationDate>
-          <name>{{ image.name }}</name>
-          {% if image.platform %}
-             <platform>{{ image.platform }}</platform>
-          {% endif %}
-          <description>{{ image.description }}</description>
-          <rootDeviceType>{{ image.root_device_type }}</rootDeviceType>
-          <rootDeviceName>{{ image.root_device_name }}</rootDeviceName>
-          <blockDeviceMapping>
-            <item>
-              <deviceName>{{ image.root_device_name }}</deviceName>
-              <ebs>
-                <snapshotId>{{ image.ebs_snapshot.id }}</snapshotId>
-                <volumeSize>15</volumeSize>
-                <deleteOnTermination>false</deleteOnTermination>
-                <volumeType>standard</volumeType>
-              </ebs>
-            </item>
-          </blockDeviceMapping>
-          <virtualizationType>{{ image.virtualization_type }}</virtualizationType>
-          <tagSet>
-            {% for tag in image.get_tags() %}
-              <item>
-                <resourceId>{{ tag.resource_id }}</resourceId>
-                <resourceType>{{ tag.resource_type }}</resourceType>
-                <key>{{ tag.key }}</key>
-                <value>{{ tag.value }}</value>
-              </item>
-            {% endfor %}
-          </tagSet>
-          <hypervisor>xen</hypervisor>
-          {% if image.source_instance_id %}
-          <sourceInstanceId>{{ image.source_instance_id }}</sourceInstanceId>
-          {% endif %}
-        </item>
-    {% endfor %}
-  </imagesSet>
-</DescribeImagesResponse>"""
-
-DESCRIBE_IMAGE_RESPONSE = """<DescribeImageAttributeResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
-   <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
-   <imageId>{{ image.id }}</imageId>
-   <{{ key }}>
-     <value>{{ value }}</value>
-   </{{key }}>
-</DescribeImageAttributeResponse>"""
-
-DEREGISTER_IMAGE_RESPONSE = """<DeregisterImageResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
-  <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
-  <return>{{ success }}</return>
-</DeregisterImageResponse>"""
-
-DESCRIBE_IMAGE_ATTRIBUTES_RESPONSE = """
-<DescribeImageAttributeResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
-   <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
-   <imageId>{{ ami_id }}</imageId>
-    <{{ attribute_name }}>
-    {% if attribute_name == 'productCodes' %}
-        {% for value in attribute_value %}
-        <item>
-            <productCode>{{ value }}</productCode>
-            <type>marketplace</type>
-        </item>
-        {% endfor %}
-    {% endif %}
-    {% if attribute_name == 'launchPermission' %}
-         {% if launch_permissions %}
-            {% for lp in launch_permissions %}
-               <item>
-                  {% if lp['UserId'] %}<userId>{{ lp['UserId'] }}</userId>{% endif %}
-                  {% if lp['Group'] %}<group>{{ lp['Group'] }}</group>{% endif %}
-                  {% if lp['OrganizationArn'] %}<organizationArn>{{ lp['OrganizationArn'] }}</organizationArn>{% endif %}
-                  {% if lp['OrganizationalUnitArn'] %}<organizationalUnitArn>{{ lp['OrganizationalUnitArn'] }}</organizationalUnitArn>{% endif %}
-               </item>
-            {% endfor %}
-         {% endif %}
-    {% endif %}
-    {% if attribute_name not in ['launchPermission', 'productCodes'] %}
-            <value>{{ attribute_value }}</value>
-    {% endif %}
-    </{{ attribute_name }}>
-</DescribeImageAttributeResponse>"""
-
-MODIFY_IMAGE_ATTRIBUTE_RESPONSE = """
-<ModifyImageAttributeResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
-   <return>true</return>
-</ModifyImageAttributeResponse>
-"""
-
-REGISTER_IMAGE_RESPONSE = """<RegisterImageResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
-   <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
-   <imageId>{{ image.id }}</imageId>
-</RegisterImageResponse>"""

tests/test_ec2/test_amis.py

@@ -71,7 +71,6 @@ def test_ami_create_and_delete():
     assert retrieved_image["ImageId"] == image_id
     assert retrieved_image["VirtualizationType"] == instance["VirtualizationType"]
     assert retrieved_image["Architecture"] == instance["Architecture"]
-    assert retrieved_image["KernelId"] == instance["KernelId"]
     assert retrieved_image["Platform"] == instance["Platform"]
     assert "CreationDate" in retrieved_image
     ec2.terminate_instances(InstanceIds=[instance_id])
@@ -224,7 +223,6 @@ def test_ami_copy():
     assert copy_image["ImageId"] == copy_image_id
     assert copy_image["VirtualizationType"] == source_image["VirtualizationType"]
     assert copy_image["Architecture"] == source_image["Architecture"]
-    assert copy_image["KernelId"] == source_image["KernelId"]
     assert copy_image["Platform"] == source_image["Platform"]
 
     # Validate auto-created snapshot