openstack
/
ec2-api
Code Issues Proposed changes

Rework fake action execute interface 

Change-Id: Iedd8ee3d2303bd8e3f6cd511be433ae443259cef
This commit is contained in:
Feodor Tersin 2015-03-06 00:56:19 +03:00
parent 94f8e9644a
commit 2c1d2dc9fb
16 changed files with 470 additions and 800 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

87
ec2api/tests/unit/base.py
View File

@ -39,6 +39,8 @@ def skip_not_implemented(test_item):
class ApiTestCase(test_base.BaseTestCase):
ANY_EXECUTE_ERROR = object()
def setUp(self):
super(ApiTestCase, self).setUp()
@ -76,42 +78,17 @@ class ApiTestCase(test_base.BaseTestCase):
self.configure(fatal_exception_format_errors=True)
def execute(self, action, args):
ec2_request = ec2api.api.apirequest.APIRequest(action, 'fake_v1', args)
ec2_context = self._create_context()
environ = {'REQUEST_METHOD': 'FAKE',
'ec2.request': ec2_request,
'ec2api.context': ec2_context}
request = ec2api.wsgi.Request(environ)
response = request.send(ec2api.api.Executor())
return self._check_and_transform_response(response, action)
status_code, response = self._execute(action, args)
self.assertEqual(200, status_code)
return response
def _create_context(self):
return ec2api.context.RequestContext(
fakes.ID_OS_USER, fakes.ID_OS_PROJECT,
service_catalog=[{'type': 'network',
'endpoints': [{'publicUrl': 'fake_url'}]}])
def _check_and_transform_response(self, response, action):
body = tools.parse_xml(response.body)
if response.status_code == 200:
action_tag = '%sResponse' % action
self.assertIn(action_tag, body)
body = body.pop(action_tag)
self.assertIn('requestId', body)
body.pop('requestId')
def assert_execution_error(self, error_code, action, args):
status_code, response = self._execute(action, args)
if error_code == self.ANY_EXECUTE_ERROR:
self.assertLessEqual(400, status_code)
else:
self.assertIn('Response', body)
body = body.pop('Response')
self.assertIn('RequestID', body)
body.pop('RequestID')
self.assertEqual(1, len(body))
self.assertIn('Errors', body)
body = body.pop('Errors')
self.assertEqual(1, len(body))
self.assertIn('Error', body)
self.assertEqual(2, len(body['Error']))
body['http_status_code'] = response.status_code
return body
self.assertEqual(400, status_code)
self.assertEqual(error_code, response['Error']['Code'])
def assert_any_call(self, func, *args, **kwargs):
calls = func.mock_calls
@ -146,14 +123,12 @@ class ApiTestCase(test_base.BaseTestCase):
resp = self.execute(operation,
{'Filter.1.Name': name,
'Filter.1.Value.1': str(value)})
self.assertEqual(200, resp['http_status_code'])
self.assertTrue(len(resp[resultset_key]) > 0,
'Filter by %s does not work' % name)
resp = self.execute(operation,
{'Filter.1.Name': name,
'Filter.1.Value.1': 'dummy filter value'})
self.assertEqual(200, resp['http_status_code'])
self.assertTrue(resp[resultset_key] is None or
len(resp[resultset_key]) == 0)
@ -167,7 +142,6 @@ class ApiTestCase(test_base.BaseTestCase):
'value': 'fake_value'}]
resp = self.execute(operation, {})
self.assertEqual(200, resp['http_status_code'])
tag_found = False
if type(resultset_key) is list:
resp_items = itertools.chain(*(r[resultset_key[1]]
@ -195,7 +169,6 @@ class ApiTestCase(test_base.BaseTestCase):
id_param = '%s%s.1' % (id_key[0].capitalize(), id_key[1:])
resp = self.execute(operation, {id_param: sample_item_id})
self.assertEqual(200, resp['http_status_code'])
self.assertTrue(
self.db_api.get_tags.call_count == 1 and
(self.db_api.get_tags.mock_calls[0] in
@ -207,3 +180,41 @@ class ApiTestCase(test_base.BaseTestCase):
[('tag-key', 'fake_key'),
('tag-value', 'fake_value'),
('tag:fake_key', 'fake_value')])
def _create_context(self):
return ec2api.context.RequestContext(
fakes.ID_OS_USER, fakes.ID_OS_PROJECT,
service_catalog=[{'type': 'network',
'endpoints': [{'publicUrl': 'fake_url'}]}])
def _execute(self, action, args):
ec2_request = ec2api.api.apirequest.APIRequest(action, 'fake_v1', args)
ec2_context = self._create_context()
environ = {'REQUEST_METHOD': 'FAKE',
'ec2.request': ec2_request,
'ec2api.context': ec2_context}
request = ec2api.wsgi.Request(environ)
response = request.send(ec2api.api.Executor())
return (response.status_code,
self._check_and_transform_response(response, action))
def _check_and_transform_response(self, response, action):
body = tools.parse_xml(response.body)
if response.status_code == 200:
action_tag = '%sResponse' % action
self.assertIn(action_tag, body)
body = body.pop(action_tag)
self.assertIn('requestId', body)
body.pop('requestId')
else:
self.assertIn('Response', body)
body = body.pop('Response')
self.assertIn('RequestID', body)
body.pop('RequestID')
self.assertEqual(1, len(body))
self.assertIn('Errors', body)
body = body.pop('Errors')
self.assertEqual(1, len(body))
self.assertIn('Error', body)
self.assertEqual(2, len(body['Error']))
return body

121
ec2api/tests/unit/test_address.py
View File

@ -32,7 +32,6 @@ class AddressTestCase(base.ApiTestCase):
fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_1))
resp = self.execute('AllocateAddress', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(fakes.IP_ADDRESS_1, resp['publicIp'])
self.assertEqual('standard', resp['domain'])
self.assertNotIn('allocationId', resp)
@ -51,7 +50,6 @@ class AddressTestCase(base.ApiTestCase):
resp = self.execute('AllocateAddress', {'Domain': 'vpc'})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(fakes.IP_ADDRESS_1, resp['publicIp'])
self.assertEqual('vpc', resp['domain'])
self.assertEqual(fakes.ID_EC2_ADDRESS_1,
@ -71,9 +69,8 @@ class AddressTestCase(base.ApiTestCase):
def test_allocate_address_invalid_parameters(self):
address.address_engine = (
address.AddressEngineNeutron())
resp = self.execute('AllocateAddress', {'Domain': 'fake_domain'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterValue', 'AllocateAddress',
{'Domain': 'fake_domain'})
self.assertEqual(0, self.db_api.add_item.call_count)
self.assertEqual(0, self.neutron.create_floatingip.call_count)
@ -82,17 +79,15 @@ class AddressTestCase(base.ApiTestCase):
address.AddressEngineNeutron())
self.neutron.create_floatingip.side_effect = (
neutron_exception.OverQuotaClient())
resp = self.execute('AllocateAddress', {'Domain': 'vpc'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('AddressLimitExceeded', resp['Error']['Code'])
self.assert_execution_error('AddressLimitExceeded', 'AllocateAddress',
{'Domain': 'vpc'})
address.address_engine = (
address.AddressEngineNeutron())
self.nova.floating_ips.create.side_effect = (
nova_exception.Forbidden(403))
resp = self.execute('AllocateAddress', {})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('AddressLimitExceeded', resp['Error']['Code'])
self.assert_execution_error('AddressLimitExceeded', 'AllocateAddress',
{})
def test_allocate_address_vpc_rollback(self):
address.address_engine = (
@ -104,7 +99,8 @@ class AddressTestCase(base.ApiTestCase):
{'floatingip': fakes.OS_FLOATING_IP_1})
self.db_api.add_item.side_effect = Exception()
self.execute('AllocateAddress', {'Domain': 'vpc'})
self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'AllocateAddress',
{'Domain': 'vpc'})
self.neutron.delete_floatingip.assert_called_once_with(
fakes.ID_OS_FLOATING_IP_1)
@ -121,7 +117,6 @@ class AddressTestCase(base.ApiTestCase):
resp = self.execute('AssociateAddress',
{'PublicIp': fakes.IP_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.nova.servers.add_floating_ip.assert_called_once_with(
@ -134,7 +129,6 @@ class AddressTestCase(base.ApiTestCase):
def do_check(params, fixed_ip):
resp = self.execute('AssociateAddress', params)
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.assertEqual(fakes.ID_EC2_ASSOCIATION_1, resp['associationId'])
@ -193,7 +187,6 @@ class AddressTestCase(base.ApiTestCase):
def do_check(params):
resp = self.execute('AssociateAddress', params)
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.assertEqual(fakes.ID_EC2_ASSOCIATION_2, resp['associationId'])
@ -218,9 +211,7 @@ class AddressTestCase(base.ApiTestCase):
address.AddressEngineNeutron())
def do_check(params, error):
resp = self.execute('AssociateAddress', params)
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error, resp['Error']['Code'])
self.assert_execution_error(error, 'AssociateAddress', params)
do_check({},
'MissingParameter')
@ -240,38 +231,31 @@ class AddressTestCase(base.ApiTestCase):
address.AddressEngineNeutron())
# NOTE(ft): ec2 classic instance vs allocation_id parameter
self.set_mock_db_items(fakes.DB_INSTANCE_2)
resp = self.execute('AssociateAddress',
{'AllocationId': 'eipalloc-0',
'InstanceId': fakes.ID_EC2_INSTANCE_2})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterCombination', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterCombination',
'AssociateAddress',
{'AllocationId': 'eipalloc-0',
'InstanceId': fakes.ID_EC2_INSTANCE_2})
# NOTE(ft): ec2 classic instance vs not existing public IP
self.nova.floating_ips.list.return_value = []
resp = self.execute('AssociateAddress',
{'PublicIp': fakes.IP_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_2})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('AuthFailure', resp['Error']['Code'])
self.assert_execution_error('AuthFailure', 'AssociateAddress',
{'PublicIp': fakes.IP_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_2})
# NOTE(ft): ec2 classic instance vs vpc public ip
self.add_mock_db_items(fakes.DB_ADDRESS_1, fakes.DB_ADDRESS_2)
self.neutron.show_floatingip.return_value = (
{'floatingip': fakes.OS_FLOATING_IP_1})
resp = self.execute('AssociateAddress',
{'PublicIp': fakes.IP_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_2})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('AuthFailure', resp['Error']['Code'])
self.assert_execution_error('AuthFailure', 'AssociateAddress',
{'PublicIp': fakes.IP_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_2})
def test_associate_address_invalid_vpc_parameters(self):
address.address_engine = (
address.AddressEngineNeutron())
def do_check(params, error):
resp = self.execute('AssociateAddress', params)
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error, resp['Error']['Code'])
self.assert_execution_error(error, 'AssociateAddress', params)
# NOTE(ft): not registered instance id vs vpc address
self.set_mock_db_items()
@ -309,9 +293,10 @@ class AddressTestCase(base.ApiTestCase):
fakes.DB_NETWORK_INTERFACE_1)
self.neutron.show_floatingip.return_value = (
{'floatingip': fakes.OS_FLOATING_IP_2})
do_check({'AllocationId': fakes.ID_EC2_ADDRESS_2,
'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1},
'Resource.AlreadyAssociated')
self.assert_execution_error(
'Resource.AlreadyAssociated', 'AssociateAddress',
{'AllocationId': fakes.ID_EC2_ADDRESS_2,
'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
# NOTE(ft): already associated address vs vpc instance
self.set_mock_db_items(
@ -323,23 +308,24 @@ class AddressTestCase(base.ApiTestCase):
fakes.ID_EC2_SUBNET_1,
fakes.IP_NETWORK_INTERFACE_1,
instance_id=fakes.ID_EC2_INSTANCE_1))
do_check({'AllocationId': fakes.ID_EC2_ADDRESS_2,
'InstanceId': fakes.ID_EC2_INSTANCE_1},
'Resource.AlreadyAssociated')
self.assert_execution_error('Resource.AlreadyAssociated',
'AssociateAddress',
{'AllocationId': fakes.ID_EC2_ADDRESS_2,
'InstanceId': fakes.ID_EC2_INSTANCE_1})
# NOTE(ft): multiple network interfaces in vpc instance
# w/o network interface selection
self.add_mock_db_items(fakes.DB_NETWORK_INTERFACE_2)
do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1},
'InvalidInstanceID')
self.assert_execution_error('InvalidInstanceID', 'AssociateAddress',
{'AllocationId': fakes.ID_EC2_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1})
# NOTE(ft): internet gateway isn't attached to the vpc
self.set_mock_db_items(fakes.DB_ADDRESS_1,
fakes.DB_NETWORK_INTERFACE_2)
do_check({'AllocationId': fakes.ID_EC2_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1},
'Gateway.NotAttached')
self.assert_execution_error('Gateway.NotAttached', 'AssociateAddress',
{'AllocationId': fakes.ID_EC2_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1})
def test_associate_address_vpc_rollback(self):
address.address_engine = (
@ -351,9 +337,9 @@ class AddressTestCase(base.ApiTestCase):
{'floatingip': fakes.OS_FLOATING_IP_1})
self.neutron.update_floatingip.side_effect = Exception()
self.execute('AssociateAddress',
{'AllocationId': fakes.ID_EC2_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1})
self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'AssociateAddress',
{'AllocationId': fakes.ID_EC2_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1})
self.db_api.update_item.assert_any_call(
mock.ANY, fakes.DB_ADDRESS_1)
@ -368,7 +354,6 @@ class AddressTestCase(base.ApiTestCase):
fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_2)])
resp = self.execute('DisassociateAddress',
{'PublicIp': fakes.IP_ADDRESS_2})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.nova.servers.remove_floating_ip.assert_called_once_with(
fakes.ID_OS_INSTANCE_1,
@ -377,7 +362,6 @@ class AddressTestCase(base.ApiTestCase):
# NOTE(Alex) Disassociate unassociated address in EC2 classic
resp = self.execute('DisassociateAddress',
{'PublicIp': fakes.IP_ADDRESS_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.assertEqual(1, self.nova.servers.remove_floating_ip.call_count)
@ -390,7 +374,6 @@ class AddressTestCase(base.ApiTestCase):
resp = self.execute('DisassociateAddress',
{'AssociationId': fakes.ID_EC2_ASSOCIATION_2})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.neutron.update_floatingip.assert_called_once_with(
@ -410,7 +393,6 @@ class AddressTestCase(base.ApiTestCase):
resp = self.execute('DisassociateAddress',
{'AssociationId': fakes.ID_EC2_ASSOCIATION_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.assertEqual(0, self.neutron.update_floatingip.call_count)
@ -421,9 +403,7 @@ class AddressTestCase(base.ApiTestCase):
address.AddressEngineNeutron())
def do_check(params, error):
resp = self.execute('DisassociateAddress', params)
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error, resp['Error']['Code'])
self.assert_execution_error(error, 'DisassociateAddress', params)
do_check({},
'MissingParameter')
@ -435,10 +415,8 @@ class AddressTestCase(base.ApiTestCase):
# NOTE(ft): EC2 Classic public IP does not exists
self.set_mock_db_items()
self.nova.floating_ips.list.return_value = []
resp = self.execute('DisassociateAddress',
{'PublicIp': fakes.IP_ADDRESS_2})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('AuthFailure', resp['Error']['Code'])
self.assert_execution_error('AuthFailure', 'DisassociateAddress',
{'PublicIp': fakes.IP_ADDRESS_2})
# NOTE(ft): vpc address vs public ip parameter
self.set_mock_db_items(fakes.DB_ADDRESS_1)
@ -466,8 +444,9 @@ class AddressTestCase(base.ApiTestCase):
{'floatingip': fakes.OS_FLOATING_IP_2})
self.neutron.update_floatingip.side_effect = Exception()
self.execute('DisassociateAddress',
{'AssociationId': fakes.ID_EC2_ASSOCIATION_2})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'DisassociateAddress',
{'AssociationId': fakes.ID_EC2_ASSOCIATION_2})
self.db_api.update_item.assert_any_call(
mock.ANY, fakes.DB_ADDRESS_2)
@ -483,7 +462,6 @@ class AddressTestCase(base.ApiTestCase):
resp = self.execute('ReleaseAddress',
{'PublicIp': fakes.IP_ADDRESS_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.nova.floating_ips.delete.assert_called_once_with(
@ -498,7 +476,6 @@ class AddressTestCase(base.ApiTestCase):
resp = self.execute('ReleaseAddress',
{'AllocationId': fakes.ID_EC2_ADDRESS_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.neutron.delete_floatingip.assert_called_once_with(
@ -511,9 +488,7 @@ class AddressTestCase(base.ApiTestCase):
address.AddressEngineNeutron())
def do_check(params, error):
resp = self.execute('ReleaseAddress', params)
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error, resp['Error']['Code'])
self.assert_execution_error(error, 'ReleaseAddress', params)
do_check({},
'MissingParameter')
@ -561,8 +536,8 @@ class AddressTestCase(base.ApiTestCase):
{'floatingip': fakes.OS_FLOATING_IP_1})
self.neutron.delete_floatingip.side_effect = Exception()
self.execute('ReleaseAddress',
{'AllocationId': fakes.ID_EC2_ADDRESS_1})
self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'ReleaseAddress',
{'AllocationId': fakes.ID_EC2_ADDRESS_1})
self.db_api.restore_item.assert_called_once_with(
mock.ANY, 'eipalloc', fakes.DB_ADDRESS_1)
@ -581,7 +556,6 @@ class AddressTestCase(base.ApiTestCase):
fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2)
resp = self.execute('DescribeAddresses', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['addressesSet'],
matchers.ListMatches([fakes.EC2_ADDRESS_1,
fakes.EC2_ADDRESS_2]))
@ -590,7 +564,6 @@ class AddressTestCase(base.ApiTestCase):
return_value=[fakes.DB_ADDRESS_1])
resp = self.execute('DescribeAddresses',
{'AllocationId.1': fakes.ID_EC2_ADDRESS_1})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['addressesSet'],
matchers.ListMatches([fakes.EC2_ADDRESS_1]))
self.db_api.get_items_by_ids.assert_called_once_with(
@ -615,12 +588,10 @@ class AddressTestCase(base.ApiTestCase):
fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_1),
fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_2)]
resp = self.execute('DescribeAddresses', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['addressesSet'],
matchers.ListMatches([fakes.EC2_ADDRESS_CLASSIC_1,
fakes.EC2_ADDRESS_CLASSIC_2]))
resp = self.execute('DescribeAddresses', {'PublicIp.1':
fakes.IP_ADDRESS_2})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['addressesSet'],
matchers.ListMatches([fakes.EC2_ADDRESS_CLASSIC_2]))

