openstack
/
ec2-api
Code Issues Proposed changes

Don't use mock shortcuts on novaclient object managers 

Change-Id: I21e6c2a086f3df50762cd7ac8e6e851c904c7a2d
This commit is contained in:
Feodor Tersin 2015-03-04 19:55:21 +03:00
parent 5032355da3
commit a7fd598800
11 changed files with 110 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -44,19 +44,7 @@ class ApiTestCase(test_base.BaseTestCase):
self.addCleanup(neutron_patcher.stop)
nova_patcher = mock.patch('novaclient.v1_1.client.Client')
nova_mock = nova_patcher.start()
self.nova_availability_zones = (
nova_mock.return_value.availability_zones)
self.nova_servers = nova_mock.return_value.servers
self.nova_flavors = nova_mock.return_value.flavors
self.nova_floating_ips = nova_mock.return_value.floating_ips
self.nova_fixed_ips = nova_mock.return_value.fixed_ips
self.nova_key_pairs = nova_mock.return_value.keypairs
self.nova_security_groups = nova_mock.return_value.security_groups
self.nova_security_group_rules = (
nova_mock.return_value.security_group_rules)
self.nova_volumes = nova_mock.return_value.volumes
self.nova_quotas = nova_mock.return_value.quotas
self.nova = nova_patcher.start().return_value
self.addCleanup(nova_patcher.stop)
glance_patcher = mock.patch('glanceclient.client.Client')

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

