diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py index 3f1f2f5b8af4..41c0459a84ec 100644 --- a/moto/ec2/exceptions.py +++ b/moto/ec2/exceptions.py @@ -1,4 +1,5 @@ from moto.core.exceptions import RESTError +from typing import List, Union # EC2 has a custom root-tag - vs @@ -258,7 +259,7 @@ def __init__(self, instance_type, error_type="InvalidInstanceType.NotFound"): class InvalidAMIIdError(EC2ClientError): - def __init__(self, ami_id): + def __init__(self, ami_id: Union[List[str], str]): super().__init__( "InvalidAMIID.NotFound", f"The image id '[{ami_id}]' does not exist", @@ -266,7 +267,7 @@ def __init__(self, ami_id): class UnvailableAMIIdError(EC2ClientError): - def __init__(self, ami_id): + def __init__(self, ami_id: str): super().__init__( "InvalidAMIID.Unavailable", f"The image id '[{ami_id}]' is no longer available", @@ -274,7 +275,7 @@ def __init__(self, ami_id): class InvalidAMIAttributeItemValueError(EC2ClientError): - def __init__(self, attribute, value): + def __init__(self, attribute: str, value: str): super().__init__( "InvalidAMIAttributeItemValue", f'Invalid attribute item value "{value}" for {attribute} item type.', @@ -282,7 +283,7 @@ def __init__(self, attribute, value): class MalformedAMIIdError(EC2ClientError): - def __init__(self, ami_id): + def __init__(self, ami_id: List[str]): super().__init__( "InvalidAMIID.Malformed", f'Invalid id: "{ami_id}" (expecting "ami-...")' ) @@ -694,7 +695,7 @@ def __init__(self, vpc_end_point_id): class InvalidTaggableResourceType(EC2ClientError): - def __init__(self, resource_type): + def __init__(self, resource_type: str): super().__init__( "InvalidParameterValue", f"'{resource_type}' is not a valid taggable resource type for this operation.", diff --git a/moto/ec2/models/amis.py b/moto/ec2/models/amis.py index dffe300a0d52..99d955373e2a 100644 --- a/moto/ec2/models/amis.py +++ b/moto/ec2/models/amis.py @@ -1,6 +1,7 @@ import json import re from os import environ +from typing import Any, Dict, List, Iterable, Optional, Set from moto.utilities.utils import load_resource from ..exceptions import ( InvalidAMIIdError, @@ -10,6 +11,7 @@ UnvailableAMIIdError, ) from .core import TaggedEC2Resource +from .instances import Instance from ..utils import ( random_ami_id, generic_filter, @@ -18,7 +20,7 @@ if "MOTO_AMIS_PATH" in environ: - with open(environ.get("MOTO_AMIS_PATH"), "r", encoding="utf-8") as f: + with open(environ["MOTO_AMIS_PATH"], "r", encoding="utf-8") as f: AMIS = json.load(f) else: AMIS = load_resource(__name__, "../resources/amis.json") @@ -27,28 +29,28 @@ class Ami(TaggedEC2Resource): def __init__( self, - ec2_backend, - ami_id, - instance=None, - source_ami=None, - name=None, - description=None, - owner_id=None, - owner_alias=None, - public=False, - virtualization_type=None, - architecture=None, - state="available", - creation_date=None, - platform=None, - image_type="machine", - image_location=None, - hypervisor=None, - root_device_type="standard", - root_device_name="/dev/sda1", - sriov="simple", - region_name="us-east-1a", - snapshot_description=None, + ec2_backend: Any, + ami_id: str, + instance: Optional[Instance] = None, + source_ami: Optional["Ami"] = None, + name: Optional[str] = None, + description: Optional[str] = None, + owner_id: Optional[str] = None, + owner_alias: Optional[str] = None, + public: bool = False, + virtualization_type: Optional[str] = None, + architecture: Optional[str] = None, + state: str = "available", + creation_date: Optional[str] = None, + platform: Optional[str] = None, + image_type: str = "machine", + image_location: Optional[str] = None, + hypervisor: Optional[str] = None, + root_device_type: str = "standard", + root_device_name: str = "/dev/sda1", + sriov: str = "simple", + region_name: str = "us-east-1a", + snapshot_description: Optional[str] = None, ): self.ec2_backend = ec2_backend self.id = ami_id @@ -92,8 +94,8 @@ def __init__( if not description: self.description = source_ami.description - self.launch_permission_groups = set() - self.launch_permission_users = set() + self.launch_permission_groups: Set[str] = set() + self.launch_permission_users: Set[str] = set() if public: self.launch_permission_groups.add("all") @@ -109,14 +111,16 @@ def __init__( self.ec2_backend.delete_volume(volume.id) @property - def is_public(self): + def is_public(self) -> bool: return "all" in self.launch_permission_groups @property - def is_public_string(self): + def is_public_string(self) -> str: return str(self.is_public).lower() - def get_filter_value(self, filter_name): + def get_filter_value( + self, filter_name: str, method_name: Optional[str] = None + ) -> Any: if filter_name == "virtualization-type": return self.virtualization_type elif filter_name == "kernel-id": @@ -142,12 +146,12 @@ def get_filter_value(self, filter_name): class AmiBackend: AMI_REGEX = re.compile("ami-[a-z0-9]+") - def __init__(self): - self.amis = {} - self.deleted_amis = list() + def __init__(self) -> None: + self.amis: Dict[str, Ami] = {} + self.deleted_amis: List[str] = list() self._load_amis() - def _load_amis(self): + def _load_amis(self) -> None: for ami in AMIS: ami_id = ami["ami_id"] # we are assuming the default loaded amis are owned by amazon @@ -157,7 +161,7 @@ def _load_amis(self): if "MOTO_AMIS_PATH" not in environ: try: latest_amis = load_resource( - __name__, f"../resources/latest_amis/{self.region_name}.json" + __name__, f"../resources/latest_amis/{self.region_name}.json" # type: ignore[attr-defined] ) for ami in latest_amis: ami_id = ami["ami_id"] @@ -169,14 +173,14 @@ def _load_amis(self): def create_image( self, - instance_id, - name=None, - description=None, - tag_specifications=None, - ): + instance_id: str, + name: str, + description: str, + tag_specifications: List[Dict[str, Any]], + ) -> Ami: # TODO: check that instance exists and pull info from it. ami_id = random_ami_id() - instance = self.get_instance(instance_id) + instance = self.get_instance(instance_id) # type: ignore[attr-defined] tags = [] for tag_specification in tag_specifications: resource_type = tag_specification["ResourceType"] @@ -202,12 +206,17 @@ def create_image( self.amis[ami_id] = ami return ami - def copy_image(self, source_image_id, source_region, name=None, description=None): + def copy_image( + self, + source_image_id: str, + source_region: str, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> Ami: from ..models import ec2_backends - source_ami = ec2_backends[self.account_id][source_region].describe_images( - ami_ids=[source_image_id] - )[0] + source_backend = ec2_backends[self.account_id][source_region] # type: ignore[attr-defined] + source_ami = source_backend.describe_images(ami_ids=[source_image_id])[0] ami_id = random_ami_id() ami = Ami( self, @@ -220,10 +229,16 @@ def copy_image(self, source_image_id, source_region, name=None, description=None self.amis[ami_id] = ami return ami - def describe_images(self, ami_ids=(), filters=None, exec_users=None, owners=None): - images = self.amis.copy().values() - - if len(ami_ids): + def describe_images( + self, + ami_ids: Optional[List[str]] = None, + filters: Optional[Dict[str, Any]] = None, + exec_users: Optional[List[str]] = None, + owners: Optional[List[str]] = None, + ) -> List[Ami]: + images = list(self.amis.copy().values()) + + if ami_ids and len(ami_ids): # boto3 seems to default to just searching based on ami ids if that parameter is passed # and if no images are found, it raises an errors # Note that we can search for images that have been previously deleted, without raising any errors @@ -254,7 +269,7 @@ def describe_images(self, ami_ids=(), filters=None, exec_users=None, owners=None # support filtering by Owners=['self'] if "self" in owners: owners = list( - map(lambda o: self.account_id if o == "self" else o, owners) + map(lambda o: self.account_id if o == "self" else o, owners) # type: ignore[attr-defined] ) images = [ ami @@ -268,24 +283,26 @@ def describe_images(self, ami_ids=(), filters=None, exec_users=None, owners=None return images - def deregister_image(self, ami_id): + def deregister_image(self, ami_id: str) -> None: if ami_id in self.amis: self.amis.pop(ami_id) self.deleted_amis.append(ami_id) - return True elif ami_id in self.deleted_amis: raise UnvailableAMIIdError(ami_id) - raise InvalidAMIIdError(ami_id) + else: + raise InvalidAMIIdError(ami_id) - def get_launch_permission_groups(self, ami_id): + def get_launch_permission_groups(self, ami_id: str) -> Iterable[str]: ami = self.describe_images(ami_ids=[ami_id])[0] return ami.launch_permission_groups - def get_launch_permission_users(self, ami_id): + def get_launch_permission_users(self, ami_id: str) -> Iterable[str]: ami = self.describe_images(ami_ids=[ami_id])[0] return ami.launch_permission_users - def validate_permission_targets(self, user_ids=None, group=None): + def validate_permission_targets( + self, user_ids: Optional[List[str]] = None, group: Optional[str] = None + ) -> None: # If anything is invalid, nothing is added. (No partial success.) if user_ids: """ @@ -300,7 +317,12 @@ def validate_permission_targets(self, user_ids=None, group=None): if group and group != "all": raise InvalidAMIAttributeItemValueError("UserGroup", group) - def add_launch_permission(self, ami_id, user_ids=None, group=None): + def add_launch_permission( + self, + ami_id: str, + user_ids: Optional[List[str]] = None, + group: Optional[str] = None, + ) -> None: ami = self.describe_images(ami_ids=[ami_id])[0] self.validate_permission_targets(user_ids=user_ids, group=group) @@ -311,9 +333,9 @@ def add_launch_permission(self, ami_id, user_ids=None, group=None): if group: ami.launch_permission_groups.add(group) - return True - - def register_image(self, name=None, description=None): + def register_image( + self, name: Optional[str] = None, description: Optional[str] = None + ) -> Ami: ami_id = random_ami_id() ami = Ami( self, @@ -326,7 +348,12 @@ def register_image(self, name=None, description=None): self.amis[ami_id] = ami return ami - def remove_launch_permission(self, ami_id, user_ids=None, group=None): + def remove_launch_permission( + self, + ami_id: str, + user_ids: Optional[List[str]] = None, + group: Optional[str] = None, + ) -> None: ami = self.describe_images(ami_ids=[ami_id])[0] self.validate_permission_targets(user_ids=user_ids, group=group) @@ -336,5 +363,3 @@ def remove_launch_permission(self, ami_id, user_ids=None, group=None): if group: ami.launch_permission_groups.discard(group) - - return True diff --git a/moto/ec2/models/availability_zones_and_regions.py b/moto/ec2/models/availability_zones_and_regions.py index 83718a6a4edf..7545cd83c65d 100644 --- a/moto/ec2/models/availability_zones_and_regions.py +++ b/moto/ec2/models/availability_zones_and_regions.py @@ -1,16 +1,23 @@ from boto3 import Session +from typing import Any, Dict, List, Optional from moto.utilities.utils import filter_resources -class Region(object): - def __init__(self, name, endpoint, opt_in_status): +class Region: + def __init__(self, name: str, endpoint: str, opt_in_status: str): self.name = name self.endpoint = endpoint self.opt_in_status = opt_in_status -class Zone(object): - def __init__(self, name, region_name, zone_id, zone_type="availability-zone"): +class Zone: + def __init__( + self, + name: str, + region_name: str, + zone_id: str, + zone_type: str = "availability-zone", + ): self.name = name self.region_name = region_name self.zone_id = zone_id @@ -292,7 +299,9 @@ class RegionsAndZonesBackend: ], } - def describe_regions(self, region_names=None): + def describe_regions( + self, region_names: Optional[List[str]] = None + ) -> List[Region]: if not region_names: return self.regions ret = [] @@ -302,9 +311,11 @@ def describe_regions(self, region_names=None): ret.append(region) return ret - def describe_availability_zones(self, filters=None): + def describe_availability_zones( + self, filters: Optional[List[Dict[str, Any]]] = None + ) -> List[Zone]: # We might not have any zones for the current region, if it was introduced recently - zones = self.zones.get(self.region_name, []) + zones = self.zones.get(self.region_name, []) # type: ignore[attr-defined] attr_pairs = ( ("zone-id", "zone_id"), ("zone-type", "zone_type"), @@ -316,7 +327,8 @@ def describe_availability_zones(self, filters=None): result = filter_resources(zones, filters, attr_pairs) return result - def get_zone_by_name(self, name): + def get_zone_by_name(self, name: str) -> Optional[Zone]: for zone in self.describe_availability_zones(): if zone.name == name: return zone + return None diff --git a/moto/ec2/models/core.py b/moto/ec2/models/core.py index 5a7c6c9f2113..bd60d7ccb71d 100644 --- a/moto/ec2/models/core.py +++ b/moto/ec2/models/core.py @@ -1,4 +1,4 @@ -from typing import Dict, List +from typing import Any, Dict, List, Optional from moto.core import BaseModel from ..exceptions import FilterNotImplementedError @@ -11,14 +11,16 @@ def get_tags(self) -> List[Dict[str, str]]: tags = self.ec2_backend.describe_tags(filters={"resource-id": [self.id]}) return tags - def add_tag(self, key, value): + def add_tag(self, key: str, value: str) -> None: self.ec2_backend.create_tags([self.id], {key: value}) - def add_tags(self, tag_map: Dict[str, str]): + def add_tags(self, tag_map: Dict[str, str]) -> None: for key, value in tag_map.items(): self.ec2_backend.create_tags([self.id], {key: value}) - def get_filter_value(self, filter_name, method_name=None): + def get_filter_value( + self, filter_name: str, method_name: Optional[str] = None + ) -> Any: tags = self.get_tags() if filter_name.startswith("tag:"): diff --git a/moto/ec2/responses/amis.py b/moto/ec2/responses/amis.py index 1bcefa387305..47925ba748bd 100644 --- a/moto/ec2/responses/amis.py +++ b/moto/ec2/responses/amis.py @@ -32,9 +32,9 @@ def copy_image(self): def deregister_image(self): ami_id = self._get_param("ImageId") if self.is_not_dryrun("DeregisterImage"): - success = self.ec2_backend.deregister_image(ami_id) + self.ec2_backend.deregister_image(ami_id) template = self.response_template(DEREGISTER_IMAGE_RESPONSE) - return template.render(success=str(success).lower()) + return template.render(success="true") def describe_images(self): self.error_on_dryrun() diff --git a/moto/ec2/utils.py b/moto/ec2/utils.py index e081708a1e65..32f44418b655 100644 --- a/moto/ec2/utils.py +++ b/moto/ec2/utils.py @@ -7,6 +7,7 @@ from cryptography.hazmat.primitives import serialization from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.asymmetric import rsa +from typing import Any, Dict, List from moto.iam import iam_backends from moto.moto_api._internal import mock_random as random @@ -71,7 +72,7 @@ def random_id(prefix="", size=8): return f"{prefix}-{random_resource_id(size)}" -def random_ami_id(): +def random_ami_id() -> str: return random_id(prefix=EC2_RESOURCE_TO_PREFIX["image"]) @@ -302,7 +303,7 @@ def create_dns_entries(service_name, vpc_endpoint_id): return dns_entries -def utc_date_and_time(): +def utc_date_and_time() -> str: x = datetime.utcnow() # Better performing alternative to x.strftime("%Y-%m-%dT%H:%M:%S.000Z") return f"{x.year}-{x.month:02d}-{x.day:02d}T{x.hour:02d}:{x.minute:02d}:{x.second:02d}.000Z" @@ -518,7 +519,7 @@ def is_filter_matching(obj, _filter, filter_value): return value in filter_value -def generic_filter(filters, objects): +def generic_filter(filters: Dict[str, Any], objects: List[Any]) -> List[Any]: if filters: for (_filter, _filter_value) in filters.items(): objects = [ diff --git a/moto/utilities/utils.py b/moto/utilities/utils.py index 7ff515706dbe..01e98499f5bb 100644 --- a/moto/utilities/utils.py +++ b/moto/utilities/utils.py @@ -3,7 +3,7 @@ import pkgutil from collections.abc import MutableMapping -from typing import Any, Dict +from typing import Any, Dict, List, TypeVar def str2bool(v): @@ -35,7 +35,12 @@ def merge_multiple_dicts(*args: Any) -> Dict[str, any]: return result -def filter_resources(resources, filters, attr_pairs): +RESOURCE_TYPE = TypeVar("RESOURCE_TYPE") + + +def filter_resources( + resources: List[RESOURCE_TYPE], filters: Any, attr_pairs: Any +) -> List[RESOURCE_TYPE]: """ Used to filter resources. Usually in get and describe apis. """ diff --git a/setup.cfg b/setup.cfg index 8f54e4ae531b..d8cc6bae631f 100644 --- a/setup.cfg +++ b/setup.cfg @@ -229,7 +229,7 @@ disable = W,C,R,E enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import [mypy] -files= moto/a*,moto/b*,moto/c*,moto/d*,moto/moto_api +files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2/models/a*,moto/moto_api show_column_numbers=True show_error_codes = True disable_error_code=abstract