13
ec2api/tests/unit/test_availability_zone.py
View File

@ -27,7 +27,6 @@ class AvailabilityZoneCase(base.ApiTestCase):
fakes.NovaAvailabilityZone(fakes.OS_AVAILABILITY_ZONE),
fakes.NovaAvailabilityZone(fakes.OS_AVAILABILITY_ZONE_INTERNAL)]
resp = self.execute('DescribeAvailabilityZones', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['availabilityZoneInfo'],
matchers.ListMatches([fakes.EC2_AVAILABILITY_ZONE]))
self.nova.availability_zones.list.assert_called_once_with(
@ -44,13 +43,11 @@ class AvailabilityZoneCase(base.ApiTestCase):
fakes.NovaAvailabilityZone(fakes.OS_AVAILABILITY_ZONE_INTERNAL)]
resp = self.execute('DescribeAvailabilityZones',
{'zoneName.1': 'verbose'})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(len(resp['availabilityZoneInfo']), 7)
self.nova.availability_zones.list.assert_called_once_with()
def test_regions(self):
resp = self.execute('DescribeRegions', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(resp['regionInfo'][0]['regionName'], 'nova')
self.assertTrue(resp['regionInfo'][0].get('regionEndpoint')
is not None)
@ -61,7 +58,6 @@ class AvailabilityZoneCase(base.ApiTestCase):
availability_zone.account_attribute_engine = (
availability_zone.AccountAttributeEngineNeutron())
resp = self.execute('DescribeAccountAttributes', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['accountAttributeSet'],
matchers.ListMatches(
[{'attributeName': 'supported-platforms',
@ -81,7 +77,6 @@ class AvailabilityZoneCase(base.ApiTestCase):
availability_zone.account_attribute_engine = (
availability_zone.AccountAttributeEngineNova())
resp = self.execute('DescribeAccountAttributes', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['accountAttributeSet'],
matchers.ListMatches(
[{'attributeName': 'supported-platforms',
@ -98,7 +93,6 @@ class AvailabilityZoneCase(base.ApiTestCase):
resp = self.execute('DescribeAccountAttributes',
{'AttributeName.1': 'default-vpc',
'AttributeName.2': 'max-instances'})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['accountAttributeSet'],
matchers.ListMatches(
[{'attributeName': 'default-vpc',
@ -109,7 +103,6 @@ class AvailabilityZoneCase(base.ApiTestCase):
{'attributeValue': 77}]}],
orderless_lists=True))
resp = self.execute('DescribeAccountAttributes',
{'AttributeName.1': 'fake'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameter', resp['Error']['Code'])
self.assert_execution_error('InvalidParameter',
'DescribeAccountAttributes',
{'AttributeName.1': 'fake'})

28
ec2api/tests/unit/test_dhcp_options.py
View File

@ -48,7 +48,6 @@ class DhcpOptionsTestCase(base.ApiTestCase):
resp = self.execute(
'CreateDhcpOptions',
gen_ec2_param_dhcp_options(ec2_fake))
self.assertEqual(200, resp['http_status_code'])
self.assertThat(ec2_fake, matchers.DictMatches(
resp['dhcpOptions'], orderless_lists=True))
self.assert_any_call(self.db_api.add_item,
@ -60,17 +59,15 @@ class DhcpOptionsTestCase(base.ApiTestCase):
check(fakes.EC2_DHCP_OPTIONS_2, fakes.DB_DHCP_OPTIONS_2)
def test_create_dhcp_options_invalid_parameters(self):
resp = self.execute('CreateDhcpOptions',
{'DhcpConfiguration.1.Key': 'InvalidParameter',
'DhcpConfiguration.1.Value.1': 'Value'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error(
'InvalidParameterValue', 'CreateDhcpOptions',
{'DhcpConfiguration.1.Key': 'InvalidParameter',
'DhcpConfiguration.1.Value.1': 'Value'})
def test_delete_dhcp_options(self):
self.set_mock_db_items(fakes.DB_DHCP_OPTIONS_1)
resp = self.execute('DeleteDhcpOptions',
{'dhcpOptionsId': fakes.ID_EC2_DHCP_OPTIONS_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.db_api.get_item_by_id.assert_any_call(
mock.ANY,
@ -88,17 +85,15 @@ class DhcpOptionsTestCase(base.ApiTestCase):
tools.update_dict(
fakes.DB_VPC_1,
{'dhcp_options_id': fakes.ID_EC2_DHCP_OPTIONS_1}))
resp = self.execute('DeleteDhcpOptions',
{'dhcpOptionsId': fakes.ID_EC2_DHCP_OPTIONS_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('DependencyViolation', resp['Error']['Code'])
self.assert_execution_error(
'DependencyViolation', 'DeleteDhcpOptions',
{'dhcpOptionsId': fakes.ID_EC2_DHCP_OPTIONS_1})
def test_describe_dhcp_options(self):
self.set_mock_db_items(fakes.DB_DHCP_OPTIONS_1,
fakes.DB_DHCP_OPTIONS_2)
resp = self.execute('DescribeDhcpOptions', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['dhcpOptionsSet'],
matchers.ListMatches([fakes.EC2_DHCP_OPTIONS_1,
fakes.EC2_DHCP_OPTIONS_2],
@ -106,7 +101,6 @@ class DhcpOptionsTestCase(base.ApiTestCase):
resp = self.execute('DescribeDhcpOptions',
{'DhcpOptionsId.1': fakes.ID_EC2_DHCP_OPTIONS_1})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['dhcpOptionsSet'],
matchers.ListMatches([fakes.EC2_DHCP_OPTIONS_1],
orderless_lists=True))
@ -131,7 +125,6 @@ class DhcpOptionsTestCase(base.ApiTestCase):
resp = self.execute('AssociateDhcpOptions',
{'dhcpOptionsId': ec2_dhcp_options_id,
'vpcId': fakes.ID_EC2_VPC_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.db_api.update_item.assert_any_call(
mock.ANY,
@ -164,9 +157,10 @@ class DhcpOptionsTestCase(base.ApiTestCase):
self.neutron.update_port.side_effect = update_port_func
self.execute('AssociateDhcpOptions',
{'dhcpOptionsId': fakes.ID_EC2_DHCP_OPTIONS_2,
'vpcId': fakes.ID_EC2_VPC_1})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'AssociateDhcpOptions',
{'dhcpOptionsId': fakes.ID_EC2_DHCP_OPTIONS_2,
'vpcId': fakes.ID_EC2_VPC_1})
self.assert_any_call(self.neutron.update_port,
fakes.ID_OS_PORT_1,

54
ec2api/tests/unit/test_image.py
View File

@ -115,8 +115,7 @@ class ImageTestCase(base.ApiTestCase):
{'InstanceId': fakes.ID_EC2_INSTANCE_2,
'Name': 'fake_name',
'NoReboot': str(no_reboot)})
self.assertEqual({'http_status_code': 200,
'imageId': image_id},
self.assertEqual({'imageId': image_id},
resp)
self.db_api.get_item_by_id.assert_called_once_with(
mock.ANY, fakes.ID_EC2_INSTANCE_2)
@ -142,11 +141,9 @@ class ImageTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_INSTANCE_1)
is_ebs_instance.return_value = False
resp = self.execute('CreateImage',
{'InstanceId': fakes.ID_EC2_INSTANCE_1,
'Name': 'fake_name'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterValue', 'CreateImage',
{'InstanceId': fakes.ID_EC2_INSTANCE_1,
'Name': 'fake_name'})
@mock.patch('ec2api.api.image._s3_create')
def test_register_image_by_s3(self, s3_create):
@ -158,8 +155,7 @@ class ImageTestCase(base.ApiTestCase):
'RegisterImage',
{'ImageLocation': fakes.LOCATION_IMAGE_1})
self.assertThat(resp, matchers.DictMatches(
{'http_status_code': 200,
'imageId': fakes.ID_EC2_IMAGE_1}))
{'imageId': fakes.ID_EC2_IMAGE_1}))
s3_create.assert_called_once_with(
mock.ANY,
@ -172,8 +168,7 @@ class ImageTestCase(base.ApiTestCase):
{'ImageLocation': fakes.LOCATION_IMAGE_1,
'Name': 'an image name'})
self.assertThat(resp, matchers.DictMatches(
{'http_status_code': 200,
'imageId': fakes.ID_EC2_IMAGE_1}))
{'imageId': fakes.ID_EC2_IMAGE_1}))
s3_create.assert_called_once_with(
mock.ANY,
@ -200,8 +195,7 @@ class ImageTestCase(base.ApiTestCase):
'BlockDeviceMapping.1.DeviceName': fakes.ROOT_DEVICE_NAME_IMAGE_2,
'BlockDeviceMapping.1.Ebs.SnapshotId': fakes.ID_EC2_SNAPSHOT_1})
self.assertThat(resp, matchers.DictMatches(
{'http_status_code': 200,
'imageId': fakes.ID_EC2_IMAGE_2}))
{'imageId': fakes.ID_EC2_IMAGE_2}))
self.db_api.add_item.assert_called_once_with(
mock.ANY, 'ami', {'os_id': fakes.ID_OS_IMAGE_2,
'is_public': False})
@ -231,17 +225,15 @@ class ImageTestCase(base.ApiTestCase):
mock.call(mock.ANY, fakes.ID_EC2_IMAGE_ARI_1)])
def test_register_image_invalid_parameters(self):
resp = self.execute('RegisterImage', {})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterCombination', resp['Error']['Code'])
self.assert_execution_error(
'InvalidParameterCombination', 'RegisterImage', {})
def test_deregister_image(self):
self._setup_model()
resp = self.execute('DeregisterImage',
{'ImageId': fakes.ID_EC2_IMAGE_1})
self.assertThat(resp, matchers.DictMatches({'http_status_code': 200,
'return': True}))
self.assertThat(resp, matchers.DictMatches({'return': True}))
self.db_api.delete_item.assert_called_once_with(
mock.ANY, fakes.ID_EC2_IMAGE_1)
self.glance.images.delete.assert_called_once_with(
@ -250,17 +242,13 @@ class ImageTestCase(base.ApiTestCase):
def test_deregister_image_invalid_parameters(self):
self._setup_model()
resp = self.execute('DeregisterImage',
{'ImageId': fakes.random_ec2_id('ami')})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidAMIID.NotFound', resp['Error']['Code'])
self.assert_execution_error('InvalidAMIID.NotFound', 'DeregisterImage',
{'ImageId': fakes.random_ec2_id('ami')})
def test_describe_images(self):
self._setup_model()
resp = self.execute('DescribeImages', {})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'imagesSet': [fakes.EC2_IMAGE_1, fakes.EC2_IMAGE_2]},
orderless_lists=True))
@ -274,8 +262,6 @@ class ImageTestCase(base.ApiTestCase):
resp = self.execute('DescribeImages',
{'ImageId.1': fakes.ID_EC2_IMAGE_1})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'imagesSet': [fakes.EC2_IMAGE_1]},
orderless_lists=True))
@ -304,17 +290,13 @@ class ImageTestCase(base.ApiTestCase):
def test_describe_images_invalid_parameters(self):
self._setup_model()
resp = self.execute('DescribeImages',
{'ImageId.1': fakes.random_ec2_id('ami')})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidAMIID.NotFound', resp['Error']['Code'])
self.assert_execution_error('InvalidAMIID.NotFound', 'DescribeImages',
{'ImageId.1': fakes.random_ec2_id('ami')})
self.glance.images.list.side_effect = lambda: []
resp = self.execute('DescribeImages',
{'ImageId.1': fakes.ID_EC2_IMAGE_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidAMIID.NotFound', resp['Error']['Code'])
self.assert_execution_error('InvalidAMIID.NotFound', 'DescribeImages',
{'ImageId.1': fakes.ID_EC2_IMAGE_1})
def test_describe_image_attributes(self):
self._setup_model()
@ -323,7 +305,6 @@ class ImageTestCase(base.ApiTestCase):
resp = self.execute('DescribeImageAttribute',
{'ImageId': ec2_image_id,
'Attribute': attr})
response['http_status_code'] = 200
response['imageId'] = ec2_image_id
self.assertThat(resp, matchers.DictMatches(response,
orderless_lists=True))
@ -367,8 +348,7 @@ class ImageTestCase(base.ApiTestCase):
'attribute': 'launchPermission',
'operationType': 'add',
'userGroup.1': 'all'})
self.assertThat(resp, matchers.DictMatches({'http_status_code': 200,
'return': True}))
self.assertThat(resp, matchers.DictMatches({'return': True}))
osimage_update.assert_called_once_with(
mock.ANY, is_public=True)
self.assertEqual(fakes.ID_OS_IMAGE_1,

110
ec2api/tests/unit/test_instance.py
View File

@ -119,8 +119,6 @@ class InstanceTestCase(base.ApiTestCase):
'MinCount': '1', 'MaxCount': '1'})
resp = self.execute('RunInstances', params)
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(expected_reservation))
if create_network_interface_kwargs is not None:
(self.network_interface_api.
@ -256,8 +254,6 @@ class InstanceTestCase(base.ApiTestCase):
'NetworkInterface.2.SubnetId': fakes.ID_EC2_SUBNET_2,
'NetworkInterface.2.DeleteOnTermination': 'False'})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(ec2_reservation),
verbose=True)
@ -313,7 +309,7 @@ class InstanceTestCase(base.ApiTestCase):
def do_check(engine, extra_kwargs={}, extra_db_instance={}):
instance_api.instance_engine = engine
resp = self.execute(
self.execute(
'RunInstances',
{'ImageId': fakes.ID_EC2_IMAGE_1,
'InstanceType': 'fake_flavor',
@ -327,7 +323,6 @@ class InstanceTestCase(base.ApiTestCase):
'BlockDeviceMapping.1.Ebs.SnapshotId': (
fakes.ID_EC2_SNAPSHOT_1),
'BlockDeviceMapping.1.Ebs.DeleteOnTermination': 'False'})
self.assertEqual(200, resp['http_status_code'])
self.nova.servers.create.assert_called_once_with(
mock.ANY, mock.ANY, mock.ANY, min_count=1, max_count=1,
@ -390,9 +385,7 @@ class InstanceTestCase(base.ApiTestCase):
'ImageId': fakes.ID_EC2_IMAGE_1,
'InstanceType': 'fake_flavor',
'ClientToken': 'client-token-1'})
self.assertEqual({'http_status_code': 200,
'key': 'value'},
resp)
self.assertEqual({'key': 'value'}, resp)
format_reservation.assert_called_once_with(
mock.ANY, instances[1]['reservation_id'],
[(instances[1], os_instances[1], 'novadb_instance')],
@ -446,9 +439,7 @@ class InstanceTestCase(base.ApiTestCase):
'ImageId': fakes.ID_EC2_IMAGE_1,
'InstanceType': 'fake_flavor',
'ClientToken': 'client-token'})
self.assertEqual({'http_status_code': 200,
'key': 'value'},
resp)
self.assertEqual({'key': 'value'}, resp)
format_reservation.assert_called_once_with(
mock.ANY, instances[0]['reservation_id'],
[(instances[0], os_instances[0], 'novadb-instance-0'),
@ -488,7 +479,8 @@ class InstanceTestCase(base.ApiTestCase):
params.update({'ImageId': fakes.ID_EC2_IMAGE_1,
'InstanceType': 'fake_flavor',
'MinCount': '1', 'MaxCount': '1'})
self.execute('RunInstances', params)
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'RunInstances', params)
calls = []
calls.append(
@ -574,8 +566,6 @@ class InstanceTestCase(base.ApiTestCase):
'InstanceType': 'fake_flavor',
'MinCount': '2', 'MaxCount': '3',
'SubnetId': fakes.ID_EC2_SUBNET_1})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp,
matchers.DictMatches(
{'reservationId': fakes.ID_EC2_RESERVATION_1,
@ -601,29 +591,21 @@ class InstanceTestCase(base.ApiTestCase):
do_check(instance_api.InstanceEngineNova())
def test_run_instances_invalid_parameters(self):
resp = self.execute('RunInstances',
{'ImageId': fakes.ID_EC2_IMAGE_1,
'MinCount': '0', 'MaxCount': '0'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterValue', 'RunInstances',
{'ImageId': fakes.ID_EC2_IMAGE_1,
'MinCount': '0', 'MaxCount': '0'})
resp = self.execute('RunInstances',
{'ImageId': fakes.ID_EC2_IMAGE_1,
'MinCount': '1', 'MaxCount': '0'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterValue', 'RunInstances',
{'ImageId': fakes.ID_EC2_IMAGE_1,
'MinCount': '1', 'MaxCount': '0'})
resp = self.execute('RunInstances',
{'ImageId': fakes.ID_EC2_IMAGE_1,
'MinCount': '0', 'MaxCount': '1'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterValue', 'RunInstances',
{'ImageId': fakes.ID_EC2_IMAGE_1,
'MinCount': '0', 'MaxCount': '1'})
resp = self.execute('RunInstances',
{'ImageId': fakes.ID_EC2_IMAGE_1,
'MinCount': '2', 'MaxCount': '1'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterValue', 'RunInstances',
{'ImageId': fakes.ID_EC2_IMAGE_1,
'MinCount': '2', 'MaxCount': '1'})
@mock.patch.object(fakes.OSInstance, 'delete', autospec=True)
@mock.patch.object(fakes.OSInstance, 'get', autospec=True)
@ -642,8 +624,6 @@ class InstanceTestCase(base.ApiTestCase):
{'InstanceId.1': fakes.ID_EC2_INSTANCE_1,
'InstanceId.2': fakes.ID_EC2_INSTANCE_2})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
fake_state_change = {'previousState': {'code': 0,
'name': 'pending'},
'currentState': {'code': 0,
@ -701,8 +681,6 @@ class InstanceTestCase(base.ApiTestCase):
{'InstanceId.1': fakes.ID_EC2_INSTANCE_1,
'InstanceId.2': fakes.ID_EC2_INSTANCE_2})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(
resp, matchers.DictMatches(ec2_terminate_instances_result))
detach_network_interface = (
@ -740,10 +718,9 @@ class InstanceTestCase(base.ApiTestCase):
deleted_enis=[])
def test_terminate_instances_invalid_parameters(self):
resp = self.execute('TerminateInstances',
{'InstanceId.1': fakes.random_ec2_id('i')})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidInstanceID.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidInstanceID.NotFound', 'TerminateInstances',
{'InstanceId.1': fakes.random_ec2_id('i')})
@mock.patch('ec2api.api.instance._get_os_instances_by_instances')
def _test_instances_operation(self, operation, os_instance_operation,
@ -761,9 +738,7 @@ class InstanceTestCase(base.ApiTestCase):
resp = self.execute(operation,
{'InstanceId.1': fakes.ID_EC2_INSTANCE_1,
'InstanceId.2': fakes.ID_EC2_INSTANCE_2})
self.assertEqual({'http_status_code': 200,
'return': True},
resp)
self.assertEqual({'return': True}, resp)
self.assertEqual([mock.call(os_instance_1), mock.call(os_instance_2)],
os_instance_operation.mock_calls)
self.db_api.get_items_by_ids.assert_called_once_with(
@ -773,11 +748,9 @@ class InstanceTestCase(base.ApiTestCase):
setattr(os_instance_2, 'OS-EXT-STS:vm_state', invalid_state)
os_instance_operation.reset_mock()
resp = self.execute('StartInstances',
{'InstanceId.1': fakes.ID_EC2_INSTANCE_1,
'InstanceId.2': fakes.ID_EC2_INSTANCE_2})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('IncorrectInstanceState', resp['Error']['Code'])
self.assert_execution_error('IncorrectInstanceState', 'StartInstances',
{'InstanceId.1': fakes.ID_EC2_INSTANCE_1,
'InstanceId.2': fakes.ID_EC2_INSTANCE_2})
self.assertEqual(0, os_instance_operation.call_count)
@mock.patch.object(fakes.OSInstance, 'start', autospec=True)
@ -806,8 +779,7 @@ class InstanceTestCase(base.ApiTestCase):
utcnow.return_value = datetime.datetime(2015, 1, 19, 23, 34, 45, 123)
resp = self.execute(operation,
{'InstanceId': fakes.ID_EC2_INSTANCE_2})
self.assertEqual({'http_status_code': 200,
'instanceId': fakes.ID_EC2_INSTANCE_2,
self.assertEqual({'instanceId': fakes.ID_EC2_INSTANCE_2,
'timestamp': '2015-01-19T23:34:45.000Z',
key: 'fake_data'},
resp)
@ -853,8 +825,6 @@ class InstanceTestCase(base.ApiTestCase):
resp = self.execute('DescribeInstances', {})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'reservationSet': [fakes.EC2_RESERVATION_1,
fakes.EC2_RESERVATION_2]},
@ -864,8 +834,6 @@ class InstanceTestCase(base.ApiTestCase):
return_value=[fakes.DB_INSTANCE_1])
resp = self.execute('DescribeInstances',
{'InstanceId.1': fakes.ID_EC2_INSTANCE_1})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'reservationSet': [fakes.EC2_RESERVATION_1]},
orderless_lists=True))
@ -932,8 +900,6 @@ class InstanceTestCase(base.ApiTestCase):
resp = self.execute('DescribeInstances', {})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'reservationSet': [fakes.EC2_RESERVATION_2]},
orderless_lists=True))
@ -973,9 +939,6 @@ class InstanceTestCase(base.ApiTestCase):
resp = self.execute('DescribeInstances', {})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
instances = [fakes.gen_ec2_instance(
inst_id, launch_index=l_i, private_ip_address=ip,
ec2_network_interfaces=enis,
@ -1028,8 +991,7 @@ class InstanceTestCase(base.ApiTestCase):
self.assertThat(resp,
matchers.DictMatches(
{'http_status_code': 200,
'reservationSet': [fakes.EC2_RESERVATION_2]},
{'reservationSet': [fakes.EC2_RESERVATION_2]},
orderless_lists=True))
remove_instances.assert_called_once_with(
mock.ANY, [fakes.DB_INSTANCE_1], purge_linked_items=False)
@ -1055,24 +1017,21 @@ class InstanceTestCase(base.ApiTestCase):
'amiLaunchIndex': instance['launch_index']}))
resp = self.execute('DescribeInstances', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(
[0, 1, 2, 3, 4],
[inst['amiLaunchIndex']
for inst in resp['reservationSet'][0]['instancesSet']])
def test_describe_instances_invalid_parameters(self):
resp = self.execute('DescribeInstances', {'InstanceId.1':
fakes.random_ec2_id('i')})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidInstanceID.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidInstanceID.NotFound', 'DescribeInstances',
{'InstanceId.1': fakes.random_ec2_id('i')})
self.set_mock_db_items(fakes.DB_INSTANCE_2)
resp = self.execute('DescribeInstances',
{'InstanceId.1': fakes.ID_EC2_INSTANCE_2,
'InstanceId.2': fakes.random_ec2_id('i')})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidInstanceID.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidInstanceID.NotFound', 'DescribeInstances',
{'InstanceId.1': fakes.ID_EC2_INSTANCE_2,
'InstanceId.2': fakes.random_ec2_id('i')})
def test_describe_instance_attributes(self):
self.set_mock_db_items(fakes.DB_INSTANCE_1, fakes.DB_INSTANCE_2,
@ -1097,8 +1056,7 @@ class InstanceTestCase(base.ApiTestCase):
resp = self.execute('DescribeInstanceAttribute',
{'InstanceId': instance_id,
'Attribute': attribute})
expected.update({'http_status_code': 200,
'instanceId': instance_id})
expected.update({'instanceId': instance_id})
self.assertThat(resp, matchers.DictMatches(expected))
do_check(fakes.ID_EC2_INSTANCE_2, 'blockDeviceMapping',

51
ec2api/tests/unit/test_internet_gateway.py
View File

@ -34,7 +34,6 @@ class IgwTestCase(base.ApiTestCase):
resp = self.execute('CreateInternetGateway', {})
self.assertEqual(resp['http_status_code'], 200)
self.assertIn('internetGateway', resp)
igw = resp['internetGateway']
self.assertThat(fakes.EC2_IGW_2, matchers.DictMatches(igw))
@ -52,7 +51,6 @@ class IgwTestCase(base.ApiTestCase):
{'VpcId': fakes.ID_EC2_VPC_2,
'InternetGatewayId': fakes.ID_EC2_IGW_2})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.db_api.get_item_by_id.assert_any_call(mock.ANY,
fakes.ID_EC2_IGW_2)
@ -70,13 +68,11 @@ class IgwTestCase(base.ApiTestCase):
def test_attach_igw_invalid_parameters(self):
def do_check(error_code):
resp = self.execute(
'AttachInternetGateway',
{'VpcId': fakes.ID_EC2_VPC_2,
'InternetGatewayId': fakes.ID_EC2_IGW_2})
self.assert_execution_error(
error_code, 'AttachInternetGateway',
{'VpcId': fakes.ID_EC2_VPC_2,
'InternetGatewayId': fakes.ID_EC2_IGW_2})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
self.assertEqual(0, self.neutron.add_gateway_router.call_count)
self.assertEqual(0, self.db_api.update_item.call_count)
@ -104,9 +100,10 @@ class IgwTestCase(base.ApiTestCase):
{'networks': [{'id': fakes.ID_OS_PUBLIC_NETWORK}]})
self.neutron.add_gateway_router.side_effect = Exception()
self.execute('AttachInternetGateway',
{'VpcId': fakes.ID_EC2_VPC_2,
'InternetGatewayId': fakes.ID_EC2_IGW_2})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'AttachInternetGateway',
{'VpcId': fakes.ID_EC2_VPC_2,
'InternetGatewayId': fakes.ID_EC2_IGW_2})
self.db_api.update_item.assert_any_call(
mock.ANY, fakes.DB_IGW_2)
@ -119,7 +116,6 @@ class IgwTestCase(base.ApiTestCase):
{'VpcId': fakes.EC2_VPC_1['vpcId'],
'InternetGatewayId': fakes.EC2_IGW_1['internetGatewayId']})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.db_api.get_item_by_id.assert_any_call(mock.ANY,
fakes.ID_EC2_IGW_1)
@ -132,13 +128,11 @@ class IgwTestCase(base.ApiTestCase):
def test_detach_igw_invalid_parameters(self):
def do_check(error_code):
resp = self.execute(
'DetachInternetGateway',
{'VpcId': fakes.ID_EC2_VPC_1,
'InternetGatewayId': fakes.ID_EC2_IGW_1})
self.assert_execution_error(
error_code, 'DetachInternetGateway',
{'VpcId': fakes.ID_EC2_VPC_1,
'InternetGatewayId': fakes.ID_EC2_IGW_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
self.assertEqual(0, self.neutron.remove_gateway_router.call_count)
self.assertEqual(0, self.db_api.update_item.call_count)
@ -164,7 +158,6 @@ class IgwTestCase(base.ApiTestCase):
{'VpcId': fakes.ID_EC2_VPC_1,
'InternetGatewayId': fakes.ID_EC2_IGW_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.neutron.remove_gateway_router.assert_called_once_with(
fakes.ID_OS_ROUTER_1)
@ -173,10 +166,10 @@ class IgwTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_IGW_1, fakes.DB_VPC_1)
self.neutron.remove_gateway_router.side_effect = Exception()
self.execute(
'DetachInternetGateway',
{'VpcId': fakes.EC2_VPC_1['vpcId'],
'InternetGatewayId': fakes.EC2_IGW_1['internetGatewayId']})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'DetachInternetGateway',
{'VpcId': fakes.EC2_VPC_1['vpcId'],
'InternetGatewayId': fakes.EC2_IGW_1['internetGatewayId']})
self.db_api.update_item.assert_any_call(
mock.ANY, fakes.DB_IGW_1)
@ -188,7 +181,6 @@ class IgwTestCase(base.ApiTestCase):
'DeleteInternetGateway',
{'InternetGatewayId': fakes.ID_EC2_IGW_2})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.db_api.get_item_by_id.assert_called_once_with(mock.ANY,
fakes.ID_EC2_IGW_2)
@ -197,13 +189,10 @@ class IgwTestCase(base.ApiTestCase):
def test_delete_igw_invalid_parameters(self):
def do_check(error_code):
resp = self.execute(
'DeleteInternetGateway',
{'InternetGatewayId': (
fakes.EC2_IGW_1['internetGatewayId'])})
self.assert_execution_error(
error_code, 'DeleteInternetGateway',
{'InternetGatewayId': fakes.ID_EC2_IGW_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
self.assertEqual(0, self.db_api.delete_item.call_count)
self.neutron.reset_mock()
@ -219,14 +208,12 @@ class IgwTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_IGW_1, fakes.DB_IGW_2)
resp = self.execute('DescribeInternetGateways', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['internetGatewaySet'],
matchers.ListMatches([fakes.EC2_IGW_1,
fakes.EC2_IGW_2]))
resp = self.execute('DescribeInternetGateways',
{'InternetGatewayId.1': fakes.ID_EC2_IGW_2})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['internetGatewaySet'],
matchers.ListMatches([fakes.EC2_IGW_2]))
self.db_api.get_items_by_ids.assert_called_once_with(

47
ec2api/tests/unit/test_key_pair.py
View File

@ -28,25 +28,22 @@ class KeyPairCase(base.ApiTestCase):
self.nova.keypairs.create.return_value = (
fakes.NovaKeyPair(fakes.OS_KEY_PAIR))
resp = self.execute('CreateKeyPair', {'KeyName': fakes.NAME_KEY_PAIR})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(fakes.EC2_KEY_PAIR, matchers.DictMatches(
tools.purge_dict(resp, {'http_status_code'})))
self.assertThat(fakes.EC2_KEY_PAIR, matchers.DictMatches(resp))
self.nova.keypairs.create.assert_called_once_with(fakes.NAME_KEY_PAIR)
def test_create_key_pair_invalid(self):
self.nova.keypairs.create.side_effect = (
nova_exception.Conflict(409))
resp = self.execute('CreateKeyPair', {'KeyName': fakes.NAME_KEY_PAIR})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidKeyPair.Duplicate', resp['Error']['Code'])
resp = self.execute('CreateKeyPair', {'KeyName': 'k' * 256})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('ValidationError', resp['Error']['Code'])
self.assert_execution_error(
'InvalidKeyPair.Duplicate', 'CreateKeyPair',
{'KeyName': fakes.NAME_KEY_PAIR})
self.assert_execution_error(
'ValidationError', 'CreateKeyPair', {'KeyName': 'k' * 256})
self.nova.keypairs.create.side_effect = (
nova_exception.OverLimit(413))
resp = self.execute('CreateKeyPair', {'KeyName': fakes.NAME_KEY_PAIR})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('ResourceLimitExceeded', resp['Error']['Code'])
self.assert_execution_error(
'ResourceLimitExceeded', 'CreateKeyPair',
{'KeyName': fakes.NAME_KEY_PAIR})
def test_import_key_pair(self):
self.nova.keypairs.create.return_value = (
@ -55,38 +52,32 @@ class KeyPairCase(base.ApiTestCase):
{'KeyName': fakes.NAME_KEY_PAIR,
'PublicKeyMaterial': base64.b64encode(
fakes.PUBLIC_KEY_KEY_PAIR)})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(
tools.purge_dict(fakes.EC2_KEY_PAIR, {'keyMaterial'}),
matchers.DictMatches(tools.purge_dict(resp, {'http_status_code'})))
matchers.DictMatches(resp))
self.nova.keypairs.create.assert_called_once_with(
fakes.NAME_KEY_PAIR, fakes.PUBLIC_KEY_KEY_PAIR)
def test_import_key_pair_invalid(self):
self.nova.keypairs.create.side_effect = (
nova_exception.OverLimit(413))
resp = self.execute('ImportKeyPair',
{'KeyName': fakes.NAME_KEY_PAIR,
'PublicKeyMaterial': base64.b64encode(
fakes.PUBLIC_KEY_KEY_PAIR)})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('ResourceLimitExceeded', resp['Error']['Code'])
self.assert_execution_error(
'ResourceLimitExceeded', 'ImportKeyPair',
{'KeyName': fakes.NAME_KEY_PAIR,
'PublicKeyMaterial': base64.b64encode(fakes.PUBLIC_KEY_KEY_PAIR)})
def test_delete_key_pair(self):
self.nova.keypairs.delete.return_value = True
resp = self.execute('DeleteKeyPair', {'KeyName': fakes.NAME_KEY_PAIR})
self.assertEqual(200, resp['http_status_code'])
self.execute('DeleteKeyPair', {'KeyName': fakes.NAME_KEY_PAIR})
self.nova.keypairs.delete.assert_called_once_with(fakes.NAME_KEY_PAIR)
self.nova.keypairs.delete.side_effect = nova_exception.NotFound(404)
resp = self.execute('DeleteKeyPair', {'KeyName': 'keyname1'})
self.assertEqual(200, resp['http_status_code'])
self.execute('DeleteKeyPair', {'KeyName': 'keyname1'})
self.nova.keypairs.delete.assert_any_call('keyname1')
def test_describe_key_pairs(self):
self.nova.keypairs.list.return_value = [fakes.NovaKeyPair(
fakes.OS_KEY_PAIR)]
resp = self.execute('DescribeKeyPairs', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['keySet'],
matchers.ListMatches([
tools.purge_dict(fakes.EC2_KEY_PAIR,
@ -101,7 +92,7 @@ class KeyPairCase(base.ApiTestCase):
def test_describe_key_pairs_invalid(self):
self.nova.keypairs.list.return_value = [fakes.NovaKeyPair(
fakes.OS_KEY_PAIR)]
resp = self.execute('DescribeKeyPairs', {'KeyName.1': 'badname'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidKeyPair.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidKeyPair.NotFound', 'DescribeKeyPairs',
{'KeyName.1': 'badname'})
self.nova.keypairs.list.assert_called_once_with()

157
ec2api/tests/unit/test_network_interface.py
View File

@ -34,7 +34,6 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
self.neutron.create_port.return_value = {'port': fakes.OS_PORT_1}
def check_response(resp, auto_ips=False):
self.assertEqual(200, resp['http_status_code'])
self.assertThat(fakes.EC2_NETWORK_INTERFACE_1,
matchers.DictMatches(resp['networkInterface']))
self.db_api.add_item.assert_called_once_with(
@ -111,7 +110,6 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
['association'])
def check_response(resp):
self.assertEqual(200, resp['http_status_code'])
self.assertThat(created_ec2_network_interface,
matchers.DictMatches(resp['networkInterface']))
self.db_api.add_item.assert_called_once_with(
@ -206,45 +204,37 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
check_response(resp)
def test_create_network_interface_invalid_parameters(self):
def check_response(resp, error_code):
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
def do_check(args, error_code):
self.neutron.reset_mock()
self.db_api.reset_mock()
self.assert_execution_error(
error_code, 'CreateNetworkInterface', args)
self.set_mock_db_items()
resp = self.execute(
'CreateNetworkInterface',
{'SubnetId': fakes.ID_EC2_SUBNET_2})
do_check({'SubnetId': fakes.ID_EC2_SUBNET_2},
'InvalidSubnetID.NotFound')
self.db_api.get_item_by_id.assert_called_once_with(
mock.ANY, fakes.ID_EC2_SUBNET_2)
check_response(resp, 'InvalidSubnetID.NotFound')
self.set_mock_db_items(fakes.DB_SUBNET_1, fakes.DB_VPC_1)
self.neutron.show_subnet.return_value = {'subnet': fakes.OS_SUBNET_1}
resp = self.execute(
'CreateNetworkInterface',
{'SubnetId': fakes.ID_EC2_SUBNET_1,
'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_2})
check_response(resp, 'InvalidParameterValue')
do_check({'SubnetId': fakes.ID_EC2_SUBNET_1,
'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_2},
'InvalidParameterValue')
for cls in [neutron_exception.OverQuotaClient,
neutron_exception.IpAddressGenerationFailureClient]:
self.neutron.create_port.side_effect = cls()
resp = self.execute(
'CreateNetworkInterface',
{'SubnetId': fakes.ID_EC2_SUBNET_1,
'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_1})
check_response(resp, 'NetworkInterfaceLimitExceeded')
do_check({'SubnetId': fakes.ID_EC2_SUBNET_1,
'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_1},
'NetworkInterfaceLimitExceeded')
for cls in [neutron_exception.IpAddressInUseClient,
neutron_exception.BadRequest]:
self.neutron.create_port.side_effect = cls()
resp = self.execute(
'CreateNetworkInterface',
{'SubnetId': fakes.ID_EC2_SUBNET_1,
'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_1})
check_response(resp, 'InvalidParameterValue')
do_check({'SubnetId': fakes.ID_EC2_SUBNET_1,
'PrivateIpAddress': fakes.IP_NETWORK_INTERFACE_1},
'InvalidParameterValue')
@mock.patch('ec2api.api.dhcp_options._add_dhcp_opts_to_port')
def test_create_network_interface_rollback(self, _add_dhcp_opts_to_port):
@ -258,8 +248,9 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
self.neutron.create_port.return_value = {'port': fakes.OS_PORT_1}
_add_dhcp_opts_to_port.side_effect = Exception()
self.execute('CreateNetworkInterface',
{'SubnetId': fakes.ID_EC2_SUBNET_1})
self.assert_execution_error(self.ANY_EXECUTE_ERROR,
'CreateNetworkInterface',
{'SubnetId': fakes.ID_EC2_SUBNET_1})
self.neutron.delete_port.assert_called_once_with(fakes.ID_OS_PORT_1)
self.db_api.delete_item.assert_called_once_with(
@ -269,9 +260,7 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1)
resp = self.execute(
'DeleteNetworkInterface',
{'NetworkInterfaceId':
fakes.ID_EC2_NETWORK_INTERFACE_1})
self.assertEqual(200, resp['http_status_code'])
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
self.assertEqual(True, resp['return'])
self.db_api.get_item_by_id.assert_any_call(
mock.ANY,
@ -288,30 +277,21 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
neutron_exception.PortNotFoundClient())
resp = self.execute(
'DeleteNetworkInterface',
{'NetworkInterfaceId':
fakes.ID_EC2_NETWORK_INTERFACE_1})
self.assertEqual(200, resp['http_status_code'])
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
self.assertEqual(True, resp['return'])
def test_delete_network_interface_no_network_interface(self):
self.set_mock_db_items()
resp = self.execute(
'DeleteNetworkInterface',
{'NetworkInterfaceId':
fakes.ID_EC2_NETWORK_INTERFACE_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidNetworkInterfaceID.NotFound',
resp['Error']['Code'])
self.assert_execution_error(
'InvalidNetworkInterfaceID.NotFound', 'DeleteNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
self.assertEqual(0, self.neutron.delete_port.call_count)
def test_delete_network_interface_is_in_use(self):
self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_2)
resp = self.execute(
'DeleteNetworkInterface',
{'NetworkInterfaceId':
fakes.ID_EC2_NETWORK_INTERFACE_2})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error(
'InvalidParameterValue', 'DeleteNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2})
self.assertEqual(0, self.neutron.delete_port.call_count)
def test_delete_network_interface_with_public_ip(self):
@ -326,7 +306,6 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
resp = self.execute(
'DeleteNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.db_api.get_item_by_id.assert_any_call(
mock.ANY,
@ -346,8 +325,9 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1)
self.neutron.delete_port.side_effect = Exception()
self.execute('DeleteNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'DeleteNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1})
self.db_api.restore_item.assert_called_once_with(
mock.ANY, 'eni', fakes.DB_NETWORK_INTERFACE_1)
@ -364,7 +344,6 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
fakes.OS_FLOATING_IP_2]})
resp = self.execute('DescribeNetworkInterfaces', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['networkInterfaceSet'],
matchers.ListMatches(
[fakes.EC2_NETWORK_INTERFACE_1,
@ -377,7 +356,6 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
resp = self.execute(
'DescribeNetworkInterfaces',
{'NetworkInterfaceId.1': fakes.ID_EC2_NETWORK_INTERFACE_1})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['networkInterfaceSet'],
matchers.ListMatches(
[fakes.EC2_NETWORK_INTERFACE_1]))
@ -417,7 +395,6 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
'DescribeNetworkInterfaceAttribute',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
'Attribute': 'description'})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(fakes.ID_EC2_NETWORK_INTERFACE_1,
resp['networkInterfaceId'])
self.assertEqual(fakes.DESCRIPTION_NETWORK_INTERFACE_1,
@ -426,33 +403,28 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
def test_modify_network_interface_attribute(self):
self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1)
resp = self.execute(
self.execute(
'ModifyNetworkInterfaceAttribute',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
'Description.Value': 'New description'})
self.assertEqual(200, resp['http_status_code'])
self.db_api.update_item.assert_called_once_with(
mock.ANY,
tools.update_dict(fakes.DB_NETWORK_INTERFACE_1,
{'description': 'New description'}))
def test_modify_network_interface_attribute_invalid_parameters(self):
resp = self.execute(
'ModifyNetworkInterfaceAttribute',
self.assert_execution_error(
'InvalidParameterCombination', 'ModifyNetworkInterfaceAttribute',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
'Description.Value': 'New description',
'SourceDestCheck.Value': 'True'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterCombination',
resp['Error']['Code'])
def test_reset_network_interface_attribute(self):
resp = self.execute(
self.execute(
'ResetNetworkInterfaceAttribute',
{'NetworkInterfaceId':
fakes.ID_EC2_NETWORK_INTERFACE_1,
'Attribute': 'sourceDestCheck'})
self.assertEqual(200, resp['http_status_code'])
def test_attach_network_interface(self):
self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1,
@ -460,12 +432,11 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
self.neutron.list_ports.return_value = (
{'ports': [fakes.OS_PORT_1]})
self.isotime.return_value = fakes.TIME_ATTACH_NETWORK_INTERFACE
resp = self.execute(
self.execute(
'AttachNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1,
'DeviceIndex': '1'})
self.assertEqual(200, resp['http_status_code'])
self.nova.servers.interface_attach.assert_called_once_with(
fakes.ID_OS_INSTANCE_1, fakes.ID_OS_PORT_1, None, None)
self.db_api.update_item.assert_called_once_with(
@ -480,27 +451,21 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
def test_attach_network_interface_invalid_parameters(self):
# NOTE(ft): eni is already attached
self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_2)
resp = self.execute(
'AttachNetworkInterface',
self.assert_execution_error(
'InvalidParameterValue', 'AttachNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2,
'InstanceId': fakes.ID_EC2_INSTANCE_2,
'DeviceIndex': '1'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue',
resp['Error']['Code'])
# NOTE(ft): device index is in use
self.set_mock_db_items(
fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2,
fakes.DB_INSTANCE_1)
resp = self.execute(
'AttachNetworkInterface',
self.assert_execution_error(
'InvalidParameterValue', 'AttachNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1,
'DeviceIndex': '0'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue',
resp['Error']['Code'])
def test_attach_network_interface_rollback(self):
self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_1,
@ -510,10 +475,11 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
self.isotime.return_value = fakes.TIME_ATTACH_NETWORK_INTERFACE
self.nova.servers.interface_attach.side_effect = Exception()
self.execute('AttachNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1,
'DeviceIndex': '1'})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'AttachNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
'InstanceId': fakes.ID_EC2_INSTANCE_1,
'DeviceIndex': '1'})
self.db_api.update_item.assert_any_call(
mock.ANY, fakes.DB_NETWORK_INTERFACE_1)
@ -524,11 +490,10 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
self.set_mock_db_items(network_interface)
self.neutron.list_ports.return_value = (
{'ports': [fakes.OS_PORT_2]})
resp = self.execute(
self.execute(
'DetachNetworkInterface',
{'AttachmentId': ec2utils.change_ec2_id_kind(
fakes.ID_EC2_NETWORK_INTERFACE_2, 'eni-attach')})
self.assertEqual(200, resp['http_status_code'])
self.neutron.update_port.assert_called_once_with(
fakes.ID_OS_PORT_2,
{'port': {'device_id': '',
@ -545,23 +510,17 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
def test_detach_network_interface_invalid_parameters(self):
# NOTE(ft): eni is not found
self.set_mock_db_items()
resp = self.execute(
'DetachNetworkInterface',
self.assert_execution_error(
'InvalidAttachmentID.NotFound', 'DetachNetworkInterface',
{'AttachmentId': ec2utils.change_ec2_id_kind(
fakes.ID_EC2_NETWORK_INTERFACE_2, 'eni-attach')})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidAttachmentID.NotFound',
resp['Error']['Code'])
# NOTE(ft): eni is attached with device index = 0
self.set_mock_db_items(fakes.DB_NETWORK_INTERFACE_2)
resp = self.execute(
'DetachNetworkInterface',
self.assert_execution_error(
'OperationNotPermitted', 'DetachNetworkInterface',
{'AttachmentId': ec2utils.change_ec2_id_kind(
fakes.ID_EC2_NETWORK_INTERFACE_2, 'eni-attach')})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('OperationNotPermitted',
resp['Error']['Code'])
def test_detach_network_interface_rollback(self):
network_interface = tools.update_dict(fakes.DB_NETWORK_INTERFACE_2,
@ -571,8 +530,8 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
{'ports': [fakes.OS_PORT_2]})
self.neutron.update_port.side_effect = Exception()
self.execute(
'DetachNetworkInterface',
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'DetachNetworkInterface',
{'AttachmentId': fakes.ID_EC2_NETWORK_INTERFACE_2_ATTACH})
self.db_api.update_item.assert_any_call(
@ -584,13 +543,12 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
{'subnet': fakes.OS_SUBNET_1})
self.neutron.show_port.return_value = (
{'port': copy.deepcopy(fakes.OS_PORT_1)})
resp = self.execute(
self.execute(
'AssignPrivateIpAddresses',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
'PrivateIpAddress.1': '10.10.1.5',
'PrivateIpAddress.2': '10.10.1.6',
})
self.assertEqual(200, resp['http_status_code'])
self.neutron.update_port.assert_called_once_with(
fakes.ID_OS_PORT_1,
{'port':
@ -599,13 +557,12 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
'ip_address': fakes.IP_NETWORK_INTERFACE_1},
{'ip_address': '10.10.1.5'},
{'ip_address': '10.10.1.6'}]}})
resp = self.execute(
self.execute(
'UnassignPrivateIpAddresses',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
'PrivateIpAddress.1': '10.10.1.5',
'PrivateIpAddress.2': '10.10.1.6',
})
self.assertEqual(200, resp['http_status_code'])
self.neutron.update_port.assert_any_call(
fakes.ID_OS_PORT_1,
{'port':
@ -621,14 +578,12 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
{'port': copy.deepcopy(fakes.OS_PORT_1)})
def do_check(error_code):
resp = self.execute(
'AssignPrivateIpAddresses',
self.assert_execution_error(
error_code, 'AssignPrivateIpAddresses',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,
'PrivateIpAddress.1': '10.10.1.5',
'PrivateIpAddress.2': '10.10.1.6',
})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
self.neutron.update_port.side_effect = (
neutron_exception.IpAddressGenerationFailureClient())
@ -649,9 +604,7 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
self.neutron.show_port.return_value = (
{'port': copy.deepcopy(fakes.OS_PORT_2)})
resp = self.execute(
'UnassignPrivateIpAddresses',
self.assert_execution_error(
'InvalidParameterValue', 'UnassignPrivateIpAddresses',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_2,
'PrivateIpAddress.1': '10.10.2.55'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])

142
ec2api/tests/unit/test_route_table.py
View File

@ -33,7 +33,6 @@ class RouteTableTestCase(base.ApiTestCase):
tools.get_db_api_add_item(fakes.ID_EC2_ROUTE_TABLE_1))
resp = self.execute('CreateRouteTable',
{'VpcId': fakes.ID_EC2_VPC_1})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(
resp['routeTable'],
matchers.DictMatches(tools.purge_dict(fakes.EC2_ROUTE_TABLE_1,
@ -49,10 +48,9 @@ class RouteTableTestCase(base.ApiTestCase):
def test_route_table_create_invalid_parameters(self):
self.set_mock_db_items()
resp = self.execute('CreateRouteTable',
{'VpcId': fakes.ID_EC2_VPC_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidVpcID.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidVpcID.NotFound', 'CreateRouteTable',
{'VpcId': fakes.ID_EC2_VPC_1})
@mock.patch('ec2api.api.route_table._update_routes_in_associated_subnets')
def test_create_route(self, routes_updater):
@ -63,7 +61,6 @@ class RouteTableTestCase(base.ApiTestCase):
def do_check(params, route_table, rollback_route_table_state):
resp = self.execute('CreateRoute', params)
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.db_api.update_item.assert_called_once_with(
@ -131,9 +128,7 @@ class RouteTableTestCase(base.ApiTestCase):
fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2)
def do_check(params, error_code):
resp = self.execute('CreateRoute', params)
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
self.assert_execution_error(error_code, 'CreateRoute', params)
do_check({'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1,
'DestinationCidrBlock': 'not_a_cidr',
@ -214,18 +209,20 @@ class RouteTableTestCase(base.ApiTestCase):
fakes.gen_db_igw(fakes.ID_EC2_IGW_2, fakes.ID_EC2_VPC_1))
routes_updater.side_effect = Exception()
self.execute('CreateRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1,
'DestinationCidrBlock': '0.0.0.0/0',
'GatewayId': fakes.ID_EC2_IGW_1})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'CreateRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1,
'DestinationCidrBlock': '0.0.0.0/0',
'GatewayId': fakes.ID_EC2_IGW_1})
self.db_api.update_item.assert_any_call(mock.ANY,
fakes.DB_ROUTE_TABLE_1)
self.execute('ReplaceRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock': '0.0.0.0/0',
'GatewayId': fakes.ID_EC2_IGW_2})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'ReplaceRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock': '0.0.0.0/0',
'GatewayId': fakes.ID_EC2_IGW_2})
self.db_api.update_item.assert_any_call(mock.ANY,
fakes.DB_ROUTE_TABLE_2)
@ -244,7 +241,6 @@ class RouteTableTestCase(base.ApiTestCase):
'DestinationCidrBlock': '0.0.0.0/0',
'NetworkInterfaceId':
fakes.ID_EC2_NETWORK_INTERFACE_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
rollback_route_table_state = route_table
@ -260,13 +256,11 @@ class RouteTableTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_ROUTE_TABLE_1,
fakes.DB_VPC_1, fakes.DB_IGW_1)
resp = self.execute(
'ReplaceRoute',
self.assert_execution_error(
'InvalidParameterValue', 'ReplaceRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1,
'DestinationCidrBlock': '11.22.33.0/24',
'GatewayId': fakes.ID_EC2_IGW_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
@mock.patch('ec2api.api.route_table._update_routes_in_associated_subnets')
def test_delete_route(self, routes_updater):
@ -275,7 +269,6 @@ class RouteTableTestCase(base.ApiTestCase):
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock':
fakes.CIDR_EXTERNAL_NETWORK})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
route_table = copy.deepcopy(fakes.DB_ROUTE_TABLE_2)
route_table['routes'] = [
@ -287,33 +280,31 @@ class RouteTableTestCase(base.ApiTestCase):
def test_delete_route_invalid_parameters(self):
self.set_mock_db_items()
resp = self.execute('DeleteRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock': '11.22.33.0/24'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidRouteTableID.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidRouteTableID.NotFound', 'DeleteRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock': '11.22.33.0/24'})
self.set_mock_db_items(fakes.DB_ROUTE_TABLE_2)
resp = self.execute('DeleteRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock': '11.22.33.0/24'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidRoute.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidRoute.NotFound', 'DeleteRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock': '11.22.33.0/24'})
resp = self.execute('DeleteRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock': fakes.CIDR_VPC_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error(
'InvalidParameterValue', 'DeleteRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock': fakes.CIDR_VPC_1})
@mock.patch('ec2api.api.route_table._update_routes_in_associated_subnets')
def test_delete_route_rollback(self, routes_updater):
self.set_mock_db_items(fakes.DB_ROUTE_TABLE_2)
routes_updater.side_effect = Exception()
self.execute('DeleteRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock': fakes.CIDR_EXTERNAL_NETWORK})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'DeleteRoute',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2,
'DestinationCidrBlock': fakes.CIDR_EXTERNAL_NETWORK})
self.db_api.update_item.assert_any_call(
mock.ANY, fakes.DB_ROUTE_TABLE_2)
@ -325,7 +316,6 @@ class RouteTableTestCase(base.ApiTestCase):
resp = self.execute('AssociateRouteTable',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1,
'SubnetId': fakes.ID_EC2_SUBNET_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(fakes.ID_EC2_SUBNET_1.replace('subnet', 'rtbassoc'),
resp['associationId'])
subnet = tools.update_dict(
@ -339,9 +329,8 @@ class RouteTableTestCase(base.ApiTestCase):
def test_associate_route_table_invalid_parameters(self):
def do_check(params, error_code):
resp = self.execute('AssociateRouteTable', params)
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
self.assert_execution_error(
error_code, 'AssociateRouteTable', params)
self.set_mock_db_items()
do_check({'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1,
@ -376,9 +365,10 @@ class RouteTableTestCase(base.ApiTestCase):
fakes.DB_SUBNET_1)
routes_updater.side_effect = Exception()
self.execute('AssociateRouteTable',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1,
'SubnetId': fakes.ID_EC2_SUBNET_1})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'AssociateRouteTable',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1,
'SubnetId': fakes.ID_EC2_SUBNET_1})
self.db_api.update_item.assert_any_call(mock.ANY, fakes.DB_SUBNET_1)
@ -390,7 +380,6 @@ class RouteTableTestCase(base.ApiTestCase):
'ReplaceRouteTableAssociation',
{'AssociationId': fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_3,
'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_2,
resp['newAssociationId'])
subnet = tools.update_dict(
@ -410,7 +399,6 @@ class RouteTableTestCase(base.ApiTestCase):
{'AssociationId':
fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_1,
'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_1,
resp['newAssociationId'])
vpc = tools.update_dict(
@ -424,9 +412,8 @@ class RouteTableTestCase(base.ApiTestCase):
def test_replace_route_table_association_invalid_parameters(self):
def do_check(params, error_code):
resp = self.execute('ReplaceRouteTableAssociation', params)
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
self.assert_execution_error(
error_code, 'ReplaceRouteTableAssociation', params)
self.set_mock_db_items()
do_check({'AssociationId': fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_1,
@ -475,18 +462,20 @@ class RouteTableTestCase(base.ApiTestCase):
fakes.DB_VPC_1)
multiply_routes_updater.side_effect = Exception()
self.execute('ReplaceRouteTableAssociation',
{'AssociationId': fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_1,
'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'ReplaceRouteTableAssociation',
{'AssociationId': fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_1,
'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2})
self.db_api.update_item.assert_any_call(
mock.ANY, fakes.DB_VPC_1)
self.db_api.reset_mock()
routes_updater.side_effect = Exception()
self.execute('ReplaceRouteTableAssociation',
{'AssociationId': fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_3,
'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2})
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'ReplaceRouteTableAssociation',
{'AssociationId': fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_3,
'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2})
self.db_api.update_item.assert_any_call(
mock.ANY, fakes.DB_SUBNET_2)
@ -498,7 +487,6 @@ class RouteTableTestCase(base.ApiTestCase):
resp = self.execute(
'DisassociateRouteTable',
{'AssociationId': fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_3})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
subnet = tools.purge_dict(fakes.DB_SUBNET_2, ('route_table_id',))
self.db_api.update_item.assert_called_once_with(
@ -510,9 +498,8 @@ class RouteTableTestCase(base.ApiTestCase):
def test_disassociate_route_table_invalid_parameter(self):
def do_check(params, error_code):
resp = self.execute('DisassociateRouteTable', params)
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
self.assert_execution_error(
error_code, 'DisassociateRouteTable', params)
self.set_mock_db_items()
do_check({'AssociationId': fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_1},
@ -533,8 +520,8 @@ class RouteTableTestCase(base.ApiTestCase):
fakes.DB_SUBNET_2, fakes.DB_VPC_1)
routes_updater.side_effect = Exception()
self.execute(
'DisassociateRouteTable',
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'DisassociateRouteTable',
{'AssociationId': fakes.ID_EC2_ROUTE_TABLE_ASSOCIATION_3})
self.db_api.update_item.assert_any_call(
@ -545,7 +532,6 @@ class RouteTableTestCase(base.ApiTestCase):
fakes.DB_SUBNET_1, fakes.DB_SUBNET_2)
resp = self.execute('DeleteRouteTable',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.db_api.delete_item.assert_called_once_with(
mock.ANY,
@ -553,25 +539,22 @@ class RouteTableTestCase(base.ApiTestCase):
def test_delete_route_table_invalid_parameters(self):
self.set_mock_db_items()
resp = self.execute('DeleteRouteTable',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidRouteTableID.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidRouteTableID.NotFound', 'DeleteRouteTable',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1})
self.set_mock_db_items(fakes.DB_ROUTE_TABLE_1, fakes.DB_VPC_1)
resp = self.execute('DeleteRouteTable',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('DependencyViolation', resp['Error']['Code'])
self.assert_execution_error(
'DependencyViolation', 'DeleteRouteTable',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_1})
subnet = tools.update_dict(
fakes.DB_SUBNET_2,
{'route_table_id': fakes.ID_EC2_ROUTE_TABLE_2})
self.set_mock_db_items(fakes.DB_ROUTE_TABLE_2, fakes.DB_VPC_1, subnet)
resp = self.execute('DeleteRouteTable',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('DependencyViolation', resp['Error']['Code'])
self.assert_execution_error(
'DependencyViolation', 'DeleteRouteTable',
{'RouteTableId': fakes.ID_EC2_ROUTE_TABLE_2})
def test_describe_route_tables(self):
self.set_mock_db_items(
@ -584,7 +567,6 @@ class RouteTableTestCase(base.ApiTestCase):
mock.NonCallableMock(status='ACTIVE'))
resp = self.execute('DescribeRouteTables', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['routeTableSet'],
matchers.ListMatches([fakes.EC2_ROUTE_TABLE_1,
fakes.EC2_ROUTE_TABLE_2,
@ -592,7 +574,6 @@ class RouteTableTestCase(base.ApiTestCase):
resp = self.execute('DescribeRouteTables',
{'RouteTableId.1': fakes.ID_EC2_ROUTE_TABLE_1})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['routeTableSet'],
matchers.ListMatches([fakes.EC2_ROUTE_TABLE_1]))
self.db_api.get_items_by_ids.assert_called_once_with(
@ -649,7 +630,6 @@ class RouteTableTestCase(base.ApiTestCase):
self.nova.servers.get.return_value = (
mock.NonCallableMock(status='DOWN'))
resp = self.execute('DescribeRouteTables', {})
self.assertEqual(200, resp['http_status_code'])
ec2_route_table_1 = copy.deepcopy(fakes.EC2_ROUTE_TABLE_1)
ec2_route_table_1['routeSet'].append({
'destinationCidrBlock': '0.0.0.0/0',

177
ec2api/tests/unit/test_security_group.py
View File

@ -38,7 +38,6 @@ class SecurityGroupTestCase(base.ApiTestCase):
'CreateSecurityGroup',
{'GroupName': 'groupname',
'GroupDescription': 'Group description'})
self.assertEqual(200, resp['http_status_code'])
self.nova.security_groups.create.assert_called_once_with(
'groupname', 'Group description')
self.nova.security_groups.reset_mock()
@ -48,7 +47,6 @@ class SecurityGroupTestCase(base.ApiTestCase):
{'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'groupname',
'GroupDescription': 'Group description'})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(fakes.ID_EC2_SECURITY_GROUP_1, resp['groupId'])
self.db_api.add_item.assert_called_once_with(
mock.ANY, 'sg',
@ -60,82 +58,62 @@ class SecurityGroupTestCase(base.ApiTestCase):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNeutron())
def check_response(resp, error_code):
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
def do_check(args, error_code):
self.neutron.reset_mock()
self.db_api.reset_mock()
self.assert_execution_error(
error_code, 'CreateSecurityGroup', args)
self.set_mock_db_items()
resp = self.execute(
'CreateSecurityGroup',
{'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'groupname',
'GroupDescription': 'Group description'})
do_check({'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'groupname',
'GroupDescription': 'Group description'},
'InvalidVpcID.NotFound')
self.db_api.get_item_by_id.assert_called_once_with(mock.ANY,
fakes.ID_EC2_VPC_1)
check_response(resp, 'InvalidVpcID.NotFound')
resp = self.execute(
'CreateSecurityGroup',
{'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'aa #^% -=99',
'GroupDescription': 'Group description'})
check_response(resp, 'ValidationError')
do_check({'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'aa #^% -=99',
'GroupDescription': 'Group description'},
'ValidationError')
resp = self.execute(
'CreateSecurityGroup',
{'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'groupname',
'GroupDescription': 'aa #^% -=99'})
check_response(resp, 'ValidationError')
do_check({'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'groupname',
'GroupDescription': 'aa #^% -=99'},
'ValidationError')
resp = self.execute(
'CreateSecurityGroup',
{'GroupName': 'aa \t\x01\x02\x7f',
'GroupDescription': 'Group description'})
check_response(resp, 'ValidationError')
do_check({'GroupName': 'aa \t\x01\x02\x7f',
'GroupDescription': 'Group description'},
'ValidationError')
resp = self.execute(
'CreateSecurityGroup',
{'GroupName': 'groupname',
'GroupDescription': 'aa \t\x01\x02\x7f'})
check_response(resp, 'ValidationError')
do_check({'GroupName': 'groupname',
'GroupDescription': 'aa \t\x01\x02\x7f'},
'ValidationError')
resp = self.execute(
'CreateSecurityGroup',
{'GroupName': 'x' * 256,
'GroupDescription': 'Group description'})
check_response(resp, 'ValidationError')
do_check({'GroupName': 'x' * 256,
'GroupDescription': 'Group description'},
'ValidationError')
resp = self.execute(
'CreateSecurityGroup',
{'GroupName': 'groupname',
'GroupDescription': 'x' * 256})
check_response(resp, 'ValidationError')
do_check({'GroupName': 'groupname',
'GroupDescription': 'x' * 256},
'ValidationError')
resp = self.execute(
'CreateSecurityGroup',
{'GroupName': 'groupname'})
check_response(resp, 'MissingParameter')
do_check({'GroupName': 'groupname'},
'MissingParameter')
resp = self.execute(
'CreateSecurityGroup',
{'GroupDescription': 'description'})
check_response(resp, 'MissingParameter')
do_check({'GroupDescription': 'description'},
'MissingParameter')
def test_create_security_group_over_quota(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNeutron())
self.nova.security_groups.create.side_effect = (
nova_exception.OverLimit(413))
resp = self.execute(
'CreateSecurityGroup',
self.assert_execution_error(
'ResourceLimitExceeded', 'CreateSecurityGroup',
{'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'groupname',
'GroupDescription': 'Group description'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('ResourceLimitExceeded', resp['Error']['Code'])
self.nova.security_groups.create.assert_called_once_with(
'groupname', 'Group description')
@ -146,8 +124,8 @@ class SecurityGroupTestCase(base.ApiTestCase):
self.db_api.add_item.side_effect = Exception()
self.nova.security_groups.create.return_value = (
fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_1))
self.execute(
'CreateSecurityGroup',
self.assert_execution_error(
self.ANY_EXECUTE_ERROR, 'CreateSecurityGroup',
{'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'groupname',
'GroupDescription': 'Group description'})
@ -162,7 +140,6 @@ class SecurityGroupTestCase(base.ApiTestCase):
'DeleteSecurityGroup',
{'GroupId':
fakes.ID_EC2_SECURITY_GROUP_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.db_api.get_item_by_id.assert_any_call(
mock.ANY,
@ -183,7 +160,6 @@ class SecurityGroupTestCase(base.ApiTestCase):
'DeleteSecurityGroup',
{'GroupName':
fakes.EC2_SECURITY_GROUP_2['groupName']})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.nova.security_groups.delete.assert_called_once_with(
fakes.ID_OS_SECURITY_GROUP_2)
@ -200,7 +176,6 @@ class SecurityGroupTestCase(base.ApiTestCase):
'DeleteSecurityGroup',
{'GroupId':
fakes.ID_OS_SECURITY_GROUP_2})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.nova.security_groups.delete.assert_called_once_with(
fakes.ID_OS_SECURITY_GROUP_2)
@ -209,27 +184,16 @@ class SecurityGroupTestCase(base.ApiTestCase):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNeutron())
self.set_mock_db_items()
resp = self.execute(
'DeleteSecurityGroup',
{'GroupId':
fakes.ID_EC2_SECURITY_GROUP_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidGroup.NotFound',
resp['Error']['Code'])
self.assert_execution_error(
'InvalidGroup.NotFound', 'DeleteSecurityGroup',
{'GroupId': fakes.ID_EC2_SECURITY_GROUP_1})
self.assertEqual(0, self.neutron.delete_port.call_count)
resp = self.execute(
'DeleteSecurityGroup',
{'GroupName':
'badname'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidGroup.NotFound',
resp['Error']['Code'])
self.assert_execution_error(
'InvalidGroup.NotFound', 'DeleteSecurityGroup',
{'GroupName': 'badname'})
self.assertEqual(0, self.neutron.delete_port.call_count)
resp = self.execute(
'DeleteSecurityGroup', {})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('MissingParameter',
resp['Error']['Code'])
self.assert_execution_error(
'MissingParameter', 'DeleteSecurityGroup', {})
self.assertEqual(0, self.neutron.delete_port.call_count)
def test_delete_security_group_is_in_use(self):
@ -238,12 +202,9 @@ class SecurityGroupTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_SECURITY_GROUP_1)
self.neutron.delete_security_group.side_effect = (
neutron_exception.Conflict())
resp = self.execute(
'DeleteSecurityGroup',
{'GroupId':
fakes.ID_EC2_SECURITY_GROUP_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('DependencyViolation', resp['Error']['Code'])
self.assert_execution_error(
'DependencyViolation', 'DeleteSecurityGroup',
{'GroupId': fakes.ID_EC2_SECURITY_GROUP_1})
self.assertEqual(0, self.db_api.delete_item.call_count)
def test_describe_security_groups(self):
@ -256,7 +217,6 @@ class SecurityGroupTestCase(base.ApiTestCase):
fakes.OS_SECURITY_GROUP_2]})
resp = self.execute('DescribeSecurityGroups', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['securityGroupInfo'],
matchers.ListMatches(
[fakes.EC2_SECURITY_GROUP_1,
@ -265,7 +225,6 @@ class SecurityGroupTestCase(base.ApiTestCase):
resp = self.execute('DescribeSecurityGroups',
{'GroupName.1': 'groupname2'})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['securityGroupInfo'],
matchers.ListMatches(
[fakes.EC2_SECURITY_GROUP_2],
@ -275,7 +234,6 @@ class SecurityGroupTestCase(base.ApiTestCase):
return_value=[fakes.DB_SECURITY_GROUP_2])
resp = self.execute('DescribeSecurityGroups',
{'GroupId.1': fakes.ID_EC2_SECURITY_GROUP_2})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['securityGroupInfo'],
matchers.ListMatches(
[fakes.EC2_SECURITY_GROUP_2],
@ -301,7 +259,6 @@ class SecurityGroupTestCase(base.ApiTestCase):
[fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_1),
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2)])
resp = self.execute('DescribeSecurityGroups', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['securityGroupInfo'],
matchers.ListMatches(
[fakes.EC2_NOVA_SECURITY_GROUP_1,
@ -321,21 +278,18 @@ class SecurityGroupTestCase(base.ApiTestCase):
params['GroupId'] = group_id
if cidr is not None:
params['IpPermissions.1.IpRanges.1.CidrIp'] = cidr
resp = self.execute(
'AuthorizeSecurityGroupIngress', params)
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
self.assert_execution_error(
error_code, 'AuthorizeSecurityGroupIngress', params)
self.neutron.reset_mock()
self.db_api.reset_mock()
resp = self.execute(
self.execute(
'AuthorizeSecurityGroupIngress',
{'GroupId': fakes.ID_EC2_SECURITY_GROUP_2,
'IpPermissions.1.FromPort': '-1',
'IpPermissions.1.ToPort': '-1',
'IpPermissions.1.IpProtocol': 'icmp',
'IpPermissions.1.IpRanges.1.CidrIp': '0.0.0.0/0'})
self.assertEqual(200, resp['http_status_code'])
# Duplicate rule
self.set_mock_db_items(fakes.DB_SECURITY_GROUP_1,
fakes.DB_SECURITY_GROUP_2)
@ -375,14 +329,12 @@ class SecurityGroupTestCase(base.ApiTestCase):
# Missing cidr
check_response('MissingParameter', 'tcp', 1, 255, None)
# Invalid remote group
resp = self.execute(
'AuthorizeSecurityGroupIngress',
self.assert_execution_error(
'InvalidGroup.NotFound', 'AuthorizeSecurityGroupIngress',
{'GroupId': fakes.ID_EC2_SECURITY_GROUP_2,
'IpPermissions.1.IpProtocol': 'icmp',
'IpPermissions.1.Groups.1.GroupName': 'somegroup',
'IpPermissions.1.Groups.1.UserId': 'i-99999999'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidGroup.NotFound', resp['Error']['Code'])
def test_authorize_security_group_ingress_ip_ranges(self):
security_group.security_group_engine = (
@ -391,28 +343,26 @@ class SecurityGroupTestCase(base.ApiTestCase):
fakes.DB_SECURITY_GROUP_2)
self.neutron.create_security_group_rule.return_value = (
{'security_group_rule': [fakes.OS_SECURITY_GROUP_RULE_1]})
resp = self.execute(
self.execute(
'AuthorizeSecurityGroupIngress',
{'GroupId': fakes.ID_EC2_SECURITY_GROUP_2,
'IpPermissions.1.FromPort': '10',
'IpPermissions.1.ToPort': '10',
'IpPermissions.1.IpProtocol': 'tcp',
'IpPermissions.1.IpRanges.1.CidrIp': '192.168.1.0/24'})
self.assertEqual(200, resp['http_status_code'])
self.neutron.create_security_group_rule.assert_called_once_with(
{'security_group_rule':
tools.purge_dict(fakes.OS_SECURITY_GROUP_RULE_1,
{'id', 'remote_group_id', 'tenant_id'})})
# NOTE(Alex): Openstack extension, AWS-incompability
# IPv6 is not supported by Amazon.
resp = self.execute(
self.execute(
'AuthorizeSecurityGroupIngress',
{'GroupId': fakes.ID_EC2_SECURITY_GROUP_2,
'IpPermissions.1.FromPort': '10',
'IpPermissions.1.ToPort': '10',
'IpPermissions.1.IpProtocol': 'tcp',
'IpPermissions.1.IpRanges.1.CidrIp': '::/0'})
self.assertEqual(200, resp['http_status_code'])
self.neutron.create_security_group_rule.assert_called_with(
{'security_group_rule':
tools.patch_dict(
@ -427,14 +377,13 @@ class SecurityGroupTestCase(base.ApiTestCase):
self.nova.security_groups.list.return_value = (
[fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_1),
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2)])
resp = self.execute(
self.execute(
'AuthorizeSecurityGroupIngress',
{'GroupName': fakes.EC2_NOVA_SECURITY_GROUP_2['groupName'],
'IpPermissions.1.FromPort': '10',
'IpPermissions.1.ToPort': '10',
'IpPermissions.1.IpProtocol': 'tcp',
'IpPermissions.1.IpRanges.1.CidrIp': '192.168.1.0/24'})
self.assertEqual(200, resp['http_status_code'])
self.nova.security_group_rules.create.assert_called_once_with(
fakes.ID_OS_SECURITY_GROUP_2, 'tcp', 10, 10,
'192.168.1.0/24', None)
@ -446,14 +395,13 @@ class SecurityGroupTestCase(base.ApiTestCase):
fakes.DB_SECURITY_GROUP_2)
self.neutron.create_security_group_rule.return_value = (
{'security_group_rule': [fakes.OS_SECURITY_GROUP_RULE_1]})
resp = self.execute(
self.execute(
'AuthorizeSecurityGroupEgress',
{'GroupId': fakes.ID_EC2_SECURITY_GROUP_2,
'IpPermissions.1.FromPort': '10',
'IpPermissions.1.IpProtocol': '100',
'IpPermissions.1.Groups.1.GroupId':
fakes.ID_EC2_SECURITY_GROUP_1})
self.assertEqual(200, resp['http_status_code'])
self.neutron.create_security_group_rule.assert_called_once_with(
{'security_group_rule':
tools.purge_dict(fakes.OS_SECURITY_GROUP_RULE_2,
@ -468,13 +416,12 @@ class SecurityGroupTestCase(base.ApiTestCase):
self.nova.security_groups.list.return_value = (
[fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_1),
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2)])
resp = self.execute(
self.execute(
'AuthorizeSecurityGroupIngress',
{'GroupName': fakes.EC2_NOVA_SECURITY_GROUP_2['groupName'],
'IpPermissions.1.IpProtocol': 'icmp',
'IpPermissions.1.Groups.1.GroupName':
fakes.EC2_NOVA_SECURITY_GROUP_1['groupName']})
self.assertEqual(200, resp['http_status_code'])
self.nova.security_group_rules.create.assert_called_once_with(
fakes.ID_OS_SECURITY_GROUP_2, 'icmp', -1, -1,
None, fakes.ID_OS_SECURITY_GROUP_1)
@ -487,14 +434,13 @@ class SecurityGroupTestCase(base.ApiTestCase):
self.neutron.show_security_group.return_value = {
'security_group': fakes.OS_SECURITY_GROUP_2}
self.neutron.delete_security_group_rule.return_value = True
resp = self.execute(
self.execute(
'RevokeSecurityGroupIngress',
{'GroupId': fakes.ID_EC2_SECURITY_GROUP_2,
'IpPermissions.1.FromPort': '10',
'IpPermissions.1.ToPort': '10',
'IpPermissions.1.IpProtocol': 'tcp',
'IpPermissions.1.IpRanges.1.CidrIp': '192.168.1.0/24'})
self.assertEqual(200, resp['http_status_code'])
self.neutron.show_security_group.assert_called_once_with(
fakes.ID_OS_SECURITY_GROUP_2)
self.neutron.delete_security_group_rule.assert_called_once_with(
@ -509,14 +455,13 @@ class SecurityGroupTestCase(base.ApiTestCase):
self.nova.security_groups.get.return_value = (
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2))
self.nova.security_group_rules.delete.return_value = True
resp = self.execute(
self.execute(
'RevokeSecurityGroupIngress',
{'GroupName': fakes.EC2_NOVA_SECURITY_GROUP_2['groupName'],
'IpPermissions.1.FromPort': '10',
'IpPermissions.1.ToPort': '10',
'IpPermissions.1.IpProtocol': 'tcp',
'IpPermissions.1.IpRanges.1.CidrIp': '192.168.1.0/24'})
self.assertEqual(200, resp['http_status_code'])
self.nova.security_group_rules.delete.assert_called_once_with(
fakes.NOVA_SECURITY_GROUP_RULE_1['id'])
@ -528,14 +473,13 @@ class SecurityGroupTestCase(base.ApiTestCase):
self.neutron.show_security_group.return_value = {
'security_group': fakes.OS_SECURITY_GROUP_2}
self.neutron.delete_security_group_rule.return_value = True
resp = self.execute(
self.execute(
'RevokeSecurityGroupEgress',
{'GroupId': fakes.ID_EC2_SECURITY_GROUP_2,
'IpPermissions.1.FromPort': '10',
'IpPermissions.1.IpProtocol': '100',
'IpPermissions.1.Groups.1.GroupId':
fakes.ID_EC2_SECURITY_GROUP_1})
self.assertEqual(200, resp['http_status_code'])
self.neutron.show_security_group.assert_called_once_with(
fakes.ID_OS_SECURITY_GROUP_2)
self.neutron.delete_security_group_rule.assert_called_once_with(
@ -550,12 +494,11 @@ class SecurityGroupTestCase(base.ApiTestCase):
self.nova.security_groups.get.return_value = (
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2))
self.nova.security_group_rules.delete.return_value = True
resp = self.execute(
self.execute(
'RevokeSecurityGroupIngress',
{'GroupName': fakes.EC2_NOVA_SECURITY_GROUP_2['groupName'],
'IpPermissions.1.IpProtocol': 'icmp',
'IpPermissions.1.Groups.1.GroupName':
fakes.EC2_NOVA_SECURITY_GROUP_1['groupName']})
self.assertEqual(200, resp['http_status_code'])
self.nova.security_group_rules.delete.assert_called_once_with(
fakes.NOVA_SECURITY_GROUP_RULE_2['id'])

31
ec2api/tests/unit/test_snapshot.py
View File

@ -31,8 +31,6 @@ class SnapshotTestCase(base.ApiTestCase):
fakes.DB_VOLUME_2)
resp = self.execute('DescribeSnapshots', {})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'snapshotSet': [fakes.EC2_SNAPSHOT_1, fakes.EC2_SNAPSHOT_2]},
orderless_lists=True))
@ -43,8 +41,6 @@ class SnapshotTestCase(base.ApiTestCase):
return_value=[fakes.DB_SNAPSHOT_1])
resp = self.execute('DescribeSnapshots',
{'SnapshotId.1': fakes.ID_EC2_SNAPSHOT_1})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'snapshotSet': [fakes.EC2_SNAPSHOT_1]},
orderless_lists=True))
@ -76,8 +72,6 @@ class SnapshotTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_SNAPSHOT_1, fakes.DB_VOLUME_2)
resp = self.execute('DescribeSnapshots', {})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'snapshotSet': []},
orderless_lists=True))
@ -92,17 +86,15 @@ class SnapshotTestCase(base.ApiTestCase):
fakes.OSSnapshot(fakes.OS_SNAPSHOT_1),
fakes.OSSnapshot(fakes.OS_SNAPSHOT_2)]
resp = self.execute('DescribeSnapshots',
{'SnapshotId.1': fakes.random_ec2_id('snap')})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidSnapshot.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidSnapshot.NotFound', 'DescribeSnapshots',
{'SnapshotId.1': fakes.random_ec2_id('snap')})
self.cinder.volume_snapshots.list.side_effect = lambda: []
resp = self.execute('DescribeSnapshots',
{'SnapshotId.1': fakes.ID_EC2_SNAPSHOT_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidSnapshot.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidSnapshot.NotFound', 'DescribeSnapshots',
{'SnapshotId.1': fakes.ID_EC2_SNAPSHOT_1})
def test_create_snapshot_from_volume(self):
self.cinder.volume_snapshots.create.return_value = (
@ -119,9 +111,7 @@ class SnapshotTestCase(base.ApiTestCase):
resp = self.execute(
'CreateSnapshot',
{'VolumeId': fakes.ID_EC2_VOLUME_2})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(fakes.EC2_SNAPSHOT_1, matchers.DictMatches(
tools.purge_dict(resp, {'http_status_code'})))
self.assertThat(fakes.EC2_SNAPSHOT_1, matchers.DictMatches(resp))
self.db_api.add_item.assert_called_once_with(
mock.ANY, 'snap',
tools.purge_dict(fakes.DB_SNAPSHOT_1, ('id',)))
@ -136,35 +126,28 @@ class SnapshotTestCase(base.ApiTestCase):
fake_snapshot.status = 'new'
resp = self.execute('DescribeSnapshots', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('pending', resp['snapshotSet'][0]['status'])
fake_snapshot.status = 'creating'
resp = self.execute('DescribeSnapshots', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('pending', resp['snapshotSet'][0]['status'])
fake_snapshot.status = 'available'
resp = self.execute('DescribeSnapshots', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('completed', resp['snapshotSet'][0]['status'])
fake_snapshot.status = 'active'
resp = self.execute('DescribeSnapshots', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('completed', resp['snapshotSet'][0]['status'])
fake_snapshot.status = 'deleting'
resp = self.execute('DescribeSnapshots', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('pending', resp['snapshotSet'][0]['status'])
fake_snapshot.status = 'error'
resp = self.execute('DescribeSnapshots', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('error', resp['snapshotSet'][0]['status'])
fake_snapshot.status = 'banana'
resp = self.execute('DescribeSnapshots', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('banana', resp['snapshotSet'][0]['status'])

84
ec2api/tests/unit/test_subnet.py
View File

@ -34,7 +34,6 @@ class SubnetTestCase(base.ApiTestCase):
tools.get_neutron_create('subnet', fakes.ID_OS_SUBNET_1))
def check_response(resp):
self.assertEqual(200, resp['http_status_code'])
self.assertThat(fakes.EC2_SUBNET_1, matchers.DictMatches(
resp['subnet']))
self.db_api.add_item.assert_called_once_with(
@ -68,39 +67,37 @@ class SubnetTestCase(base.ApiTestCase):
check_response(resp)
def test_create_subnet_invalid_parameters(self):
def check_response(resp, error_code):
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
def do_check(args, error_code):
self.neutron.reset_mock()
self.db_api.reset_mock()
self.assert_execution_error(error_code, 'CreateSubnet', args)
self.assertEqual(0, self.neutron.create_network.call_count)
self.assertEqual(0, self.neutron.create_subnet.call_count)
self.assertEqual(0, self.neutron.add_interface_router.call_count)
self.neutron.reset_mock()
self.db_api.reset_mock()
self.set_mock_db_items()
resp = self.execute('CreateSubnet', {'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': fakes.CIDR_SUBNET_1})
do_check({'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': fakes.CIDR_SUBNET_1},
'InvalidVpcID.NotFound')
self.db_api.get_item_by_id.assert_called_once_with(mock.ANY,
fakes.ID_EC2_VPC_1)
check_response(resp, 'InvalidVpcID.NotFound')
self.set_mock_db_items(fakes.DB_VPC_1)
resp = self.execute('CreateSubnet', {'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': 'invalid_cidr'})
do_check({'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': 'invalid_cidr'},
'InvalidParameterValue')
self.assertEqual(0, self.db_api.get_item_by_id.call_count)
check_response(resp, 'InvalidParameterValue')
resp = self.execute('CreateSubnet', {'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': '10.10.0.0/30'})
do_check({'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': '10.10.0.0/30'},
'InvalidSubnet.Range')
self.assertEqual(0, self.db_api.get_item_by_id.call_count)
check_response(resp, 'InvalidSubnet.Range')
resp = self.execute('CreateSubnet', {'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': '10.20.0.0/24'})
do_check({'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': '10.20.0.0/24'},
'InvalidSubnet.Range')
self.db_api.get_item_by_id.assert_called_once_with(mock.ANY,
fakes.ID_EC2_VPC_1)
check_response(resp, 'InvalidSubnet.Range')
def test_create_subnet_overlapped(self):
self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_ROUTE_TABLE_1)
@ -112,10 +109,9 @@ class SubnetTestCase(base.ApiTestCase):
self.neutron.add_interface_router.side_effect = (
neutron_exception.BadRequest())
resp = self.execute('CreateSubnet', {'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': fakes.CIDR_SUBNET_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidSubnet.Conflict', resp['Error']['Code'])
self.assert_execution_error('InvalidSubnet.Conflict', 'CreateSubnet',
{'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': fakes.CIDR_SUBNET_1})
def test_create_subnet_overlimit(self):
self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_ROUTE_TABLE_1)
@ -130,12 +126,9 @@ class SubnetTestCase(base.ApiTestCase):
saved_side_effect = func.side_effect
func.side_effect = neutron_exception.OverQuotaClient
resp = self.execute('CreateSubnet',
{'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': fakes.CIDR_SUBNET_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('SubnetLimitExceeded', resp['Error']['Code'])
self.assert_execution_error('SubnetLimitExceeded', 'CreateSubnet',
{'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': fakes.CIDR_SUBNET_1})
func.side_effect = saved_side_effect
test_overlimit(self.neutron.create_network)
@ -152,8 +145,9 @@ class SubnetTestCase(base.ApiTestCase):
tools.get_neutron_create('subnet', fakes.ID_OS_SUBNET_1))
self.neutron.update_network.side_effect = Exception()
self.execute('CreateSubnet', {'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': fakes.CIDR_SUBNET_1})
self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'CreateSubnet',
{'VpcId': fakes.ID_EC2_VPC_1,
'CidrBlock': fakes.CIDR_SUBNET_1})
self.neutron.assert_has_calls([
mock.call.remove_interface_router(
@ -169,9 +163,8 @@ class SubnetTestCase(base.ApiTestCase):
{'subnet': fakes.OS_SUBNET_1})
resp = self.execute('DeleteSubnet',
{'subnetId': fakes.ID_EC2_SUBNET_1})
{'SubnetId': fakes.ID_EC2_SUBNET_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.db_api.delete_item.assert_called_once_with(
mock.ANY,
@ -199,15 +192,13 @@ class SubnetTestCase(base.ApiTestCase):
neutron_exception.NetworkInUseClient())
resp = self.execute('DeleteSubnet',
{'subnetId': fakes.ID_EC2_SUBNET_1})
self.assertEqual(200, resp['http_status_code'])
{'SubnetId': fakes.ID_EC2_SUBNET_1})
self.assertEqual(True, resp['return'])
self.neutron.show_subnet.side_effect = neutron_exception.NotFound()
resp = self.execute('DeleteSubnet',
{'subnetId': fakes.ID_EC2_SUBNET_1})
self.assertEqual(200, resp['http_status_code'])
{'SubnetId': fakes.ID_EC2_SUBNET_1})
self.assertEqual(True, resp['return'])
def test_delete_subnet_invalid_parameters(self):
@ -215,11 +206,8 @@ class SubnetTestCase(base.ApiTestCase):
self.neutron.show_subnet.return_value = fakes.OS_SUBNET_1
self.neutron.show_network.return_value = fakes.OS_NETWORK_1
resp = self.execute('DeleteSubnet',
{'subnetId': fakes.ID_EC2_SUBNET_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidSubnetID.NotFound', resp['Error']['Code'])
self.assert_execution_error('InvalidSubnetID.NotFound', 'DeleteSubnet',
{'SubnetId': fakes.ID_EC2_SUBNET_1})
self.assertEqual(0, self.neutron.delete_network.call_count)
self.assertEqual(0, self.neutron.delete_subnet.call_count)
self.assertEqual(0, self.neutron.remove_interface_router.call_count)
@ -229,16 +217,15 @@ class SubnetTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_SUBNET_1)
describe_network_interfaces.return_value = (
{'networkInterfaceSet': [fakes.EC2_NETWORK_INTERFACE_1]})
resp = self.execute('DeleteSubnet',
{'subnetId': fakes.ID_EC2_SUBNET_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('DependencyViolation', resp['Error']['Code'])
self.assert_execution_error('DependencyViolation', 'DeleteSubnet',
{'SubnetId': fakes.ID_EC2_SUBNET_1})
def test_delete_subnet_rollback(self):
self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_SUBNET_1)
self.neutron.show_subnet.side_effect = Exception()
self.execute('DeleteSubnet', {'subnetId': fakes.ID_EC2_SUBNET_1})
self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'DeleteSubnet',
{'SubnetId': fakes.ID_EC2_SUBNET_1})
self.db_api.restore_item.assert_called_once_with(
mock.ANY, 'subnet', fakes.DB_SUBNET_1)
@ -253,7 +240,6 @@ class SubnetTestCase(base.ApiTestCase):
{'networks': [fakes.OS_NETWORK_1, fakes.OS_NETWORK_2]})
resp = self.execute('DescribeSubnets', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['subnetSet'],
matchers.ListMatches([fakes.EC2_SUBNET_1,
fakes.EC2_SUBNET_2]))
@ -262,7 +248,6 @@ class SubnetTestCase(base.ApiTestCase):
return_value=[fakes.DB_SUBNET_2])
resp = self.execute('DescribeSubnets',
{'SubnetId.1': fakes.ID_EC2_SUBNET_2})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['subnetSet'],
matchers.ListMatches([fakes.EC2_SUBNET_2]))
self.db_api.get_items_by_ids.assert_called_once_with(
@ -292,5 +277,4 @@ class SubnetTestCase(base.ApiTestCase):
{'networks': [fakes.OS_NETWORK_1]})
resp = self.execute('DescribeSubnets', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual([], resp['subnetSet'])

81
ec2api/tests/unit/test_tag.py
View File

@ -34,9 +34,7 @@ class TagTestCase(base.ApiTestCase):
'Tag.1.Value': '',
'Tag.2.Key': 'admin',
'Tag.2.Value': 'John Smith'})
self.assertEqual({'http_status_code': 200,
'return': True},
resp)
self.assertEqual({'return': True}, resp)
self.assertEqual(1, self.db_api.add_tags.call_count)
self.assertEqual(2, len(self.db_api.add_tags.call_args))
self.assertThat(self.db_api.add_tags.call_args[0][1],
@ -67,9 +65,7 @@ class TagTestCase(base.ApiTestCase):
params.update({'Tag.1.Key': 'tag',
'Tag.1.Value': 'value'})
resp = self.execute('CreateTags', params)
self.assertEqual({'http_status_code': 200,
'return': True},
resp)
self.assertEqual({'return': True}, resp)
# NOTE(ft): check create a tag for non-existing images
self.db_api.get_item_by_id.return_value = None
@ -79,61 +75,49 @@ class TagTestCase(base.ApiTestCase):
'ResourceId.3': fakes.ID_EC2_IMAGE_ARI_1,
'Tag.1.Key': 'Oracle RAC node',
'Tag.1.Value': ''})
self.assertEqual({'http_status_code': 200,
'return': True},
resp)
self.assertEqual({'return': True}, resp)
def test_create_tags_invalid_parameters(self):
# NOTE(ft): check tag validity checks
resp = self.execute('CreateTags',
{'ResourceId.1': fakes.ID_EC2_VPC_1,
'Tag.1.Value': ''})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterValue', 'CreateTags',
{'ResourceId.1': fakes.ID_EC2_VPC_1,
'Tag.1.Value': ''})
resp = self.execute('CreateTags',
{'ResourceId.1': fakes.ID_EC2_VPC_1,
'Tag.1.Key': ''})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterValue', 'CreateTags',
{'ResourceId.1': fakes.ID_EC2_VPC_1,
'Tag.1.Key': ''})
resp = self.execute('CreateTags',
{'ResourceId.1': fakes.ID_EC2_VPC_1,
'Tag.1.Key': 'a' * 128})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterValue', 'CreateTags',
{'ResourceId.1': fakes.ID_EC2_VPC_1,
'Tag.1.Key': 'a' * 128})
resp = self.execute('CreateTags',
{'ResourceId.1': fakes.ID_EC2_VPC_1,
'Tag.1.Key': 'fake-key',
'Tag.1.Value': 'a' * 256})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidParameterValue', resp['Error']['Code'])
self.assert_execution_error('InvalidParameterValue', 'CreateTags',
{'ResourceId.1': fakes.ID_EC2_VPC_1,
'Tag.1.Key': 'fake-key',
'Tag.1.Value': 'a' * 256})
# NOTE(ft): check resource type check
resp = self.execute('CreateTags',
{'ResourceId.1': fakes.random_ec2_id('fake'),
'Tag.1.Key': 'fake-key',
'Tag.1.Value': 'fake-value'})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidID', resp['Error']['Code'])
self.assert_execution_error(
'InvalidID', 'CreateTags',
{'ResourceId.1': fakes.random_ec2_id('fake'),
'Tag.1.Key': 'fake-key',
'Tag.1.Value': 'fake-value'})
# NOTE(ft): check resource existence check
self.db_api.get_item_by_id.return_value = None
for r_id in tag_api.RESOURCE_TYPES:
if r_id in ('ami', 'ari', 'aki'):
continue
resp = self.execute('CreateTags',
{'ResourceId.1': fakes.random_ec2_id(r_id),
'Tag.1.Key': 'fake-key',
'Tag.1.Value': 'fake-value'})
self.assertEqual(400, resp['http_status_code'])
exc_class = ec2utils.NOT_FOUND_EXCEPTION_MAP[r_id]
try:
error_code = exc_class.ec2_code
except AttributeError:
error_code = exc_class.__name__
self.assertEqual(error_code, resp['Error']['Code'])
self.assert_execution_error(
error_code, 'CreateTags',
{'ResourceId.1': fakes.random_ec2_id(r_id),
'Tag.1.Key': 'fake-key',
'Tag.1.Value': 'fake-value'})
def test_delete_tag(self):
resp = self.execute('DeleteTags',
@ -143,9 +127,7 @@ class TagTestCase(base.ApiTestCase):
'Tag.2.Value': 'value2',
'Tag.3.Key': 'key3',
'Tag.3.Value': 'value3'})
self.assertEqual({'http_status_code': 200,
'return': True},
resp)
self.assertEqual({'return': True}, resp)
self.db_api.delete_tags.assert_called_once_with(
mock.ANY, [fakes.ID_EC2_VPC_1, fakes.ID_EC2_SUBNET_1],
[{'key': 'key1'},
@ -155,9 +137,7 @@ class TagTestCase(base.ApiTestCase):
resp = self.execute('DeleteTags',
{'ResourceId.1': fakes.ID_EC2_VPC_1})
self.assertEqual({'http_status_code': 200,
'return': True},
resp)
self.assertEqual({'return': True}, resp)
self.db_api.delete_tags.assert_called_with(
mock.ANY, [fakes.ID_EC2_VPC_1], None)
@ -173,8 +153,6 @@ class TagTestCase(base.ApiTestCase):
'value': 'value3'}
]
resp = self.execute('DescribeTags', {})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp,
matchers.DictMatches(
{'tagSet': [{'resourceType': 'vpc',
@ -219,8 +197,7 @@ class TagTestCase(base.ApiTestCase):
'key': 'fake-key',
'value': 'fake-value'}]
resp = self.execute('DescribeTags', {})
self.assertEqual({'http_status_code': 200,
'tagSet': [{'resourceType': r_type,
self.assertEqual({'tagSet': [{'resourceType': r_type,
'resourceId': item_id,
'key': 'fake-key',
'value': 'fake-value'}]},

46
ec2api/tests/unit/test_volume.py
View File

@ -35,8 +35,6 @@ class VolumeTestCase(base.ApiTestCase):
tools.get_db_api_add_item(fakes.ID_EC2_VOLUME_3))
resp = self.execute('DescribeVolumes', {})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'volumeSet': [fakes.EC2_VOLUME_1, fakes.EC2_VOLUME_2,
fakes.EC2_VOLUME_3]},
@ -48,8 +46,6 @@ class VolumeTestCase(base.ApiTestCase):
return_value=[fakes.DB_VOLUME_1])
resp = self.execute('DescribeVolumes',
{'VolumeId.1': fakes.ID_EC2_VOLUME_1})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'volumeSet': [fakes.EC2_VOLUME_1]},
orderless_lists=True))
@ -74,8 +70,6 @@ class VolumeTestCase(base.ApiTestCase):
self.cinder.volumes.list.return_value = []
self.set_mock_db_items(fakes.DB_VOLUME_1, fakes.DB_VOLUME_2)
resp = self.execute('DescribeVolumes', {})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertThat(resp, matchers.DictMatches(
{'volumeSet': []}))
@ -89,17 +83,15 @@ class VolumeTestCase(base.ApiTestCase):
fakes.CinderVolume(fakes.OS_VOLUME_1),
fakes.CinderVolume(fakes.OS_VOLUME_2)]
resp = self.execute('DescribeVolumes',
{'VolumeId.1': fakes.random_ec2_id('vol')})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidVolume.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidVolume.NotFound', 'DescribeVolumes',
{'VolumeId.1': fakes.random_ec2_id('vol')})
self.cinder.volumes.list.side_effect = lambda: []
resp = self.execute('DescribeVolumes',
{'VolumeId.1': fakes.ID_EC2_VOLUME_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidVolume.NotFound', resp['Error']['Code'])
self.assert_execution_error(
'InvalidVolume.NotFound', 'DescribeVolumes',
{'VolumeId.1': fakes.ID_EC2_VOLUME_1})
def test_create_volume(self):
self.cinder.volumes.create.return_value = (
@ -110,9 +102,7 @@ class VolumeTestCase(base.ApiTestCase):
resp = self.execute(
'CreateVolume',
{'AvailabilityZone': fakes.NAME_AVAILABILITY_ZONE})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(fakes.EC2_VOLUME_1, matchers.DictMatches(
tools.purge_dict(resp, {'http_status_code'})))
self.assertThat(fakes.EC2_VOLUME_1, matchers.DictMatches(resp))
self.db_api.add_item.assert_called_once_with(
mock.ANY, 'vol',
tools.purge_dict(fakes.DB_VOLUME_1, ('id',)))
@ -132,9 +122,7 @@ class VolumeTestCase(base.ApiTestCase):
'CreateVolume',
{'AvailabilityZone': fakes.NAME_AVAILABILITY_ZONE,
'SnapshotId': fakes.ID_EC2_SNAPSHOT_1})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(fakes.EC2_VOLUME_3, matchers.DictMatches(
tools.purge_dict(resp, {'http_status_code'})))
self.assertThat(fakes.EC2_VOLUME_3, matchers.DictMatches(resp))
self.db_api.add_item.assert_called_once_with(
mock.ANY, 'vol',
tools.purge_dict(fakes.DB_VOLUME_3, ('id',)))
@ -147,8 +135,6 @@ class VolumeTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_VOLUME_1)
resp = self.execute('DeleteVolume',
{'VolumeId': fakes.ID_EC2_VOLUME_1})
self.assertEqual(200, resp['http_status_code'])
resp.pop('http_status_code')
self.assertEqual({'return': True}, resp)
self.cinder.volumes.delete.assert_called_once_with(
fakes.ID_OS_VOLUME_1)
@ -161,22 +147,18 @@ class VolumeTestCase(base.ApiTestCase):
fake_volume.status = 'creating'
resp = self.execute('DescribeVolumes', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('creating', resp['volumeSet'][0]['status'])
fake_volume.status = 'attaching'
resp = self.execute('DescribeVolumes', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('in-use', resp['volumeSet'][0]['status'])
fake_volume.status = 'detaching'
resp = self.execute('DescribeVolumes', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('in-use', resp['volumeSet'][0]['status'])
fake_volume.status = 'banana'
resp = self.execute('DescribeVolumes', {})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual('banana', resp['volumeSet'][0]['status'])
def test_attach_volume(self):
@ -191,8 +173,7 @@ class VolumeTestCase(base.ApiTestCase):
{'VolumeId': fakes.ID_EC2_VOLUME_3,
'InstanceId': fakes.ID_EC2_INSTANCE_2,
'Device': '/dev/vdf'})
self.assertEqual({'http_status_code': 200,
'device': '/dev/vdf',
self.assertEqual({'device': '/dev/vdf',
'instanceId': fakes.ID_EC2_INSTANCE_2,
'status': 'attaching',
'volumeId': fakes.ID_EC2_VOLUME_3},
@ -211,8 +192,7 @@ class VolumeTestCase(base.ApiTestCase):
resp = self.execute('DetachVolume',
{'VolumeId': fakes.ID_EC2_VOLUME_2})
self.assertEqual({'http_status_code': 200,
'device': os_volume.attachments[0]['device'],
self.assertEqual({'device': os_volume.attachments[0]['device'],
'instanceId': fakes.ID_EC2_INSTANCE_2,
'status': 'detaching',
'volumeId': fakes.ID_EC2_VOLUME_2},
@ -226,7 +206,5 @@ class VolumeTestCase(base.ApiTestCase):
self.cinder.volumes.get.return_value = (
fakes.CinderVolume(fakes.OS_VOLUME_1))
resp = self.execute('DetachVolume',
{'VolumeId': fakes.ID_EC2_VOLUME_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('IncorrectState', resp['Error']['Code'])
self.assert_execution_error('IncorrectState', 'DetachVolume',
{'VolumeId': fakes.ID_EC2_VOLUME_1})

41
ec2api/tests/unit/test_vpc.py
View File

@ -37,7 +37,6 @@ class VpcTestCase(base.ApiTestCase):
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_1))
def check_response(response):
self.assertEqual(200, response['http_status_code'])
self.assertIn('vpc', response)
vpc = resp['vpc']
self.assertThat(fakes.EC2_VPC_1, matchers.DictMatches(vpc))
@ -74,19 +73,15 @@ class VpcTestCase(base.ApiTestCase):
self.db_api.add_item.side_effect = tools.get_db_api_add_item(
fakes.ID_EC2_VPC_1)
def check_response(resp, error_code):
self.assertEqual(400, resp['http_status_code'])
self.assertEqual(error_code, resp['Error']['Code'])
def do_check(args, error_code):
self.assert_execution_error(error_code, 'CreateVpc', args)
self.assertEqual(0, self.neutron.create_router.call_count)
self.neutron.reset_mock()
self.db_api.reset_mock()
resp = self.execute('CreateVpc', {'CidrBlock': 'bad_cidr'})
check_response(resp, 'InvalidParameterValue')
resp = self.execute('CreateVpc', {'CidrBlock': '10.0.0.0/8'})
check_response(resp, 'InvalidVpc.Range')
do_check({'CidrBlock': 'bad_cidr'}, 'InvalidParameterValue')
do_check({'CidrBlock': '10.0.0.0/8'}, 'InvalidVpc.Range')
def test_create_vpc_overlimit(self):
self.neutron.create_router.side_effect = (
@ -94,10 +89,8 @@ class VpcTestCase(base.ApiTestCase):
self.db_api.add_item.side_effect = tools.get_db_api_add_item(
fakes.ID_EC2_VPC_1)
resp = self.execute('CreateVpc', {'CidrBlock': fakes.CIDR_VPC_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('VpcLimitExceeded', resp['Error']['Code'])
self.assert_execution_error('VpcLimitExceeded', 'CreateVpc',
{'CidrBlock': fakes.CIDR_VPC_1})
self.neutron.create_router.assert_called_with({'router': {}})
self.assertEqual(0, self.db_api.add_item.call_count)
@ -110,7 +103,8 @@ class VpcTestCase(base.ApiTestCase):
'rtb': fakes.ID_EC2_ROUTE_TABLE_1}))
self.neutron.update_router.side_effect = Exception()
self.execute('CreateVpc', {'CidrBlock': fakes.CIDR_VPC_1})
self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'CreateVpc',
{'CidrBlock': fakes.CIDR_VPC_1})
self.neutron.delete_router.assert_called_once_with(
fakes.ID_OS_ROUTER_1)
@ -123,7 +117,6 @@ class VpcTestCase(base.ApiTestCase):
resp = self.execute('DeleteVpc', {'VpcId': fakes.ID_EC2_VPC_1})
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.neutron.delete_router.assert_called_once_with(
fakes.ID_OS_ROUTER_1)
@ -137,18 +130,15 @@ class VpcTestCase(base.ApiTestCase):
def test_delete_vpc_not_found(self):
self.set_mock_db_items()
resp = self.execute('DeleteVpc', {'VpcId': fakes.ID_EC2_VPC_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('InvalidVpcID.NotFound', resp['Error']['Code'])
self.assert_execution_error('InvalidVpcID.NotFound', 'DeleteVpc',
{'VpcId': fakes.ID_EC2_VPC_1})
self.assertEqual(0, self.neutron.delete_router.call_count)
self.assertEqual(0, self.db_api.delete_item.call_count)
def test_delete_vpc_dependency_violation(self):
def do_check():
resp = self.execute('DeleteVpc',
{'VpcId': fakes.ID_EC2_VPC_1})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('DependencyViolation', resp['Error']['Code'])
self.assert_execution_error('DependencyViolation', 'DeleteVpc',
{'VpcId': fakes.ID_EC2_VPC_1})
self.assertEqual(0, self.neutron.delete_router.call_count)
self.assertEqual(0, self.db_api.delete_item.call_count)
@ -166,7 +156,6 @@ class VpcTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_ROUTE_TABLE_1)
def check_response(resp):
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.neutron.delete_router.assert_called_once_with(
fakes.ID_OS_ROUTER_1)
@ -192,7 +181,8 @@ class VpcTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_ROUTE_TABLE_1)
self.neutron.delete_router.side_effect = Exception()
self.execute('DeleteVpc', {'VpcId': fakes.ID_EC2_VPC_1})
self.assert_execution_error(self.ANY_EXECUTE_ERROR, 'DeleteVpc',
{'VpcId': fakes.ID_EC2_VPC_1})
self.db_api.restore_item.assert_any_call(
mock.ANY, 'vpc', fakes.DB_VPC_1)
@ -205,7 +195,6 @@ class VpcTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_VPC_1, fakes.DB_VPC_2)
resp = self.execute('DescribeVpcs', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['vpcSet'],
matchers.ListMatches([fakes.EC2_VPC_1,
fakes.EC2_VPC_2]))
@ -213,7 +202,6 @@ class VpcTestCase(base.ApiTestCase):
resp = self.execute('DescribeVpcs',
{'VpcId.1': fakes.ID_EC2_VPC_1})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['vpcSet'],
matchers.ListMatches([fakes.EC2_VPC_1]))
self.db_api.get_items_by_ids.assert_called_once_with(
@ -237,7 +225,6 @@ class VpcTestCase(base.ApiTestCase):
resp = self.execute('DescribeVpcs', {})
self.assertEqual(200, resp['http_status_code'])
self.assertThat(resp['vpcSet'],
matchers.ListMatches([fakes.EC2_VPC_1]))
self.db_api.get_items.assert_called_once_with(mock.ANY, 'vpc')