1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
|
from typing import Any, Optional
from moto.core.responses import BaseResponse
from moto.core.serialize import return_if_not_empty
from ..exceptions import (
DryRunClientError,
EC2ClientError,
EmptyTagSpecError,
InvalidParameter,
)
from ..utils import convert_tag_spec
class EC2BaseResponse(BaseResponse):
RESPONSE_KEY_PATH_TO_TRANSFORMER = {
# ENI
"DescribeNetworkInterfacesResult.NetworkInterfaces.NetworkInterface.Association": return_if_not_empty,
# IAM Instance Profiles
"AssociateIamInstanceProfileResult.IamInstanceProfileAssociation.State": lambda _: "associating",
"DisassociateIamInstanceProfileResult.IamInstanceProfileAssociation.State": lambda _: "disassociating",
"ReplaceIamInstanceProfileAssociationResult.IamInstanceProfileAssociation.State": lambda _: "associating",
# Instances
"Instances.Instance.Tags": return_if_not_empty,
"RunInstances.Reservation.Instances.Instance.State": lambda _: {
"Code": 0,
"Name": "pending",
},
# Key Pairs
"KeyPair.Tags": return_if_not_empty,
}
@property
def ec2_backend(self) -> Any: # type: ignore[misc]
from moto.ec2.models import ec2_backends
return ec2_backends[self.current_account][self.region]
def _filters_from_querystring(self) -> dict[str, str]:
# [{"Name": x1, "Value": y1}, ..]
_filters = self._get_param("Filters", [])
# EC2 is inconsistent with parameter name, so check singular as well.
if not _filters:
_filters = self._get_param("Filter", [])
# return {x1: y1, ...}
try:
return {f["Name"]: f.get("Values", []) for f in _filters}
except KeyError:
raise EC2ClientError(
"InvalidParameterValue", "The filter 'null' is invalid."
)
def _parse_tag_specification(
self, expected_type: Optional[str] = None
) -> dict[str, dict[str, str]]:
# [{"ResourceType": _type, "Tag": [{"Key": k, "Value": v}, ..]}]
tag_spec_set = self._get_param("TagSpecification", [])
if not tag_spec_set:
tag_spec_set = self._get_param("TagSpecifications", [])
if not tag_spec_set:
return {}
tags_dict = (
tag_spec_set[0] if isinstance(tag_spec_set, list) else tag_spec_set
) # awscli allows for a json string to be passed and it should be allowed
if "ResourceType" not in tags_dict:
raise InvalidParameter("Tag specification resource type must have a value")
if expected_type and tags_dict["ResourceType"] != expected_type:
raise InvalidParameter(
f"'{tags_dict['ResourceType']}' is not a valid taggable resource type for this operation."
)
if "Tags" not in tags_dict:
if tags_dict.get("ResourceType") == "subnet":
raise InvalidParameter("Tag specification must have at least one tag")
raise EmptyTagSpecError
return convert_tag_spec(tag_spec_set)
def error_on_dryrun(self) -> None:
if self._get_param("DryRun", False):
raise DryRunClientError()
|
