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

Techdebt: MyPy EC2 (a-models) (#5888)

Browse Source
This commit is contained in:
Bert Blommers 2023-02-01 10:27:13 -01:00 committed by GitHub
parent 91157895d4
commit 67ecc3b1db
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 131 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
moto/ec2/exceptions.py
View File

@ -1,4 +1,5 @@
from moto.core.exceptions import RESTError
from typing import List, Union
# EC2 has a custom root-tag - <Response> vs <ErrorResponse>
@ -258,7 +259,7 @@ class InvalidInstanceTypeError(EC2ClientError):
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 @@ class InvalidAMIIdError(EC2ClientError):
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 @@ class UnvailableAMIIdError(EC2ClientError):
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 @@ class InvalidAMIAttributeItemValueError(EC2ClientError):
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 @@ class InvalidVpcEndPointIdError(EC2ClientError):
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.",

143
moto/ec2/models/amis.py
View File

@ -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 @@ from ..exceptions import (
UnvailableAMIIdError,
)
from .core import TaggedEC2Resource
from .instances import Instance
from ..utils import (
random_ami_id,
generic_filter,
@ -18,7 +20,7 @@ from ..utils import (
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 @@ else:
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 @@ class Ami(TaggedEC2Resource):
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 @@ class Ami(TaggedEC2Resource):
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 @@ class Ami(TaggedEC2Resource):
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 @@ class AmiBackend:
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 @@ class AmiBackend:
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 @@ class AmiBackend:
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 @@ class AmiBackend:
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()
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 len(ami_ids):
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 @@ class AmiBackend:
# 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 @@ class AmiBackend:
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)
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 @@ class AmiBackend:
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 @@ class AmiBackend:
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 @@ class AmiBackend:
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 @@ class AmiBackend:
if group:
ami.launch_permission_groups.discard(group)
return True

28
moto/ec2/models/availability_zones_and_regions.py
View File

@ -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 @@ class RegionsAndZonesBackend:
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 @@ class RegionsAndZonesBackend:
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

10
moto/ec2/models/core.py
View File

@ -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 @@ class TaggedEC2Resource(BaseModel):
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:"):

4
moto/ec2/responses/amis.py
View File

@ -32,9 +32,9 @@ class AmisResponse(EC2BaseResponse):
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()

7
moto/ec2/utils.py
View File

@ -7,6 +7,7 @@ from datetime import datetime
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 = [

9
moto/utilities/utils.py
View File

@ -3,7 +3,7 @@ import hashlib
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.
"""

2
setup.cfg
View File

@ -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