catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Some flake8 cleanup.

Browse source
This commit is contained in:
Steve Pulec 2014-11-15 09:35:52 -05:00
commit 8bc8f09b47
20 changed files with 90 additions and 86 deletions
Download patch file Download diff file Expand all files Collapse all files

3
moto/ec2/__init__.py
View file

@ -1,7 +1,8 @@
from __future__ import unicode_literals
from .models import ec2_backends, ec2_backend
from .models import ec2_backend, ec2_backends # flake8: noqa
from ..core.models import MockAWS
def mock_ec2(func=None):
if func:
return MockAWS(ec2_backends)(func)

2
moto/ec2/exceptions.py
View file

@ -163,7 +163,7 @@ class InvalidSnapshotIdError(EC2ClientError):
def __init__(self, snapshot_id):
super(InvalidSnapshotIdError, self).__init__(
"InvalidSnapshot.NotFound",
"") # Note: AWS returns empty message for this, as of 2014.08.22.
"") # Note: AWS returns empty message for this, as of 2014.08.22.
class InvalidVolumeIdError(EC2ClientError):

27
moto/ec2/models.py
View file

@ -1,8 +1,9 @@
from __future__ import unicode_literals
import copy
import itertools
from collections import defaultdict
import copy
from datetime import datetime
import itertools
import re
import six
@ -11,6 +12,7 @@ from boto.ec2.instance import Instance as BotoInstance, Reservation
from boto.ec2.blockdevicemapping import BlockDeviceMapping, BlockDeviceType
from boto.ec2.spotinstancerequest import SpotInstanceRequest as BotoSpotRequest
from boto.ec2.launchspecification import LaunchSpecification
import six
from moto.core import BaseBackend
from moto.core.models import Model
@ -240,12 +242,11 @@ class NetworkInterfaceBackend(object):
for (_filter, _filter_value) in filters.items():
if _filter == 'network-interface-id':
_filter = 'id'
enis = [ eni for eni in enis if getattr(eni, _filter) in _filter_value ]
enis = [eni for eni in enis if getattr(eni, _filter) in _filter_value]
elif _filter == 'group-id':
original_enis = enis
enis = []
for eni in original_enis:
group_ids = []
for group in eni.group_set:
if group.id in _filter_value:
enis.append(eni)
@ -817,7 +818,7 @@ class Ami(TaggedEC2Resource):
elif filter_name == 'kernel-id':
return self.kernel_id
elif filter_name in ['architecture', 'platform']:
return getattr(self,filter_name)
return getattr(self, filter_name)
elif filter_name == 'image-id':
return self.id
elif filter_name == 'state':
@ -856,7 +857,6 @@ class AmiBackend(object):
def describe_images(self, ami_ids=(), filters=None):
if filters:
images = self.amis.values()
return generic_filter(filters, images)
else:
images = []
@ -1008,10 +1008,7 @@ class SecurityGroup(object):
def physical_resource_id(self):
return self.id
def matches_filter(self, key, filter_value):
result = True
def to_attr(filter_name):
attr = None
@ -1031,7 +1028,7 @@ class SecurityGroup(object):
ingress_attr = to_attr(match.groups()[0])
for ingress in self.ingress_rules:
if getattr(ingress, ingress_attr) in filters[key]:
if getattr(ingress, ingress_attr) in filter_value:
return True
else:
attr_name = to_attr(key)
@ -1302,7 +1299,7 @@ class EBSBackend(object):
def detach_volume(self, volume_id, instance_id, device_path):
volume = self.get_volume(volume_id)
instance = self.get_instance(instance_id)
self.get_instance(instance_id)
old_attachment = volume.attachment
if not old_attachment:
@ -1406,7 +1403,7 @@ class VPCBackend(object):
self.vpcs[vpc_id] = vpc
# AWS creates a default main route table and security group.
main_route_table = self.create_route_table(vpc_id, main=True)
self.create_route_table(vpc_id, main=True)
default = self.get_security_group_from_name('default', vpc_id=vpc_id)
if not default:
@ -1429,7 +1426,7 @@ class VPCBackend(object):
def delete_vpc(self, vpc_id):
# Delete route table if only main route table remains.
route_tables = self.get_all_route_tables(filters={'vpc-id':vpc_id})
route_tables = self.get_all_route_tables(filters={'vpc-id': vpc_id})
if len(route_tables) > 1:
raise DependencyViolationError(
"The vpc {0} has dependencies and cannot be deleted."
@ -1599,7 +1596,7 @@ class SubnetBackend(object):
def create_subnet(self, vpc_id, cidr_block):
subnet_id = random_subnet_id()
subnet = Subnet(self, subnet_id, vpc_id, cidr_block)
vpc = self.get_vpc(vpc_id) # Validate VPC exists
self.get_vpc(vpc_id) # Validate VPC exists
self.subnets[subnet_id] = subnet
return subnet
@ -1719,7 +1716,7 @@ class RouteTableBackend(object):
route_tables = self.route_tables.values()
if route_table_ids:
route_tables = [ route_table for route_table in route_tables if route_table.id in route_table_ids ]
route_tables = [route_table for route_table in route_tables if route_table.id in route_table_ids]
if len(route_tables) != len(route_table_ids):
invalid_id = list(set(route_table_ids).difference(set([route_table.id for route_table in route_tables])))[0]
raise InvalidRouteTableIdError(invalid_id)

4
moto/ec2/responses/elastic_block_store.py
View file

@ -35,12 +35,12 @@ class ElasticBlockStore(BaseResponse):
def delete_snapshot(self):
snapshot_id = self.querystring.get('SnapshotId')[0]
success = self.ec2_backend.delete_snapshot(snapshot_id)
self.ec2_backend.delete_snapshot(snapshot_id)
return DELETE_SNAPSHOT_RESPONSE
def delete_volume(self):
volume_id = self.querystring.get('VolumeId')[0]
success = self.ec2_backend.delete_volume(volume_id)
self.ec2_backend.delete_volume(volume_id)
return DELETE_VOLUME_RESPONSE
def describe_snapshots(self):

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

@ -59,9 +59,9 @@ class ElasticIPAddresses(BaseResponse):
def disassociate_address(self):
if "PublicIp" in self.querystring:
disassociated = self.ec2_backend.disassociate_address(address=self.querystring['PublicIp'][0])
self.ec2_backend.disassociate_address(address=self.querystring['PublicIp'][0])
elif "AssociationId" in self.querystring:
disassociated = self.ec2_backend.disassociate_address(association_id=self.querystring['AssociationId'][0])
self.ec2_backend.disassociate_address(association_id=self.querystring['AssociationId'][0])
else:
self.ec2_backend.raise_error("MissingParameter", "Invalid request, expect PublicIp/AssociationId parameter.")
@ -69,9 +69,9 @@ class ElasticIPAddresses(BaseResponse):
def release_address(self):
if "PublicIp" in self.querystring:
released = self.ec2_backend.release_address(address=self.querystring['PublicIp'][0])
self.ec2_backend.release_address(address=self.querystring['PublicIp'][0])
elif "AllocationId" in self.querystring:
released = self.ec2_backend.release_address(allocation_id=self.querystring['AllocationId'][0])
self.ec2_backend.release_address(allocation_id=self.querystring['AllocationId'][0])
else:
self.ec2_backend.raise_error("MissingParameter", "Invalid request, expect PublicIp/AllocationId parameter.")

4
moto/ec2/responses/elastic_network_interfaces.py
View file

@ -25,7 +25,7 @@ class ElasticNetworkInterfaces(BaseResponse):
raise NotImplementedError('ElasticNetworkInterfaces(AmazonVPC).describe_network_interface_attribute is not yet implemented')
def describe_network_interfaces(self):
#Partially implemented. Supports only network-interface-id and group-id filters
# Partially implemented. Supports only network-interface-id and group-id filters
filters = filters_from_querystring(self.querystring)
enis = self.ec2_backend.describe_network_interfaces(filters)
template = Template(DESCRIBE_NETWORK_INTERFACES_RESPONSE)
@ -46,7 +46,7 @@ class ElasticNetworkInterfaces(BaseResponse):
return template.render()
def modify_network_interface_attribute(self):
#Currently supports modifying one and only one security group
# Currently supports modifying one and only one security group
eni_id = self.querystring.get('NetworkInterfaceId')[0]
group_id = self.querystring.get('SecurityGroupId.1')[0]
self.ec2_backend.modify_network_interface_attribute(eni_id, group_id)

1
moto/ec2/responses/internet_gateways.py
View file

@ -112,4 +112,3 @@ DETACH_INTERNET_GATEWAY_RESPONSE = u"""<DetachInternetGatewayResponse xmlns="htt
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<return>true</return>
</DetachInternetGatewayResponse>"""

4
moto/ec2/responses/route_tables.py
View file

@ -22,7 +22,7 @@ class RouteTables(BaseResponse):
interface_id = optional_from_querystring('NetworkInterfaceId', self.querystring)
pcx_id = optional_from_querystring('VpcPeeringConnectionId', self.querystring)
route = self.ec2_backend.create_route(route_table_id, destination_cidr_block,
self.ec2_backend.create_route(route_table_id, destination_cidr_block,
gateway_id=internet_gateway_id,
instance_id=instance_id,
interface_id=interface_id,
@ -72,7 +72,7 @@ class RouteTables(BaseResponse):
interface_id = optional_from_querystring('NetworkInterfaceId', self.querystring)
pcx_id = optional_from_querystring('VpcPeeringConnectionId', self.querystring)
route = self.ec2_backend.replace_route(route_table_id, destination_cidr_block,
self.ec2_backend.replace_route(route_table_id, destination_cidr_block,
gateway_id=internet_gateway_id,
instance_id=instance_id,
interface_id=interface_id,

3
moto/ec2/responses/vpc_peering_connections.py
View file

@ -31,7 +31,7 @@ class VPCPeeringConnections(BaseResponse):
def reject_vpc_peering_connection(self):
vpc_pcx_id = self.querystring.get('VpcPeeringConnectionId')[0]
vpc_pcx = self.ec2_backend.reject_vpc_peering_connection(vpc_pcx_id)
self.ec2_backend.reject_vpc_peering_connection(vpc_pcx_id)
template = Template(REJECT_VPC_PEERING_CONNECTION_RESPONSE)
return template.render()
@ -125,4 +125,3 @@ REJECT_VPC_PEERING_CONNECTION_RESPONSE = """
<return>true</return>
</RejectVpcPeeringConnectionResponse>
"""

2
moto/ec2/utils.py
View file

@ -130,7 +130,7 @@ def generate_route_id(route_table_id, cidr_block):
def split_route_id(route_id):
values = string.split(route_id, '~')
values = route_id.split('~')
return values[0], values[1]