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

EC2 - Simplify parameter handling (#4965)

Browse Source
This commit is contained in:
Bert Blommers 2022-03-24 12:57:13 -01:00 committed by GitHub
parent 472b1dab0c
commit 39b82c77cf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
28 changed files with 113 additions and 197 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
moto/ec2/models.py
View File

@ -680,7 +680,7 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel):
self.subnet_id = kwargs.get("subnet_id")
if not self.subnet_id:
self.subnet_id = next(
(n["SubnetId"] for n in nics.values() if "SubnetId" in n), None
(n["SubnetId"] for n in nics if "SubnetId" in n), None
)
in_ec2_classic = not bool(self.subnet_id)
self.key_name = kwargs.get("key_name")
@ -1008,11 +1008,11 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel):
# If empty NIC spec but primary NIC values provided, create NIC from
# them.
if primary_nic and not nic_spec:
nic_spec[0] = primary_nic
nic_spec = [primary_nic]
nic_spec[0]["DeviceIndex"] = 0
# Flesh out data structures and associations
for nic in nic_spec.values():
for nic in nic_spec:
device_index = int(nic.get("DeviceIndex"))
nic_id = nic.get("NetworkInterfaceId")

9
moto/ec2/responses/_base_response.py Normal file
View File

@ -0,0 +1,9 @@
from moto.core.responses import BaseResponse
class EC2BaseResponse(BaseResponse):
def _filters_from_querystring(self):
# [{"Name": x1, "Value": y1}, ..]
_filters = self._get_multi_param("Filter.")
# return {x1: y1, ...}
return {f["Name"]: f["Value"] for f in _filters}

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

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class AmisResponse(BaseResponse):
class AmisResponse(EC2BaseResponse):
def create_image(self):
name = self.querystring.get("Name")[0]
description = self._get_param("Description", if_none="")
@ -40,7 +39,7 @@ class AmisResponse(BaseResponse):
def describe_images(self):
self.error_on_dryrun()
ami_ids = self._get_multi_param("ImageId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
owners = self._get_multi_param("Owner")
exec_users = self._get_multi_param("ExecutableBy")
images = self.ec2_backend.describe_images(

8
moto/ec2/responses/carrier_gateways.py
View File

@ -1,8 +1,8 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring, add_tag_specification
from moto.ec2.utils import add_tag_specification
from ._base_response import EC2BaseResponse
class CarrierGateway(BaseResponse):
class CarrierGateway(EC2BaseResponse):
def create_carrier_gateway(self):
vpc_id = self._get_param("VpcId")
tags = self._get_multi_param("TagSpecification")
@ -23,7 +23,7 @@ class CarrierGateway(BaseResponse):
def describe_carrier_gateways(self):
carrier_gateway_ids = self._get_multi_param("CarrierGatewayId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
carrier_gateways = self.ec2_backend.describe_carrier_gateways(
carrier_gateway_ids, filters

7
moto/ec2/responses/customer_gateways.py
View File

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class CustomerGateways(BaseResponse):
class CustomerGateways(EC2BaseResponse):
def create_customer_gateway(self):
# raise NotImplementedError('CustomerGateways(AmazonVPC).create_customer_gateway is not yet implemented')
gateway_type = self._get_param("Type")
@ -26,7 +25,7 @@ class CustomerGateways(BaseResponse):
def describe_customer_gateways(self):
self.error_on_dryrun()
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
customer_gateway_ids = self._get_multi_param("CustomerGatewayId")
customer_gateways = self.ec2_backend.get_all_customer_gateways(
filters, customer_gateway_ids

10
moto/ec2/responses/dhcp_options.py
View File

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring, dhcp_configuration_from_querystring
from ._base_response import EC2BaseResponse
class DHCPOptions(BaseResponse):
class DHCPOptions(EC2BaseResponse):
def associate_dhcp_options(self):
dhcp_opt_id = self._get_param("DhcpOptionsId")
vpc_id = self._get_param("VpcId")
@ -16,7 +15,8 @@ class DHCPOptions(BaseResponse):
return template.render()
def create_dhcp_options(self):
dhcp_config = dhcp_configuration_from_querystring(self.querystring)
dhcp_config = self._get_multi_param("DhcpConfiguration")
dhcp_config = {f["Key"]: f["Value"] for f in dhcp_config}
# TODO validate we only got the options we know about
@ -45,7 +45,7 @@ class DHCPOptions(BaseResponse):
def describe_dhcp_options(self):
dhcp_opt_ids = self._get_multi_param("DhcpOptionsId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
dhcp_opts = self.ec2_backend.describe_dhcp_options(dhcp_opt_ids, filters)
template = self.response_template(DESCRIBE_DHCP_OPTIONS_RESPONSE)
return template.render(dhcp_options=dhcp_opts)

9
moto/ec2/responses/elastic_block_store.py
View File

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class ElasticBlockStore(BaseResponse):
class ElasticBlockStore(EC2BaseResponse):
def attach_volume(self):
volume_id = self._get_param("VolumeId")
instance_id = self._get_param("InstanceId")
@ -88,7 +87,7 @@ class ElasticBlockStore(BaseResponse):
return DELETE_VOLUME_RESPONSE
def describe_snapshots(self):
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
snapshot_ids = self._get_multi_param("SnapshotId")
snapshots = self.ec2_backend.describe_snapshots(
snapshot_ids=snapshot_ids, filters=filters
@ -97,7 +96,7 @@ class ElasticBlockStore(BaseResponse):
return template.render(snapshots=snapshots)
def describe_volumes(self):
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
volume_ids = self._get_multi_param("VolumeId")
volumes = self.ec2_backend.describe_volumes(
volume_ids=volume_ids, filters=filters

8
moto/ec2/responses/elastic_ip_addresses.py
View File

@ -1,8 +1,8 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring, add_tag_specification
from moto.ec2.utils import add_tag_specification
from ._base_response import EC2BaseResponse
class ElasticIPAddresses(BaseResponse):
class ElasticIPAddresses(EC2BaseResponse):
def allocate_address(self):
domain = self._get_param("Domain", if_none="standard")
reallocate_address = self._get_param("Address", if_none=None)
@ -72,7 +72,7 @@ class ElasticIPAddresses(BaseResponse):
self.error_on_dryrun()
allocation_ids = self._get_multi_param("AllocationId")
public_ips = self._get_multi_param("PublicIp")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
addresses = self.ec2_backend.describe_addresses(
allocation_ids, public_ips, filters
)

12
moto/ec2/responses/elastic_network_interfaces.py
View File

@ -1,12 +1,8 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import (
filters_from_querystring,
get_attribute_value,
add_tag_specification,
)
from moto.ec2.utils import get_attribute_value, add_tag_specification
from ._base_response import EC2BaseResponse
class ElasticNetworkInterfaces(BaseResponse):
class ElasticNetworkInterfaces(EC2BaseResponse):
def create_network_interface(self):
subnet_id = self._get_param("SubnetId")
private_ip_address = self._get_param("PrivateIpAddress")
@ -49,7 +45,7 @@ class ElasticNetworkInterfaces(BaseResponse):
def describe_network_interfaces(self):
eni_ids = self._get_multi_param("NetworkInterfaceId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
enis = self.ec2_backend.get_all_network_interfaces(eni_ids, filters)
template = self.response_template(DESCRIBE_NETWORK_INTERFACES_RESPONSE)
return template.render(enis=enis)

7
moto/ec2/responses/flow_logs.py
View File

@ -1,9 +1,8 @@
from moto.core.responses import BaseResponse
from moto.ec2.models import validate_resource_ids
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class FlowLogs(BaseResponse):
class FlowLogs(EC2BaseResponse):
def create_flow_logs(self):
resource_type = self._get_param("ResourceType")
resource_ids = self._get_multi_param("ResourceId")
@ -37,7 +36,7 @@ class FlowLogs(BaseResponse):
def describe_flow_logs(self):
flow_log_ids = self._get_multi_param("FlowLogId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
flow_logs = self.ec2_backend.describe_flow_logs(flow_log_ids, filters)
if self.is_not_dryrun("DescribeFlowLogs"):
template = self.response_template(DESCRIBE_FLOW_LOGS_RESPONSE)

12
moto/ec2/responses/instances.py
View File

@ -1,20 +1,20 @@
from moto.core.responses import BaseResponse
from moto.core.utils import camelcase_to_underscores
from moto.ec2.exceptions import (
MissingParameterError,
InvalidParameterCombination,
InvalidRequest,
)
from moto.ec2.utils import filters_from_querystring, dict_from_querystring
from moto.core import ACCOUNT_ID
from copy import deepcopy
from ._base_response import EC2BaseResponse
class InstanceResponse(BaseResponse):
class InstanceResponse(EC2BaseResponse):
def describe_instances(self):
self.error_on_dryrun()
filter_dict = filters_from_querystring(self.querystring)
filter_dict = self._filters_from_querystring()
instance_ids = self._get_multi_param("InstanceId")
token = self._get_param("NextToken")
if instance_ids:
@ -55,7 +55,7 @@ class InstanceResponse(BaseResponse):
"owner_id": owner_id,
"key_name": self._get_param("KeyName"),
"security_group_ids": self._get_multi_param("SecurityGroupId"),
"nics": dict_from_querystring("NetworkInterface", self.querystring),
"nics": self._get_multi_param("NetworkInterface."),
"private_ip": self._get_param("PrivateIpAddress"),
"associate_public_ip": self._get_param("AssociatePublicIpAddress"),
"tags": self._parse_tag_specification("TagSpecification"),
@ -154,7 +154,7 @@ class InstanceResponse(BaseResponse):
def describe_instance_type_offerings(self):
location_type_filters = self._get_param("LocationType")
filter_dict = filters_from_querystring(self.querystring)
filter_dict = self._filters_from_querystring()
offerings = self.ec2_backend.describe_instance_type_offerings(
location_type_filters, filter_dict
)

7
moto/ec2/responses/internet_gateways.py
View File

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class InternetGateways(BaseResponse):
class InternetGateways(EC2BaseResponse):
def attach_internet_gateway(self):
igw_id = self._get_param("InternetGatewayId")
vpc_id = self._get_param("VpcId")
@ -30,7 +29,7 @@ class InternetGateways(BaseResponse):
return template.render()
def describe_internet_gateways(self):
filter_dict = filters_from_querystring(self.querystring)
filter_dict = self._filters_from_querystring()
if "InternetGatewayId.1" in self.querystring:
igw_ids = self._get_multi_param("InternetGatewayId")
igws = self.ec2_backend.describe_internet_gateways(

7
moto/ec2/responses/key_pairs.py
View File

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class KeyPairs(BaseResponse):
class KeyPairs(EC2BaseResponse):
def create_key_pair(self):
name = self._get_param("KeyName")
if self.is_not_dryrun("CreateKeyPair"):
@ -20,7 +19,7 @@ class KeyPairs(BaseResponse):
def describe_key_pairs(self):
names = self._get_multi_param("KeyName")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
keypairs = self.ec2_backend.describe_key_pairs(names, filters)
template = self.response_template(DESCRIBE_KEY_PAIRS_RESPONSE)
return template.render(keypairs=keypairs)

9
moto/ec2/responses/launch_templates.py
View File

@ -1,8 +1,7 @@
import uuid
from moto.core.responses import BaseResponse
from moto.ec2.models import OWNER_ID
from moto.ec2.exceptions import FilterNotImplementedError
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
from xml.etree import ElementTree
from xml.dom import minidom
@ -91,7 +90,7 @@ def parse_lists(data):
return data
class LaunchTemplates(BaseResponse):
class LaunchTemplates(EC2BaseResponse):
def create_launch_template(self):
name = self._get_param("LaunchTemplateName")
version_description = self._get_param("VersionDescription")
@ -195,7 +194,7 @@ class LaunchTemplates(BaseResponse):
min_version = self._get_int_param("MinVersion")
max_version = self._get_int_param("MaxVersion")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
if filters:
raise FilterNotImplementedError(
"all filters", "DescribeLaunchTemplateVersions"
@ -255,7 +254,7 @@ class LaunchTemplates(BaseResponse):
max_results = self._get_int_param("MaxResults", 15)
template_names = self._get_multi_param("LaunchTemplateName")
template_ids = self._get_multi_param("LaunchTemplateId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
if self.is_not_dryrun("DescribeLaunchTemplates"):
tree = ElementTree.Element("DescribeLaunchTemplatesResponse")

8
moto/ec2/responses/nat_gateways.py
View File

@ -1,8 +1,8 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring, add_tag_specification
from moto.ec2.utils import add_tag_specification
from ._base_response import EC2BaseResponse
class NatGateways(BaseResponse):
class NatGateways(EC2BaseResponse):
def create_nat_gateway(self):
subnet_id = self._get_param("SubnetId")
allocation_id = self._get_param("AllocationId")
@ -26,7 +26,7 @@ class NatGateways(BaseResponse):
return template.render(nat_gateway=nat_gateway)
def describe_nat_gateways(self):
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
nat_gateway_ids = self._get_multi_param("NatGatewayId")
nat_gateways = self.ec2_backend.describe_nat_gateways(filters, nat_gateway_ids)
template = self.response_template(DESCRIBE_NAT_GATEWAYS_RESPONSE)

7
moto/ec2/responses/network_acls.py
View File

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class NetworkACLs(BaseResponse):
class NetworkACLs(EC2BaseResponse):
def create_network_acl(self):
vpc_id = self._get_param("VpcId")
tags = self._get_multi_param("TagSpecification")
@ -84,7 +83,7 @@ class NetworkACLs(BaseResponse):
def describe_network_acls(self):
network_acl_ids = self._get_multi_param("NetworkAclId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
network_acls = self.ec2_backend.describe_network_acls(network_acl_ids, filters)
template = self.response_template(DESCRIBE_NETWORK_ACL_RESPONSE)
return template.render(network_acls=network_acls)

7
moto/ec2/responses/route_tables.py
View File

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class RouteTables(BaseResponse):
class RouteTables(EC2BaseResponse):
def associate_route_table(self):
route_table_id = self._get_param("RouteTableId")
gateway_id = self._get_param("GatewayId")
@ -76,7 +75,7 @@ class RouteTables(BaseResponse):
def describe_route_tables(self):
route_table_ids = self._get_multi_param("RouteTableId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
route_tables = self.ec2_backend.describe_route_tables(route_table_ids, filters)
template = self.response_template(DESCRIBE_ROUTE_TABLES_RESPONSE)
return template.render(route_tables=route_tables)

7
moto/ec2/responses/security_groups.py
View File

@ -1,5 +1,4 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
def try_parse_int(value, default=None):
@ -68,7 +67,7 @@ def parse_sg_attributes_from_dict(sg_attributes):
return (ip_protocol, from_port, to_port, ip_ranges, source_groups, prefix_list_ids)
class SecurityGroups(BaseResponse):
class SecurityGroups(EC2BaseResponse):
def _process_rules_from_querystring(self):
group_name_or_id = self._get_param("GroupName") or self._get_param("GroupId")
@ -186,7 +185,7 @@ class SecurityGroups(BaseResponse):
def describe_security_groups(self):
groupnames = self._get_multi_param("GroupName")
group_ids = self._get_multi_param("GroupId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
groups = self.ec2_backend.describe_security_groups(
group_ids=group_ids, groupnames=groupnames, filters=filters

9
moto/ec2/responses/spot_instances.py
View File

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class SpotInstances(BaseResponse):
class SpotInstances(EC2BaseResponse):
def cancel_spot_instance_requests(self):
request_ids = self._get_multi_param("SpotInstanceRequestId")
if self.is_not_dryrun("CancelSpotInstance"):
@ -29,7 +28,7 @@ class SpotInstances(BaseResponse):
def describe_spot_instance_requests(self):
spot_instance_ids = self._get_multi_param("SpotInstanceRequestId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
requests = self.ec2_backend.describe_spot_instance_requests(
filters=filters, spot_instance_ids=spot_instance_ids
)
@ -38,7 +37,7 @@ class SpotInstances(BaseResponse):
def describe_spot_price_history(self):
instance_types_filters = self._get_multi_param("InstanceType")
filter_dict = filters_from_querystring(self.querystring)
filter_dict = self._filters_from_querystring()
prices = self.ec2_backend.describe_spot_price_history(
instance_types_filters, filter_dict
)

8
moto/ec2/responses/subnets.py
View File

@ -1,10 +1,10 @@
import random
from moto.core.responses import BaseResponse
from moto.core.utils import camelcase_to_underscores
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class Subnets(BaseResponse):
class Subnets(EC2BaseResponse):
def create_subnet(self):
vpc_id = self._get_param("VpcId")
cidr_block = self._get_param("CidrBlock")
@ -39,7 +39,7 @@ class Subnets(BaseResponse):
def describe_subnets(self):
self.error_on_dryrun()
subnet_ids = self._get_multi_param("SubnetId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
subnets = self.ec2_backend.get_all_subnets(subnet_ids, filters)
template = self.response_template(DESCRIBE_SUBNETS_RESPONSE)
return template.render(subnets=subnets)

7
moto/ec2/responses/tags.py
View File

@ -1,10 +1,9 @@
from moto.core.responses import BaseResponse
from moto.ec2.models import validate_resource_ids
from moto.ec2.utils import filters_from_querystring
from moto.core.utils import tags_from_query_string
from ._base_response import EC2BaseResponse
class TagResponse(BaseResponse):
class TagResponse(EC2BaseResponse):
def create_tags(self):
resource_ids = self._get_multi_param("ResourceId")
validate_resource_ids(resource_ids)
@ -23,7 +22,7 @@ class TagResponse(BaseResponse):
return DELETE_RESPONSE
def describe_tags(self):
filters = filters_from_querystring(querystring_dict=self.querystring)
filters = self._filters_from_querystring()
tags = self.ec2_backend.describe_tags(filters=filters)
template = self.response_template(DESCRIBE_RESPONSE)
return template.render(tags=tags)

12
moto/ec2/responses/transit_gateway_attachments.py
View File

@ -1,8 +1,8 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring, add_tag_specification
from moto.ec2.utils import add_tag_specification
from ._base_response import EC2BaseResponse
class TransitGatewayAttachment(BaseResponse):
class TransitGatewayAttachment(EC2BaseResponse):
def create_transit_gateway_vpc_attachment(self):
options = self._get_multi_param_dict("Options")
subnet_ids = self._get_multi_param("SubnetIds")
@ -30,7 +30,7 @@ class TransitGatewayAttachment(BaseResponse):
transit_gateways_attachment_ids = self._get_multi_param(
"TransitGatewayAttachmentIds"
)
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
transit_gateway_vpc_attachments = (
self.ec2_backend.describe_transit_gateway_vpc_attachments(
transit_gateways_attachment_ids=transit_gateways_attachment_ids,
@ -63,7 +63,7 @@ class TransitGatewayAttachment(BaseResponse):
transit_gateways_attachment_ids = self._get_multi_param(
"TransitGatewayAttachmentIds"
)
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
transit_gateway_attachments = (
self.ec2_backend.describe_transit_gateway_attachments(
transit_gateways_attachment_ids=transit_gateways_attachment_ids,
@ -154,7 +154,7 @@ class TransitGatewayAttachment(BaseResponse):
transit_gateways_attachment_ids = self._get_multi_param(
"TransitGatewayAttachmentIds"
)
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
transit_gateway_peering_attachments = (
self.ec2_backend.describe_transit_gateway_peering_attachments(
transit_gateways_attachment_ids=transit_gateways_attachment_ids,

13
moto/ec2/responses/transit_gateway_route_tables.py
View File

@ -1,9 +1,8 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
from moto.utilities.utils import str2bool
class TransitGatewayRouteTable(BaseResponse):
class TransitGatewayRouteTable(EC2BaseResponse):
def create_transit_gateway_route_table(self):
transit_gateway_id = self._get_param("TransitGatewayId")
tags = self._get_multi_param("TagSpecifications")
@ -20,7 +19,7 @@ class TransitGatewayRouteTable(BaseResponse):
return template.render(transit_gateway_route_table=transit_gateway_route_table)
def describe_transit_gateway_route_tables(self):
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
transit_gateway_route_table_ids = (
self._get_multi_param("TransitGatewayRouteTableIds") or None
)
@ -78,7 +77,7 @@ class TransitGatewayRouteTable(BaseResponse):
def search_transit_gateway_routes(self):
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
max_results = self._get_param("MaxResults")
transit_gateway_routes = self.ec2_backend.search_transit_gateway_routes(
transit_gateway_route_table_id=transit_gateway_route_table_id,
@ -90,7 +89,7 @@ class TransitGatewayRouteTable(BaseResponse):
def get_transit_gateway_route_table_associations(self):
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
transit_gateway_route_table_associations = (
self.ec2_backend.get_all_transit_gateway_route_table_associations(
transit_gateway_route_table_id, filters
@ -105,7 +104,7 @@ class TransitGatewayRouteTable(BaseResponse):
def get_transit_gateway_route_table_propagations(self):
transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
transit_gateway_route_table_propagations = (
self.ec2_backend.get_all_transit_gateway_route_table_propagations(
transit_gateway_route_table_id, filters

7
moto/ec2/responses/transit_gateways.py
View File

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class TransitGateways(BaseResponse):
class TransitGateways(EC2BaseResponse):
def create_transit_gateway(self):
description = self._get_param("Description") or None
options = self._get_multi_param_dict("Options")
@ -41,7 +40,7 @@ class TransitGateways(BaseResponse):
def describe_transit_gateways(self):
transit_gateway_ids = self._get_multi_param("TransitGatewayIds")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
transit_gateways = self.ec2_backend.describe_transit_gateways(
filters, transit_gateway_ids
)

7
moto/ec2/responses/virtual_private_gateways.py
View File

@ -1,8 +1,7 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring
from ._base_response import EC2BaseResponse
class VirtualPrivateGateways(BaseResponse):
class VirtualPrivateGateways(EC2BaseResponse):
def attach_vpn_gateway(self):
vpn_gateway_id = self._get_param("VpnGatewayId")
vpc_id = self._get_param("VpcId")
@ -34,7 +33,7 @@ class VirtualPrivateGateways(BaseResponse):
return template.render(vpn_gateway=vpn_gateway)
def describe_vpn_gateways(self):
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
vpn_gw_ids = self._get_multi_param("VpnGatewayId")
vpn_gateways = self.ec2_backend.describe_vpn_gateways(filters, vpn_gw_ids)
template = self.response_template(DESCRIBE_VPN_GATEWAYS_RESPONSE)

18
moto/ec2/responses/vpcs.py
View File

@ -1,10 +1,10 @@
from moto.core import ACCOUNT_ID
from moto.core.responses import BaseResponse
from moto.core.utils import camelcase_to_underscores
from moto.ec2.utils import add_tag_specification, filters_from_querystring
from moto.ec2.utils import add_tag_specification
from ._base_response import EC2BaseResponse
class VPCs(BaseResponse):
class VPCs(EC2BaseResponse):
def _get_doc_date(self):
return (
"2013-10-15"
@ -41,7 +41,7 @@ class VPCs(BaseResponse):
def describe_vpcs(self):
self.error_on_dryrun()
vpc_ids = self._get_multi_param("VpcId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
vpcs = self.ec2_backend.describe_vpcs(vpc_ids=vpc_ids, filters=filters)
doc_date = (
"2013-10-15"
@ -68,7 +68,7 @@ class VPCs(BaseResponse):
def describe_vpc_classic_link_dns_support(self):
vpc_ids = self._get_multi_param("VpcIds")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
vpcs = self.ec2_backend.describe_vpcs(vpc_ids=vpc_ids, filters=filters)
doc_date = self._get_doc_date()
template = self.response_template(
@ -100,7 +100,7 @@ class VPCs(BaseResponse):
def describe_vpc_classic_link(self):
vpc_ids = self._get_multi_param("VpcId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
vpcs = self.ec2_backend.describe_vpcs(vpc_ids=vpc_ids, filters=filters)
doc_date = self._get_doc_date()
template = self.response_template(DESCRIBE_VPC_CLASSIC_LINK_RESPONSE)
@ -219,7 +219,7 @@ class VPCs(BaseResponse):
def describe_vpc_endpoints(self):
vpc_end_points_ids = self._get_multi_param("VpcEndpointId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
vpc_end_points = self.ec2_backend.describe_vpc_endpoints(
vpc_end_point_ids=vpc_end_points_ids, filters=filters
)
@ -255,7 +255,7 @@ class VPCs(BaseResponse):
def describe_managed_prefix_lists(self):
prefix_list_ids = self._get_multi_param("PrefixListId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
managed_prefix_lists = self.ec2_backend.describe_managed_prefix_lists(
prefix_list_ids=prefix_list_ids, filters=filters
)
@ -291,7 +291,7 @@ class VPCs(BaseResponse):
def describe_prefix_lists(self):
prefix_list_ids = self._get_multi_param("PrefixListId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
managed_pls = self.ec2_backend.describe_managed_prefix_lists(
prefix_list_ids=prefix_list_ids, filters=filters
)

8
moto/ec2/responses/vpn_connections.py
View File

@ -1,9 +1,9 @@
from moto.core.responses import BaseResponse
from moto.ec2.utils import filters_from_querystring, add_tag_specification
from moto.ec2.utils import add_tag_specification
from ._base_response import EC2BaseResponse
from xml.sax.saxutils import escape
class VPNConnections(BaseResponse):
class VPNConnections(EC2BaseResponse):
def create_vpn_connection(self):
vpn_conn_type = self._get_param("Type")
cgw_id = self._get_param("CustomerGatewayId")
@ -41,7 +41,7 @@ class VPNConnections(BaseResponse):
def describe_vpn_connections(self):
vpn_connection_ids = self._get_multi_param("VpnConnectionId")
filters = filters_from_querystring(self.querystring)
filters = self._filters_from_querystring()
vpn_connections = self.ec2_backend.get_all_vpn_connections(
vpn_connection_ids=vpn_connection_ids, filters=filters
)

74
moto/ec2/utils.py
View File

@ -302,80 +302,6 @@ def split_route_id(route_id):
return values[0], values[1]
def dhcp_configuration_from_querystring(querystring, option="DhcpConfiguration"):
"""
turn:
{u'AWSAccessKeyId': [u'the_key'],
u'Action': [u'CreateDhcpOptions'],
u'DhcpConfiguration.1.Key': [u'domain-name'],
u'DhcpConfiguration.1.Value.1': [u'example.com'],
u'DhcpConfiguration.2.Key': [u'domain-name-servers'],
u'DhcpConfiguration.2.Value.1': [u'10.0.0.6'],
u'DhcpConfiguration.2.Value.2': [u'10.0.0.7'],
u'Signature': [u'uUMHYOoLM6r+sT4fhYjdNT6MHw22Wj1mafUpe0P0bY4='],
u'SignatureMethod': [u'HmacSHA256'],
u'SignatureVersion': [u'2'],
u'Timestamp': [u'2014-03-18T21:54:01Z'],
u'Version': [u'2013-10-15']}
into:
{u'domain-name': [u'example.com'], u'domain-name-servers': [u'10.0.0.6', u'10.0.0.7']}
"""
key_needle = re.compile("{0}.[0-9]+.Key".format(option), re.UNICODE)
response_values = {}
for key, value in querystring.items():
if key_needle.match(key):
values = []
key_index = key.split(".")[1]
value_index = 1
while True:
value_key = "{0}.{1}.Value.{2}".format(option, key_index, value_index)
if value_key in querystring:
values.extend(querystring[value_key])
else:
break
value_index += 1
response_values[value[0]] = values
return response_values
def filters_from_querystring(querystring_dict):
response_values = {}
last_tag_key = None
for key, value in sorted(querystring_dict.items()):
match = re.search(r"Filter.(\d).Name", key)
if match:
filter_index = match.groups()[0]
value_prefix = "Filter.{0}.Value".format(filter_index)
filter_values = [
filter_value[0]
for filter_key, filter_value in querystring_dict.items()
if filter_key.startswith(value_prefix)
]
if value[0] == "tag-key":
last_tag_key = "tag:" + filter_values[0]
elif last_tag_key and value[0] == "tag-value":
response_values[last_tag_key] = filter_values
response_values[value[0]] = filter_values
return response_values
def dict_from_querystring(parameter, querystring_dict):
use_dict = {}
for key, value in querystring_dict.items():
match = re.search(r"{0}.(\d).(\w+)".format(parameter), key)
if match:
use_dict_index = match.groups()[0]
use_dict_element_property = match.groups()[1]
if not use_dict.get(use_dict_index):
use_dict[use_dict_index] = {}
use_dict[use_dict_index][use_dict_element_property] = value[0]
return use_dict
def get_attribute_value(parameter, querystring_dict):
for key, value in querystring_dict.items():
match = re.search(r"{0}.Value".format(parameter), key)