@ -28,7 +28,7 @@ class AddressTestCase(base.ApiTestCase):
def test_allocate_ec2_classic_address(self):
address.address_engine = (
address.AddressEngineNeutron())
self.nova_floating_ips.create.return_value = (
self.nova.floating_ips.create.return_value = (
fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_1))
resp = self.execute('AllocateAddress', {})
@ -37,7 +37,7 @@ class AddressTestCase(base.ApiTestCase):
self.assertEqual('standard', resp['domain'])
self.assertNotIn('allocationId', resp)
self.assertEqual(0, self.db_api.add_item.call_count)
self.nova_floating_ips.create.assert_called_once_with()
self.nova.floating_ips.create.assert_called_once_with()
def test_allocate_vpc_address(self):
address.address_engine = (
@ -88,7 +88,7 @@ class AddressTestCase(base.ApiTestCase):
address.address_engine = (
address.AddressEngineNeutron())
self.nova_floating_ips.create.side_effect = (
self.nova.floating_ips.create.side_effect = (
nova_exception.Forbidden(403))
resp = self.execute('AllocateAddress', {})
self.assertEqual(400, resp['http_status_code'])
@ -113,10 +113,10 @@ class AddressTestCase(base.ApiTestCase):
address.address_engine = (
address.AddressEngineNeutron())
self.set_mock_db_items(fakes.DB_INSTANCE_1)
self.nova_floating_ips.list.return_value = (
self.nova.floating_ips.list.return_value = (
[fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_1),
fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_2)])
self.nova_servers.add_floating_ip.return_value = True
self.nova.servers.add_floating_ip.return_value = True
resp = self.execute('AssociateAddress',
{'PublicIp': fakes.IP_ADDRESS_1,
@ -124,7 +124,7 @@ class AddressTestCase(base.ApiTestCase):
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.nova_servers.add_floating_ip.assert_called_once_with(
self.nova.servers.add_floating_ip.assert_called_once_with(
fakes.ID_OS_INSTANCE_1,
fakes.IP_ADDRESS_1)
@ -247,7 +247,7 @@ class AddressTestCase(base.ApiTestCase):
self.assertEqual('InvalidParameterCombination', resp['Error']['Code'])
# NOTE(ft): ec2 classic instance vs not existing public IP
self.nova_floating_ips.list.return_value = []
self.nova.floating_ips.list.return_value = []
resp = self.execute('AssociateAddress',
{'PublicIp': fakes.IP_ADDRESS_1,
'InstanceId': fakes.ID_EC2_INSTANCE_2})
@ -362,15 +362,15 @@ class AddressTestCase(base.ApiTestCase):
address.address_engine = (
address.AddressEngineNeutron())
self.set_mock_db_items()
self.nova_servers.remove_floating_ip.return_value = True
self.nova_floating_ips.list.return_value = (
self.nova.servers.remove_floating_ip.return_value = True
self.nova.floating_ips.list.return_value = (
[fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_1),
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(
self.nova.servers.remove_floating_ip.assert_called_once_with(
fakes.ID_OS_INSTANCE_1,
fakes.IP_ADDRESS_2)
@ -379,7 +379,7 @@ class AddressTestCase(base.ApiTestCase):
{'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)
self.assertEqual(1, self.nova.servers.remove_floating_ip.call_count)
def test_dissassociate_address_vpc(self):
address.address_engine = (
@ -434,7 +434,7 @@ 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 = []
self.nova.floating_ips.list.return_value = []
resp = self.execute('DisassociateAddress',
{'PublicIp': fakes.IP_ADDRESS_2})
self.assertEqual(400, resp['http_status_code'])
@ -476,8 +476,8 @@ class AddressTestCase(base.ApiTestCase):
address.address_engine = (
address.AddressEngineNeutron())
self.set_mock_db_items()
self.nova_floating_ips.delete.return_value = True
self.nova_floating_ips.list.return_value = (
self.nova.floating_ips.delete.return_value = True
self.nova.floating_ips.list.return_value = (
[fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_1),
fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_2)])
@ -486,7 +486,7 @@ class AddressTestCase(base.ApiTestCase):
self.assertEqual(200, resp['http_status_code'])
self.assertEqual(True, resp['return'])
self.nova_floating_ips.delete.assert_called_once_with(
self.nova.floating_ips.delete.assert_called_once_with(
fakes.NOVA_FLOATING_IP_1['id'])
def test_release_address_vpc(self):
@ -523,7 +523,7 @@ class AddressTestCase(base.ApiTestCase):
'InvalidParameterCombination')
# NOTE(ft): EC2 Classic public IP is not found
self.nova_floating_ips.list.return_value = []
self.nova.floating_ips.list.return_value = []
do_check({'PublicIp': fakes.IP_ADDRESS_1},
'AuthFailure')
@ -611,7 +611,7 @@ class AddressTestCase(base.ApiTestCase):
address.address_engine = (
address.AddressEngineNova())
self.set_mock_db_items(fakes.DB_INSTANCE_1)
self.nova_floating_ips.list.return_value = [
self.nova.floating_ips.list.return_value = [
fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_1),
fakes.NovaFloatingIp(fakes.NOVA_FLOATING_IP_2)]
resp = self.execute('DescribeAddresses', {})

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

@ -23,14 +23,14 @@ from ec2api.tests.unit import matchers
class AvailabilityZoneCase(base.ApiTestCase):
def test_describe_availability_zones(self):
self.nova_availability_zones.list.return_value = [
self.nova.availability_zones.list.return_value = [
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()
self.nova.availability_zones.list.assert_called_once()
self.check_filtering(
'DescribeAvailabilityZones', 'availabilityZoneInfo',
@ -38,14 +38,14 @@ class AvailabilityZoneCase(base.ApiTestCase):
('zone-name', fakes.NAME_AVAILABILITY_ZONE)])
def test_describe_availability_zones_verbose(self):
self.nova_availability_zones.list.return_value = [
self.nova.availability_zones.list.return_value = [
fakes.NovaAvailabilityZone(fakes.OS_AVAILABILITY_ZONE),
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()
self.nova.availability_zones.list.assert_called_once()
def test_regions(self):
resp = self.execute('DescribeRegions', {})
@ -55,7 +55,7 @@ class AvailabilityZoneCase(base.ApiTestCase):
is not None)
def test_describe_account_attributes(self):
self.nova_quotas.get.return_value = mock.Mock(instances=77)
self.nova.quotas.get.return_value = mock.Mock(instances=77)
availability_zone.account_attribute_engine = (
availability_zone.AccountAttributeEngineNeutron())
@ -74,7 +74,7 @@ class AvailabilityZoneCase(base.ApiTestCase):
'attributeValueSet': [
{'attributeValue': 77}]}],
orderless_lists=True))
self.nova_quotas.get.assert_called_once_with(
self.nova.quotas.get.assert_called_once_with(
fakes.ID_OS_PROJECT, fakes.ID_OS_USER)
availability_zone.account_attribute_engine = (

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

@ -106,7 +106,7 @@ class ImageTestCase(base.ApiTestCase):
os_image = mock.MagicMock()
os_image.configure_mock(id=fakes.random_os_id())
os_instance.create_image.return_value = os_image
self.nova_servers.get.return_value = os_instance
self.nova.servers.get.return_value = os_instance
is_ebs_instance.return_value = True
image_id = fakes.random_ec2_id('ami')
self.db_api.add_item.side_effect = tools.get_db_api_add_item(image_id)
@ -120,7 +120,7 @@ class ImageTestCase(base.ApiTestCase):
resp)
self.db_api.get_item_by_id.assert_called_once_with(
mock.ANY, fakes.ID_EC2_INSTANCE_2)
self.nova_servers.get.assert_called_once_with(fakes.ID_OS_INSTANCE_2)
self.nova.servers.get.assert_called_once_with(fakes.ID_OS_INSTANCE_2)
is_ebs_instance.assert_called_once_with(mock.ANY, os_instance)
self.db_api.add_item.assert_called_once_with(
mock.ANY, 'ami', {'os_id': os_image.id,
@ -131,7 +131,7 @@ class ImageTestCase(base.ApiTestCase):
os_instance.start.assert_called_once_with()
os_instance.create_image.assert_called_once_with('fake_name')
self.db_api.reset_mock()
self.nova_servers.reset_mock()
self.nova.servers.reset_mock()
def test_create_image(self):
self._test_create_image('ACTIVE', False)

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

@ -57,8 +57,8 @@ class InstanceTestCase(base.ApiTestCase):
self.fake_flavor = mock.Mock()
self.fake_flavor.configure_mock(name='fake_flavor')
self.nova_flavors.get.return_value = self.fake_flavor
self.nova_flavors.list.return_value = [self.fake_flavor]
self.nova.flavors.get.return_value = self.fake_flavor
self.nova.flavors.list.return_value = [self.fake_flavor]
@mock.patch('ec2api.api.instance.InstanceEngineNeutron.'
'get_ec2_network_interfaces')
@ -77,7 +77,7 @@ class InstanceTestCase(base.ApiTestCase):
{'networkInterface': fakes.EC2_NETWORK_INTERFACE_1})
self.db_api.add_item.return_value = fakes.DB_INSTANCE_1
self.nova_servers.create.return_value = (
self.nova.servers.create.return_value = (
fakes.OSInstance(
fakes.ID_OS_INSTANCE_1, {'id': 'fakeFlavorId'},
image={'id': fakes.ID_OS_IMAGE_1}))
@ -127,7 +127,7 @@ class InstanceTestCase(base.ApiTestCase):
create_network_interface.assert_called_once_with(
mock.ANY, fakes.ID_EC2_SUBNET_1,
**create_network_interface_kwargs))
self.nova_servers.create.assert_called_once_with(
self.nova.servers.create.assert_called_once_with(
'%s-%s' % (fakes.ID_EC2_RESERVATION_1, 0),
fakes.ID_OS_IMAGE_1, self.fake_flavor,
min_count=1, max_count=1,
@ -155,7 +155,7 @@ class InstanceTestCase(base.ApiTestCase):
mock.ANY, 'ari', (fakes.ID_OS_IMAGE_ARI_1,))
self.network_interface_api.reset_mock()
self.nova_servers.reset_mock()
self.nova.servers.reset_mock()
self.db_api.reset_mock()
self.novadb.reset_mock()
get_ec2_network_interfaces.reset_mock()
@ -237,7 +237,7 @@ class InstanceTestCase(base.ApiTestCase):
self.network_interface_api.create_network_interface.side_effect = (
[{'networkInterface': eni}
for eni in self.EC2_DETACHED_ENIS])
self.nova_servers.create.side_effect = [
self.nova.servers.create.side_effect = [
fakes.OSInstance(os_instance_id, {'id': 'fakeFlavorId'})
for os_instance_id in self.IDS_OS_INSTANCE]
self.novadb.instance_get_by_uuid.side_effect = self.NOVADB_INSTANCES
@ -264,7 +264,7 @@ class InstanceTestCase(base.ApiTestCase):
self.network_interface_api.create_network_interface.assert_has_calls([
mock.call(mock.ANY, ec2_subnet_id)
for ec2_subnet_id in self.IDS_EC2_SUBNET_BY_PORT])
self.nova_servers.create.assert_has_calls([
self.nova.servers.create.assert_has_calls([
mock.call(
'%s-%s' % (fakes.ID_EC2_RESERVATION_1, launch_index),
fakes.ID_OS_IMAGE_1, self.fake_flavor,
@ -329,14 +329,14 @@ class InstanceTestCase(base.ApiTestCase):
'BlockDeviceMapping.1.Ebs.DeleteOnTermination': 'False'})
self.assertEqual(200, resp['http_status_code'])
self.nova_servers.create.assert_called_once_with(
self.nova.servers.create.assert_called_once_with(
mock.ANY, mock.ANY, mock.ANY, min_count=1, max_count=1,
userdata=None, kernel_id=fakes.ID_OS_IMAGE_AKI_1,
ramdisk_id=fakes.ID_OS_IMAGE_ARI_1, key_name=None,
block_device_mapping='fake_bdm',
availability_zone='fake_zone', security_groups=['default'],
**extra_kwargs)
self.nova_servers.reset_mock()
self.nova.servers.reset_mock()
db_instance = {'os_id': mock.ANY,
'reservation_id': mock.ANY,
'launch_index': 0,
@ -473,7 +473,7 @@ class InstanceTestCase(base.ApiTestCase):
{'networkInterface': fakes.EC2_NETWORK_INTERFACE_1})
self.db_api.add_item.return_value = fakes.DB_INSTANCE_1
self.utils_generate_uid.return_value = fakes.ID_EC2_RESERVATION_1
self.nova_servers.create.return_value = (
self.nova.servers.create.return_value = (
fakes.OSInstance(fakes.ID_OS_INSTANCE_1, {'id': 'fakeFlavorId'},
image={'id': fakes.ID_OS_IMAGE_1}))
self.novadb.instance_get_by_uuid.side_effect = Exception()
@ -483,7 +483,7 @@ class InstanceTestCase(base.ApiTestCase):
mock_manager.attach_mock(self.network_interface_api,
'network_interface_api')
mock_manager.attach_mock(self.neutron, 'neutron')
mock_manager.attach_mock(self.nova_servers, 'nova_servers')
mock_manager.attach_mock(self.nova.servers, 'nova_servers')
params.update({'ImageId': fakes.ID_EC2_IMAGE_1,
'InstanceType': 'fake_flavor',
@ -513,7 +513,7 @@ class InstanceTestCase(base.ApiTestCase):
self.network_interface_api.reset_mock()
self.neutron.reset_mock()
self.nova_servers.reset_mock()
self.nova.servers.reset_mock()
self.db_api.reset_mock()
do_check({'SubnetId': fakes.ID_EC2_SUBNET_1})
@ -559,7 +559,7 @@ class InstanceTestCase(base.ApiTestCase):
{'networkInterface': {'networkInterfaceId': eni['id']}}
for eni in network_interfaces]
self.db_api.add_item.side_effect = instances
self.nova_servers.create.side_effect = os_instances
self.nova.servers.create.side_effect = os_instances
self.novadb.instance_get_by_uuid.side_effect = [
{}, {}, Exception()]
format_reservation.side_effect = (
@ -583,12 +583,12 @@ class InstanceTestCase(base.ApiTestCase):
{'instanceId': inst['id']}
for inst in instances[:2]]}))
self.nova_servers.delete.assert_called_once_with(
self.nova.servers.delete.assert_called_once_with(
instances[2]['os_id'])
self.db_api.delete_item.assert_called_once_with(
mock.ANY, instances[2]['id'])
self.nova_servers.reset_mock()
self.nova.servers.reset_mock()
self.db_api.reset_mock()
do_check(instance_api.InstanceEngineNeutron())
@ -635,7 +635,7 @@ class InstanceTestCase(base.ApiTestCase):
fakes.DB_INSTANCE_1, fakes.DB_INSTANCE_2,
fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2,
fakes.DB_ADDRESS_1, fakes.DB_ADDRESS_2)
self.nova_servers.get.side_effect = [fakes.OS_INSTANCE_1,
self.nova.servers.get.side_effect = [fakes.OS_INSTANCE_1,
fakes.OS_INSTANCE_2]
resp = self.execute('TerminateInstances',
@ -661,9 +661,9 @@ class InstanceTestCase(base.ApiTestCase):
(self.network_interface_api.
detach_network_interface.assert_called_once_with(
mock.ANY, fakes.ID_EC2_NETWORK_INTERFACE_2_ATTACH))
self.assertEqual(2, self.nova_servers.get.call_count)
self.nova_servers.get.assert_any_call(fakes.ID_OS_INSTANCE_1)
self.nova_servers.get.assert_any_call(fakes.ID_OS_INSTANCE_2)
self.assertEqual(2, self.nova.servers.get.call_count)
self.nova.servers.get.assert_any_call(fakes.ID_OS_INSTANCE_1)
self.nova.servers.get.assert_any_call(fakes.ID_OS_INSTANCE_2)
self.assertEqual(
0, self.address_api.dissassociate_address_item.call_count)
self.assertFalse(self.db_api.delete_item.called)
@ -690,7 +690,7 @@ class InstanceTestCase(base.ApiTestCase):
fake_state_change),
tools.update_dict({'instanceId': fakes.ID_EC2_INSTANCE_2},
fake_state_change)]}
self.nova_servers.get.side_effect = (
self.nova.servers.get.side_effect = (
lambda ec2_id: fakes.OSInstance(ec2_id, vm_state='active'))
def do_check(mock_eni_list=[], detached_enis=[], deleted_enis=[]):
@ -801,7 +801,7 @@ class InstanceTestCase(base.ApiTestCase):
@mock.patch('oslo_utils.timeutils.utcnow')
def _test_instance_get_operation(self, operation, getter, key, utcnow):
self.set_mock_db_items(fakes.DB_INSTANCE_2)
self.nova_servers.get.return_value = fakes.OS_INSTANCE_2
self.nova.servers.get.return_value = fakes.OS_INSTANCE_2
getter.return_value = 'fake_data'
utcnow.return_value = datetime.datetime(2015, 1, 19, 23, 34, 45, 123)
resp = self.execute(operation,
@ -813,7 +813,7 @@ class InstanceTestCase(base.ApiTestCase):
resp)
self.db_api.get_item_by_id.assert_called_once_with(
mock.ANY, fakes.ID_EC2_INSTANCE_2)
self.nova_servers.get.assert_called_once_with(fakes.ID_OS_INSTANCE_2)
self.nova.servers.get.assert_called_once_with(fakes.ID_OS_INSTANCE_2)
getter.assert_called_once_with(fakes.OS_INSTANCE_2)
@mock.patch.object(fakes.OSInstance, 'get_password', autospec=True)
@ -836,7 +836,7 @@ class InstanceTestCase(base.ApiTestCase):
fakes.DB_IMAGE_1, fakes.DB_IMAGE_2,
fakes.DB_IMAGE_ARI_1, fakes.DB_IMAGE_AKI_1,
fakes.DB_VOLUME_1, fakes.DB_VOLUME_2, fakes.DB_VOLUME_3)
self.nova_servers.list.return_value = [fakes.OS_INSTANCE_1,
self.nova.servers.list.return_value = [fakes.OS_INSTANCE_1,
fakes.OS_INSTANCE_2]
self.novadb.instance_get_by_uuid.side_effect = (
tools.get_by_2nd_arg_getter({
@ -924,7 +924,7 @@ class InstanceTestCase(base.ApiTestCase):
self.set_mock_db_items(
fakes.DB_INSTANCE_2, fakes.DB_IMAGE_1, fakes.DB_IMAGE_2,
fakes.DB_VOLUME_1, fakes.DB_VOLUME_2, fakes.DB_VOLUME_3)
self.nova_servers.list.return_value = [fakes.OS_INSTANCE_2]
self.nova.servers.list.return_value = [fakes.OS_INSTANCE_2]
self.novadb.instance_get_by_uuid.return_value = (
fakes.NOVADB_INSTANCE_2)
self.novadb.block_device_mapping_get_all_by_instance.return_value = (
@ -959,7 +959,7 @@ class InstanceTestCase(base.ApiTestCase):
describe_network_interfaces.return_value = copy.deepcopy(
{'networkInterfaceSet': list(
itertools.chain(*ec2_enis_by_instance))})
self.nova_servers.list.return_value = [
self.nova.servers.list.return_value = [
fakes.OSInstance(
os_id, {'id': 'fakeFlavorId'},
addresses=dict((subnet_name,
@ -1018,7 +1018,7 @@ class InstanceTestCase(base.ApiTestCase):
def test_describe_instances_auto_remove(self, remove_instances):
self.set_mock_db_items(fakes.DB_INSTANCE_1, fakes.DB_INSTANCE_2,
fakes.DB_VOLUME_2)
self.nova_servers.list.return_value = [fakes.OS_INSTANCE_2]
self.nova.servers.list.return_value = [fakes.OS_INSTANCE_2]
self.novadb.instance_get_by_uuid.return_value = (
fakes.NOVADB_INSTANCE_2)
self.novadb.block_device_mapping_get_all_by_instance.return_value = (
@ -1048,7 +1048,7 @@ class InstanceTestCase(base.ApiTestCase):
os_instances = [
fakes.OSInstance(inst['os_id'])
for inst in db_instances]
self.nova_servers.list.return_value = os_instances
self.nova.servers.list.return_value = os_instances
format_instance.side_effect = (
lambda context, instance, *args: (
{'instanceId': instance['id'],
@ -1078,7 +1078,7 @@ class InstanceTestCase(base.ApiTestCase):
self.set_mock_db_items(fakes.DB_INSTANCE_1, fakes.DB_INSTANCE_2,
fakes.DB_IMAGE_ARI_1, fakes.DB_IMAGE_AKI_1,
fakes.DB_VOLUME_2)
self.nova_servers.get.side_effect = (
self.nova.servers.get.side_effect = (
tools.get_by_1st_arg_getter({
fakes.ID_OS_INSTANCE_1: fakes.OS_INSTANCE_1,
fakes.ID_OS_INSTANCE_2: fakes.OS_INSTANCE_2}))

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

@ -25,16 +25,16 @@ from ec2api.tests.unit import tools
class KeyPairCase(base.ApiTestCase):
def test_create_key_pair(self):
self.nova_key_pairs.create.return_value = (
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.nova_key_pairs.create.assert_called_once_with(fakes.NAME_KEY_PAIR)
self.nova.keypairs.create.assert_called_once_with(fakes.NAME_KEY_PAIR)
def test_create_key_pair_invalid(self):
self.nova_key_pairs.create.side_effect = (
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'])
@ -42,14 +42,14 @@ class KeyPairCase(base.ApiTestCase):
resp = self.execute('CreateKeyPair', {'KeyName': 'k' * 256})
self.assertEqual(400, resp['http_status_code'])
self.assertEqual('ValidationError', resp['Error']['Code'])
self.nova_key_pairs.create.side_effect = (
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'])
def test_import_key_pair(self):
self.nova_key_pairs.create.return_value = (
self.nova.keypairs.create.return_value = (
fakes.NovaKeyPair(fakes.OS_KEY_PAIR))
resp = self.execute('ImportKeyPair',
{'KeyName': fakes.NAME_KEY_PAIR,
@ -59,11 +59,11 @@ class KeyPairCase(base.ApiTestCase):
self.assertThat(
tools.purge_dict(fakes.EC2_KEY_PAIR, {'keyMaterial'}),
matchers.DictMatches(tools.purge_dict(resp, {'http_status_code'})))
self.nova_key_pairs.create.assert_called_once_with(
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_key_pairs.create.side_effect = (
self.nova.keypairs.create.side_effect = (
nova_exception.OverLimit(413))
resp = self.execute('ImportKeyPair',
{'KeyName': fakes.NAME_KEY_PAIR,
@ -73,17 +73,17 @@ class KeyPairCase(base.ApiTestCase):
self.assertEqual('ResourceLimitExceeded', resp['Error']['Code'])
def test_delete_key_pair(self):
self.nova_key_pairs.delete.return_value = True
self.nova.keypairs.delete.return_value = True
resp = self.execute('DeleteKeyPair', {'KeyName': fakes.NAME_KEY_PAIR})
self.assertEqual(200, resp['http_status_code'])
self.nova_key_pairs.delete.assert_called_once_with(fakes.NAME_KEY_PAIR)
self.nova_key_pairs.delete.side_effect = nova_exception.NotFound(404)
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.nova_key_pairs.delete.assert_any_call('keyname1')
self.nova.keypairs.delete.assert_any_call('keyname1')
def test_describe_key_pairs(self):
self.nova_key_pairs.list.return_value = [fakes.NovaKeyPair(
self.nova.keypairs.list.return_value = [fakes.NovaKeyPair(
fakes.OS_KEY_PAIR)]
resp = self.execute('DescribeKeyPairs', {})
self.assertEqual(200, resp['http_status_code'])
@ -91,7 +91,7 @@ class KeyPairCase(base.ApiTestCase):
matchers.ListMatches([
tools.purge_dict(fakes.EC2_KEY_PAIR,
{'keyMaterial'})]))
self.nova_key_pairs.list.assert_called_once()
self.nova.keypairs.list.assert_called_once()
self.check_filtering(
'DescribeKeyPairs', 'keySet',
@ -99,9 +99,9 @@ class KeyPairCase(base.ApiTestCase):
('key-name', fakes.NAME_KEY_PAIR)])
def test_describe_key_pairs_invalid(self):
self.nova_key_pairs.list.return_value = [fakes.NovaKeyPair(
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.nova_key_pairs.list.assert_called_once()
self.nova.keypairs.list.assert_called_once()

14
ec2api/tests/unit/test_metadata_api.py
View File

@ -55,16 +55,16 @@ class MetadataApiTestCase(base.ApiTestCase):
self.assertEqual('\n'.join(api.VERSIONS + ['latest']), retval)
def test_get_instance_and_project_id(self):
self.nova_servers.list.return_value = [fakes.OS_INSTANCE_1,
self.nova.servers.list.return_value = [fakes.OS_INSTANCE_1,
fakes.OS_INSTANCE_2]
self.nova_fixed_ips.get.return_value = mock.Mock(hostname='fake_name')
self.nova.fixed_ips.get.return_value = mock.Mock(hostname='fake_name')
self.assertEqual(
(fakes.ID_OS_INSTANCE_1, fakes.ID_OS_PROJECT),
api.get_os_instance_and_project_id(self.fake_context,
fakes.IP_NETWORK_INTERFACE_2))
self.nova_fixed_ips.get.assert_called_with(
self.nova.fixed_ips.get.assert_called_with(
fakes.IP_NETWORK_INTERFACE_2)
self.nova_servers.list.assert_called_with(
self.nova.servers.list.assert_called_with(
search_opts={'hostname': 'fake_name',
'all_tenants': True})
@ -74,11 +74,11 @@ class MetadataApiTestCase(base.ApiTestCase):
self.fake_context,
fakes.IP_NETWORK_INTERFACE_2)
self.nova_servers.list.return_value = [fakes.OS_INSTANCE_2]
self.nova.servers.list.return_value = [fakes.OS_INSTANCE_2]
check_raise()
self.nova_fixed_ips.get.side_effect = nova_exception.NotFound('fake')
self.nova_servers.list.return_value = [fakes.OS_INSTANCE_1,
self.nova.fixed_ips.get.side_effect = nova_exception.NotFound('fake')
self.nova.servers.list.return_value = [fakes.OS_INSTANCE_1,
fakes.OS_INSTANCE_2]
check_raise()

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

@ -466,7 +466,7 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
'InstanceId': fakes.ID_EC2_INSTANCE_1,
'DeviceIndex': '1'})
self.assertEqual(200, resp['http_status_code'])
self.nova_servers.interface_attach.assert_called_once_with(
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(
mock.ANY,
@ -508,7 +508,7 @@ class NetworkInterfaceTestCase(base.ApiTestCase):
self.neutron.list_ports.return_value = (
{'ports': [fakes.OS_PORT_2]})
self.isotime.return_value = fakes.TIME_ATTACH_NETWORK_INTERFACE
self.nova_servers.interface_attach.side_effect = Exception()
self.nova.servers.interface_attach.side_effect = Exception()
self.execute('AttachNetworkInterface',
{'NetworkInterfaceId': fakes.ID_EC2_NETWORK_INTERFACE_1,

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

@ -582,7 +582,7 @@ class RouteTableTestCase(base.ApiTestCase):
fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2,
fakes.DB_INSTANCE_1)
fake_server_class = collections.namedtuple('FakeServer', ['status'])
self.nova_servers.get.return_value = fake_server_class('ACTIVE')
self.nova.servers.get.return_value = fake_server_class('ACTIVE')
resp = self.execute('DescribeRouteTables', {})
self.assertEqual(200, resp['http_status_code'])
@ -648,7 +648,7 @@ class RouteTableTestCase(base.ApiTestCase):
igw_1, igw_2, subnet_1, subnet_2,
fakes.DB_NETWORK_INTERFACE_1, fakes.DB_NETWORK_INTERFACE_2)
fake_server_class = collections.namedtuple('FakeServer', ['status'])
self.nova_servers.get.return_value = fake_server_class('DOWN')
self.nova.servers.get.return_value = fake_server_class('DOWN')
resp = self.execute('DescribeRouteTables', {})
self.assertEqual(200, resp['http_status_code'])
ec2_route_table_1 = copy.deepcopy(fakes.EC2_ROUTE_TABLE_1)

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

@ -31,7 +31,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
security_group.SecurityGroupEngineNeutron())
self.set_mock_db_items(fakes.DB_VPC_1)
self.db_api.add_item.return_value = fakes.DB_SECURITY_GROUP_1
self.nova_security_groups.create.return_value = (
self.nova.security_groups.create.return_value = (
fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_1))
resp = self.execute(
@ -39,9 +39,9 @@ class SecurityGroupTestCase(base.ApiTestCase):
{'GroupName': 'groupname',
'GroupDescription': 'Group description'})
self.assertEqual(200, resp['http_status_code'])
self.nova_security_groups.create.assert_called_once_with(
self.nova.security_groups.create.assert_called_once_with(
'groupname', 'Group description')
self.nova_security_groups.reset_mock()
self.nova.security_groups.reset_mock()
resp = self.execute(
'CreateSecurityGroup',
@ -53,7 +53,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
self.db_api.add_item.assert_called_once_with(
mock.ANY, 'sg',
tools.purge_dict(fakes.DB_SECURITY_GROUP_1, ('id',)))
self.nova_security_groups.create.assert_called_once_with(
self.nova.security_groups.create.assert_called_once_with(
'groupname', 'Group description')
def test_create_security_group_invalid(self):
@ -127,7 +127,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
def test_create_security_group_over_quota(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNeutron())
self.nova_security_groups.create.side_effect = (
self.nova.security_groups.create.side_effect = (
nova_exception.OverLimit(413))
resp = self.execute(
'CreateSecurityGroup',
@ -136,7 +136,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
'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(
self.nova.security_groups.create.assert_called_once_with(
'groupname', 'Group description')
def test_create_security_group_rollback(self):
@ -144,14 +144,14 @@ class SecurityGroupTestCase(base.ApiTestCase):
security_group.SecurityGroupEngineNova())
self.set_mock_db_items(fakes.DB_VPC_1)
self.db_api.add_item.side_effect = Exception()
self.nova_security_groups.create.return_value = (
self.nova.security_groups.create.return_value = (
fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_1))
self.execute(
'CreateSecurityGroup',
{'VpcId': fakes.ID_EC2_VPC_1,
'GroupName': 'groupname',
'GroupDescription': 'Group description'})
self.nova_security_groups.delete.assert_called_once_with(
self.nova.security_groups.delete.assert_called_once_with(
fakes.ID_OS_SECURITY_GROUP_1)
def test_delete_security_group(self):
@ -176,7 +176,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
def test_delete_security_group_nova(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNova())
self.nova_security_groups.list.return_value = (
self.nova.security_groups.list.return_value = (
[fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_1),
fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_2)])
resp = self.execute(
@ -185,7 +185,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
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(
self.nova.security_groups.delete.assert_called_once_with(
fakes.ID_OS_SECURITY_GROUP_2)
# NOTE(Alex) This test is disabled because it checks using non-AWS id.
@ -193,7 +193,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
def test_delete_security_group_nova_os_id(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNova())
self.nova_security_groups.list.return_value = (
self.nova.security_groups.list.return_value = (
[fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_1),
fakes.NovaSecurityGroup(fakes.OS_SECURITY_GROUP_2)])
resp = self.execute(
@ -202,7 +202,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
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(
self.nova.security_groups.delete.assert_called_once_with(
fakes.ID_OS_SECURITY_GROUP_2)
def test_delete_security_group_invalid(self):
@ -297,7 +297,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNova())
self.set_mock_db_items()
self.nova_security_groups.list.return_value = (
self.nova.security_groups.list.return_value = (
[fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_1),
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2)])
resp = self.execute('DescribeSecurityGroups', {})
@ -422,9 +422,9 @@ class SecurityGroupTestCase(base.ApiTestCase):
def test_authorize_security_group_ip_ranges_nova(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNova())
self.nova_security_group_rules.create.return_value = (
self.nova.security_group_rules.create.return_value = (
{'security_group_rule': [fakes.NOVA_SECURITY_GROUP_RULE_1]})
self.nova_security_groups.list.return_value = (
self.nova.security_groups.list.return_value = (
[fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_1),
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2)])
resp = self.execute(
@ -435,7 +435,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
'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(
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)
@ -463,9 +463,9 @@ class SecurityGroupTestCase(base.ApiTestCase):
def test_authorize_security_group_groups_nova(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNova())
self.nova_security_group_rules.create.return_value = (
self.nova.security_group_rules.create.return_value = (
{'security_group_rule': [fakes.NOVA_SECURITY_GROUP_RULE_2]})
self.nova_security_groups.list.return_value = (
self.nova.security_groups.list.return_value = (
[fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_1),
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2)])
resp = self.execute(
@ -475,7 +475,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
'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(
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)
@ -503,12 +503,12 @@ class SecurityGroupTestCase(base.ApiTestCase):
def test_revoke_security_group_ingress_ip_ranges_nova(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNova())
self.nova_security_groups.list.return_value = (
self.nova.security_groups.list.return_value = (
[fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_1),
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2)])
self.nova_security_groups.get.return_value = (
self.nova.security_groups.get.return_value = (
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2))
self.nova_security_group_rules.delete.return_value = True
self.nova.security_group_rules.delete.return_value = True
resp = self.execute(
'RevokeSecurityGroupIngress',
{'GroupName': fakes.EC2_NOVA_SECURITY_GROUP_2['groupName'],
@ -517,7 +517,7 @@ class SecurityGroupTestCase(base.ApiTestCase):
'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(
self.nova.security_group_rules.delete.assert_called_once_with(
fakes.NOVA_SECURITY_GROUP_RULE_1['id'])
def test_revoke_security_group_egress_groups(self):
@ -544,12 +544,12 @@ class SecurityGroupTestCase(base.ApiTestCase):
def test_revoke_security_group_groups_nova(self):
security_group.security_group_engine = (
security_group.SecurityGroupEngineNova())
self.nova_security_groups.list.return_value = (
self.nova.security_groups.list.return_value = (
[fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_1),
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2)])
self.nova_security_groups.get.return_value = (
self.nova.security_groups.get.return_value = (
fakes.NovaSecurityGroup(fakes.NOVA_SECURITY_GROUP_2))
self.nova_security_group_rules.delete.return_value = True
self.nova.security_group_rules.delete.return_value = True
resp = self.execute(
'RevokeSecurityGroupIngress',
{'GroupName': fakes.EC2_NOVA_SECURITY_GROUP_2['groupName'],
@ -557,5 +557,5 @@ class SecurityGroupTestCase(base.ApiTestCase):
'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(
self.nova.security_group_rules.delete.assert_called_once_with(
fakes.NOVA_SECURITY_GROUP_RULE_2['id'])

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

@ -199,7 +199,7 @@ class VolumeTestCase(base.ApiTestCase):
'status': 'attaching',
'volumeId': fakes.ID_EC2_VOLUME_3},
resp)
self.nova_volumes.create_server_volume.assert_called_once_with(
self.nova.volumes.create_server_volume.assert_called_once_with(
fakes.ID_OS_INSTANCE_2, fakes.ID_OS_VOLUME_3, '/dev/vdf')
@mock.patch.object(fakes.CinderVolume, 'get', autospec=True)
@ -219,7 +219,7 @@ class VolumeTestCase(base.ApiTestCase):
'status': 'detaching',
'volumeId': fakes.ID_EC2_VOLUME_2},
resp)
self.nova_volumes.delete_server_volume.assert_called_once_with(
self.nova.volumes.delete_server_volume.assert_called_once_with(
fakes.ID_OS_INSTANCE_2, fakes.ID_OS_VOLUME_2)
self.cinder.volumes.get.assert_called_once_with(fakes.ID_OS_VOLUME_2)