From bd96e42915772c66c744b1a10fe4b6b957f85fbf Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Mon, 6 Feb 2023 12:20:03 -0100 Subject: [PATCH] Techdebt: MyPy EC2 (k-l-m-models) (#5905) --- moto/ec2/exceptions.py | 14 ++-- moto/ec2/models/key_pairs.py | 29 +++---- moto/ec2/models/launch_templates.py | 118 ++++++++++++++++++---------- moto/ec2/models/managed_prefixes.py | 87 ++++++++++---------- moto/ec2/responses/key_pairs.py | 8 +- moto/ec2/utils.py | 16 ++-- setup.cfg | 2 +- 7 files changed, 158 insertions(+), 116 deletions(-) diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py index 6d80940ea..31560d69c 100644 --- a/moto/ec2/exceptions.py +++ b/moto/ec2/exceptions.py @@ -43,7 +43,7 @@ class DependencyViolationError(EC2ClientError): class MissingParameterError(EC2ClientError): - def __init__(self, parameter): + def __init__(self, parameter: str): super().__init__( "MissingParameter", f"The request must contain the parameter {parameter}", @@ -77,21 +77,21 @@ class MalformedDHCPOptionsIdError(EC2ClientError): class InvalidKeyPairNameError(EC2ClientError): - def __init__(self, key): + def __init__(self, key: Iterable[str]): super().__init__( "InvalidKeyPair.NotFound", f"The keypair '{key}' does not exist." ) class InvalidKeyPairDuplicateError(EC2ClientError): - def __init__(self, key): + def __init__(self, key: str): super().__init__( "InvalidKeyPair.Duplicate", f"The keypair '{key}' already exists." ) class InvalidKeyPairFormatError(EC2ClientError): - def __init__(self): + def __init__(self) -> None: super().__init__( "InvalidKeyPair.Format", "Key is not in valid OpenSSH public key format" ) @@ -640,7 +640,7 @@ class OperationNotPermitted4(EC2ClientError): class InvalidLaunchTemplateNameAlreadyExistsError(EC2ClientError): - def __init__(self): + def __init__(self) -> None: super().__init__( "InvalidLaunchTemplateName.AlreadyExistsException", "Launch template name already in use.", @@ -648,7 +648,7 @@ class InvalidLaunchTemplateNameAlreadyExistsError(EC2ClientError): class InvalidLaunchTemplateNameNotFoundError(EC2ClientError): - def __init__(self): + def __init__(self) -> None: super().__init__( "InvalidLaunchTemplateName.NotFoundException", "At least one of the launch templates specified in the request does not exist.", @@ -656,7 +656,7 @@ class InvalidLaunchTemplateNameNotFoundError(EC2ClientError): class InvalidLaunchTemplateNameNotFoundWithNameError(EC2ClientError): - def __init__(self, name): + def __init__(self, name: str): super().__init__( "InvalidLaunchTemplateName.NotFoundException", f"The specified launch template, with template name {name}, does not exist", diff --git a/moto/ec2/models/key_pairs.py b/moto/ec2/models/key_pairs.py index 99a46b456..b2dab1235 100644 --- a/moto/ec2/models/key_pairs.py +++ b/moto/ec2/models/key_pairs.py @@ -1,3 +1,5 @@ +from typing import Any, Dict, List + from moto.core import BaseModel from ..exceptions import ( FilterNotImplementedError, @@ -15,13 +17,13 @@ from ..utils import ( class KeyPair(BaseModel): - def __init__(self, name, fingerprint, material): + def __init__(self, name: str, fingerprint: str, material: str): self.id = random_key_pair_id() self.name = name self.fingerprint = fingerprint self.material = material - def get_filter_value(self, filter_name): + def get_filter_value(self, filter_name: str) -> str: if filter_name == "key-name": return self.name elif filter_name == "fingerprint": @@ -31,23 +33,22 @@ class KeyPair(BaseModel): class KeyPairBackend: - def __init__(self): - self.keypairs = {} + def __init__(self) -> None: + self.keypairs: Dict[str, KeyPair] = {} - def create_key_pair(self, name): + def create_key_pair(self, name: str) -> KeyPair: if name in self.keypairs: raise InvalidKeyPairDuplicateError(name) keypair = KeyPair(name, **random_key_pair()) self.keypairs[name] = keypair return keypair - def delete_key_pair(self, name): - if name in self.keypairs: - self.keypairs.pop(name) - return True + def delete_key_pair(self, name: str) -> None: + self.keypairs.pop(name, None) - def describe_key_pairs(self, key_names=None, filters=None): - results = [] + def describe_key_pairs( + self, key_names: List[str], filters: Any = None + ) -> List[KeyPair]: if any(key_names): results = [ keypair @@ -55,17 +56,17 @@ class KeyPairBackend: if keypair.name in key_names ] if len(key_names) > len(results): - unknown_keys = set(key_names) - set(results) + unknown_keys = set(key_names) - set(results) # type: ignore raise InvalidKeyPairNameError(unknown_keys) else: - results = self.keypairs.values() + results = list(self.keypairs.values()) if filters: return generic_filter(filters, results) else: return results - def import_key_pair(self, key_name, public_key_material): + def import_key_pair(self, key_name: str, public_key_material: str) -> KeyPair: if key_name in self.keypairs: raise InvalidKeyPairDuplicateError(key_name) diff --git a/moto/ec2/models/launch_templates.py b/moto/ec2/models/launch_templates.py index 488813043..8ce6909ff 100644 --- a/moto/ec2/models/launch_templates.py +++ b/moto/ec2/models/launch_templates.py @@ -1,4 +1,5 @@ from collections import OrderedDict +from typing import Any, Dict, List, Optional from moto.core import CloudFormationModel from .core import TaggedEC2Resource @@ -16,8 +17,14 @@ from ..exceptions import ( ) -class LaunchTemplateVersion(object): - def __init__(self, template, number, data, description): +class LaunchTemplateVersion: + def __init__( + self, + template: "LaunchTemplate", + number: int, + data: Dict[str, Any], + description: str, + ): self.template = template self.number = number self.data = data @@ -25,85 +32,101 @@ class LaunchTemplateVersion(object): self.create_time = utc_date_and_time() @property - def image_id(self): + def image_id(self) -> str: return self.data.get("ImageId", "") @property - def instance_type(self): + def instance_type(self) -> str: return self.data.get("InstanceType", "") @property - def security_groups(self): + def security_groups(self) -> List[str]: return self.data.get("SecurityGroups", []) @property - def user_data(self): + def user_data(self) -> str: return self.data.get("UserData", "") class LaunchTemplate(TaggedEC2Resource, CloudFormationModel): - def __init__(self, backend, name, template_data, version_description, tag_spec): + def __init__( + self, + backend: Any, + name: str, + template_data: Dict[str, Any], + version_description: str, + tag_spec: Dict[str, Dict[str, str]], + ): self.ec2_backend = backend self.name = name self.id = random_launch_template_id() self.create_time = utc_date_and_time() - tag_map = tag_spec.get("launch-template", {}) + tag_map: Dict[str, str] = tag_spec.get("launch-template", {}) self.add_tags(tag_map) self.tags = self.get_tags() - self.versions = [] + self.versions: List[LaunchTemplateVersion] = [] self.create_version(template_data, version_description) self.default_version_number = 1 - def create_version(self, data, description): + def create_version( + self, data: Dict[str, Any], description: str + ) -> LaunchTemplateVersion: num = len(self.versions) + 1 version = LaunchTemplateVersion(self, num, data, description) self.versions.append(version) return version - def is_default(self, version): - return self.default_version == version.number + def is_default(self, version: LaunchTemplateVersion) -> bool: + return self.default_version == version.number # type: ignore - def get_version(self, num): + def get_version(self, num: Any) -> LaunchTemplateVersion: if str(num).lower() == "$latest": return self.versions[-1] if str(num).lower() == "$default": return self.default_version() return self.versions[int(num) - 1] - def default_version(self): + def default_version(self) -> LaunchTemplateVersion: return self.versions[self.default_version_number - 1] - def latest_version(self): + def latest_version(self) -> LaunchTemplateVersion: return self.versions[-1] @property - def latest_version_number(self): + def latest_version_number(self) -> int: return self.latest_version().number @property - def physical_resource_id(self): + def physical_resource_id(self) -> str: return self.id - def get_filter_value(self, filter_name): + def get_filter_value( + self, filter_name: str, method_name: Optional[str] = None + ) -> Any: if filter_name == "launch-template-name": return self.name else: return super().get_filter_value(filter_name, "DescribeLaunchTemplates") @staticmethod - def cloudformation_name_type(): + def cloudformation_name_type() -> str: return "LaunchTemplateName" @staticmethod - def cloudformation_type(): + def cloudformation_type() -> str: # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-ec2-launchtemplate.html return "AWS::EC2::LaunchTemplate" @classmethod - def create_from_cloudformation_json( - cls, resource_name, cloudformation_json, account_id, region_name, **kwargs - ): + def create_from_cloudformation_json( # type: ignore[misc] + cls, + resource_name: str, + cloudformation_json: Any, + account_id: str, + region_name: str, + **kwargs: Any + ) -> "LaunchTemplate": from ..models import ec2_backends @@ -124,14 +147,14 @@ class LaunchTemplate(TaggedEC2Resource, CloudFormationModel): return launch_template @classmethod - def update_from_cloudformation_json( + def update_from_cloudformation_json( # type: ignore[misc] cls, - original_resource, - new_resource_name, - cloudformation_json, - account_id, - region_name, - ): + original_resource: Any, + new_resource_name: str, + cloudformation_json: Any, + account_id: str, + region_name: str, + ) -> "LaunchTemplate": from ..models import ec2_backends @@ -150,9 +173,13 @@ class LaunchTemplate(TaggedEC2Resource, CloudFormationModel): return launch_template @classmethod - def delete_from_cloudformation_json( - cls, resource_name, cloudformation_json, account_id, region_name - ): + def delete_from_cloudformation_json( # type: ignore[misc] + cls, + resource_name: str, + cloudformation_json: Any, + account_id: str, + region_name: str, + ) -> None: from ..models import ec2_backends @@ -166,12 +193,18 @@ class LaunchTemplate(TaggedEC2Resource, CloudFormationModel): class LaunchTemplateBackend: - def __init__(self): - self.launch_template_name_to_ids = {} - self.launch_templates = OrderedDict() - self.launch_template_insert_order = [] + def __init__(self) -> None: + self.launch_template_name_to_ids: Dict[str, str] = {} + self.launch_templates: Dict[str, LaunchTemplate] = OrderedDict() + self.launch_template_insert_order: List[str] = [] - def create_launch_template(self, name, description, template_data, tag_spec): + def create_launch_template( + self, + name: str, + description: str, + template_data: Dict[str, Any], + tag_spec: Dict[str, Any], + ) -> LaunchTemplate: if name in self.launch_template_name_to_ids: raise InvalidLaunchTemplateNameAlreadyExistsError() template = LaunchTemplate(self, name, template_data, description, tag_spec) @@ -188,7 +221,7 @@ class LaunchTemplateBackend: raise InvalidLaunchTemplateNameNotFoundWithNameError(name) return self.get_launch_template(self.launch_template_name_to_ids[name]) - def delete_launch_template(self, name, tid): + def delete_launch_template(self, name: str, tid: Optional[str]) -> LaunchTemplate: if name: tid = self.launch_template_name_to_ids.get(name) if tid is None: @@ -200,8 +233,11 @@ class LaunchTemplateBackend: return template def describe_launch_templates( - self, template_names=None, template_ids=None, filters=None - ): + self, + template_names: Optional[List[str]] = None, + template_ids: Optional[List[str]] = None, + filters: Any = None, + ) -> List[LaunchTemplate]: if template_names and not template_ids: template_ids = [] for name in template_names: diff --git a/moto/ec2/models/managed_prefixes.py b/moto/ec2/models/managed_prefixes.py index fa3e4ead9..eb2f90898 100644 --- a/moto/ec2/models/managed_prefixes.py +++ b/moto/ec2/models/managed_prefixes.py @@ -1,3 +1,5 @@ +from typing import Any, Dict, List, Optional + from moto.utilities.utils import filter_resources from .core import TaggedEC2Resource from ..utils import random_managed_prefix_list_id, describe_tag_filter @@ -6,14 +8,14 @@ from ..utils import random_managed_prefix_list_id, describe_tag_filter class ManagedPrefixList(TaggedEC2Resource): def __init__( self, - backend, - address_family=None, - entry=None, - max_entries=None, - prefix_list_name=None, - region=None, - tags=None, - owner_id=None, + backend: Any, + region: str, + address_family: Optional[str] = None, + entry: Optional[List[Dict[str, str]]] = None, + max_entries: Optional[str] = None, + prefix_list_name: Optional[str] = None, + tags: Optional[Dict[str, str]] = None, + owner_id: Optional[str] = None, ): self.ec2_backend = backend self.address_family = address_family @@ -23,49 +25,51 @@ class ManagedPrefixList(TaggedEC2Resource): self.state = "create-complete" self.state_message = "create complete" self.add_tags(tags or {}) - self.version = 1 + self.version: Optional[int] = 1 self.entries = {self.version: entry} if entry else {} self.resource_owner_id = owner_id if owner_id else None self.prefix_list_arn = self.arn(region, self.owner_id) self.delete_counter = 1 - def arn(self, region, owner_id): + def arn(self, region: str, owner_id: str) -> str: return f"arn:aws:ec2:{region}:{owner_id}:prefix-list/{self.id}" @property - def owner_id(self): + def owner_id(self) -> str: return self.resource_owner_id or self.ec2_backend.account_id class ManagedPrefixListBackend: - def __init__(self): - self.managed_prefix_lists = {} + def __init__(self) -> None: + self.managed_prefix_lists: Dict[str, ManagedPrefixList] = {} self.create_default_pls() def create_managed_prefix_list( self, - address_family=None, - entry=None, - max_entries=None, - prefix_list_name=None, - tags=None, - owner_id=None, - ): + address_family: Optional[str] = None, + entry: Optional[List[Dict[str, str]]] = None, + max_entries: Optional[str] = None, + prefix_list_name: Optional[str] = None, + tags: Optional[Dict[str, str]] = None, + owner_id: Optional[str] = None, + ) -> ManagedPrefixList: managed_prefix_list = ManagedPrefixList( self, address_family=address_family, entry=entry, max_entries=max_entries, prefix_list_name=prefix_list_name, - region=self.region_name, + region=self.region_name, # type: ignore[attr-defined] tags=tags, owner_id=owner_id, ) self.managed_prefix_lists[managed_prefix_list.id] = managed_prefix_list return managed_prefix_list - def describe_managed_prefix_lists(self, prefix_list_ids=None, filters=None): - managed_prefix_lists = list(self.managed_prefix_lists.copy().values()) + def describe_managed_prefix_lists( + self, prefix_list_ids: Optional[List[str]] = None, filters: Any = None + ) -> List[ManagedPrefixList]: + managed_prefix_lists = list(self.managed_prefix_lists.values()) attr_pairs = ( ("owner-id", "owner_id"), ("prefix-list-id", "id"), @@ -92,29 +96,30 @@ class ManagedPrefixListBackend: item.delete_counter -= 1 return result - def get_managed_prefix_list_entries(self, prefix_list_id=None): - managed_prefix_list = self.managed_prefix_lists.get(prefix_list_id) - return managed_prefix_list + def get_managed_prefix_list_entries( + self, prefix_list_id: str + ) -> Optional[ManagedPrefixList]: + return self.managed_prefix_lists.get(prefix_list_id) - def delete_managed_prefix_list(self, prefix_list_id): - managed_prefix_list = self.managed_prefix_lists.get(prefix_list_id) + def delete_managed_prefix_list(self, prefix_list_id: str) -> ManagedPrefixList: + managed_prefix_list: ManagedPrefixList = self.managed_prefix_lists.get(prefix_list_id) # type: ignore managed_prefix_list.state = "delete-complete" return managed_prefix_list def modify_managed_prefix_list( self, - add_entry=None, - prefix_list_id=None, - current_version=None, - prefix_list_name=None, - remove_entry=None, - ): - managed_pl = self.managed_prefix_lists.get(prefix_list_id) + add_entry: List[Dict[str, str]], + remove_entry: List[Dict[str, str]], + prefix_list_id: Optional[str] = None, + current_version: Optional[str] = None, + prefix_list_name: Optional[str] = None, + ) -> ManagedPrefixList: + managed_pl: ManagedPrefixList = self.managed_prefix_lists.get(prefix_list_id) # type: ignore managed_pl.prefix_list_name = prefix_list_name if remove_entry or add_entry: - latest_version = managed_pl.entries.get(managed_pl.version) + latest_version = managed_pl.entries.get(managed_pl.version) # type: ignore[arg-type] entries = ( - managed_pl.entries.get(current_version, latest_version).copy() + managed_pl.entries.get(current_version, latest_version).copy() # type: ignore if managed_pl.entries else [] ) @@ -125,12 +130,12 @@ class ManagedPrefixListBackend: for item in add_entry: if item not in entries.copy(): entries.append(item) - managed_pl.version += 1 + managed_pl.version += 1 # type: ignore[operator] managed_pl.entries[managed_pl.version] = entries managed_pl.state = "modify-complete" return managed_pl - def create_default_pls(self): + def create_default_pls(self) -> None: entry = [ {"Cidr": "52.216.0.0/15", "Description": "default"}, {"Cidr": "3.5.0.0/19", "Description": "default"}, @@ -140,7 +145,7 @@ class ManagedPrefixListBackend: managed_prefix_list = self.create_managed_prefix_list( address_family="IPv4", entry=entry, - prefix_list_name=f"com.amazonaws.{self.region_name}.s3", + prefix_list_name=f"com.amazonaws.{self.region_name}.s3", # type: ignore[attr-defined] owner_id="aws", ) managed_prefix_list.version = None @@ -157,7 +162,7 @@ class ManagedPrefixListBackend: managed_prefix_list = self.create_managed_prefix_list( address_family="IPv4", entry=entry, - prefix_list_name=f"com.amazonaws.{self.region_name}.dynamodb", + prefix_list_name=f"com.amazonaws.{self.region_name}.dynamodb", # type: ignore[attr-defined] owner_id="aws", ) managed_prefix_list.version = None diff --git a/moto/ec2/responses/key_pairs.py b/moto/ec2/responses/key_pairs.py index 041b3aaa0..bb5a57cfd 100644 --- a/moto/ec2/responses/key_pairs.py +++ b/moto/ec2/responses/key_pairs.py @@ -12,10 +12,8 @@ class KeyPairs(EC2BaseResponse): def delete_key_pair(self): name = self._get_param("KeyName") if self.is_not_dryrun("DeleteKeyPair"): - success = str(self.ec2_backend.delete_key_pair(name)).lower() - return self.response_template(DELETE_KEY_PAIR_RESPONSE).render( - success=success - ) + self.ec2_backend.delete_key_pair(name) + return self.response_template(DELETE_KEY_PAIR_RESPONSE).render() def describe_key_pairs(self): names = self._get_multi_param("KeyName") @@ -57,7 +55,7 @@ CREATE_KEY_PAIR_RESPONSE = """ 59dbff89-35bd-4eac-99ed-be587EXAMPLE - {{ success }} + true """ IMPORT_KEYPAIR_RESPONSE = """ diff --git a/moto/ec2/utils.py b/moto/ec2/utils.py index bc6e542b4..8c46e5e63 100644 --- a/moto/ec2/utils.py +++ b/moto/ec2/utils.py @@ -150,7 +150,7 @@ def random_volume_id() -> str: return random_id(prefix=EC2_RESOURCE_TO_PREFIX["volume"]) -def random_key_pair_id(): +def random_key_pair_id() -> str: return random_id(prefix=EC2_RESOURCE_TO_PREFIX["key-pair"]) @@ -224,7 +224,7 @@ def random_transit_gateway_attachment_id(): ) -def random_launch_template_id(): +def random_launch_template_id() -> str: return random_id(prefix=EC2_RESOURCE_TO_PREFIX["launch-template"], size=17) @@ -290,7 +290,7 @@ def generate_route_id( return f"{route_table_id}~{cidr_block}" -def random_managed_prefix_list_id(): +def random_managed_prefix_list_id() -> str: return random_id(prefix=EC2_RESOURCE_TO_PREFIX["managed-prefix-list"], size=8) @@ -547,7 +547,7 @@ def simple_aws_filter_to_re(filter_string): return tmp_filter -def random_key_pair(): +def random_key_pair() -> Dict[str, str]: private_key = rsa.generate_private_key( public_exponent=65537, key_size=2048, backend=default_backend() ) @@ -639,7 +639,7 @@ def generate_instance_identity_document(instance): return document -def rsa_public_key_parse(key_material): +def rsa_public_key_parse(key_material: Any) -> Any: # These imports take ~.5s; let's keep them local import sshpubkeys.exceptions from sshpubkeys.keys import SSHKey @@ -659,7 +659,7 @@ def rsa_public_key_parse(key_material): return public_key.rsa -def rsa_public_key_fingerprint(rsa_public_key): +def rsa_public_key_fingerprint(rsa_public_key: Any) -> str: key_data = rsa_public_key.public_bytes( encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo, @@ -719,7 +719,9 @@ def filter_iam_instance_profiles( return instance_profile -def describe_tag_filter(filters, instances): +def describe_tag_filter( + filters: Any, instances: List[FILTER_TYPE] +) -> List[FILTER_TYPE]: result = instances.copy() for instance in instances: for key in filters: diff --git a/setup.cfg b/setup.cfg index 041342997..ced7f63a3 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/ebs/,moto/ec2/models/a*,moto/ec2/models/c*,moto/ec2/models/d*,moto/ec2/models/e*,moto/ec2/models/f*,moto/ec2/models/h*,moto/ec2/models/i*,moto/moto_api +files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2/models/a*,moto/ec2/models/c*,moto/ec2/models/d*,moto/ec2/models/e*,moto/ec2/models/f*,moto/ec2/models/h*,moto/ec2/models/i*,moto/ec2/models/k*,moto/ec2/models/l*,moto/ec2/models/m*,moto/moto_api show_column_numbers=True show_error_codes = True disable_error_code=abstract