x
/
omni
Code Issues Proposed changes

[AWS] Added Neutron router plugin unit test cases 

Files modified:
- aws_router_plugin.py: Fixed bugs encountered while running test cases

- run_tests.sh: Added support for running AWS test cases for Neutron

- omni-requirements.txt: Changed version of moto and added boto3 package

- test_ec2.py in Cinder and Nova: With latest moto version, '@mock_ec2'
is deprecated. Hence Cinder and Nova test cases were unable to mock
EC2 credentials. Hence used '@mock_ec2_deprecated' instead of '@mock_ec2'

Closes-Bug: #1708585

Change-Id: I51d2bb99ef7d13b75940df6bbb33475de4a4630d
This commit is contained in:
Pratik Shah 2017-08-22 23:04:12 +05:30
parent 6538ef949e
commit 0ac440c7d6
6 changed files with 434 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
cinder/tests/unit/volume/drivers/test_ec2.py
View File

@ -19,12 +19,12 @@ from cinder import test
from cinder.volume.drivers.aws import ebs
from cinder.volume.drivers.aws.exception import AvailabilityZoneNotFound
import mock
from moto import mock_ec2
from moto import mock_ec2_deprecated
from oslo_service import loopingcall
class EBSVolumeTestCase(test.TestCase):
@mock_ec2
@mock_ec2_deprecated
def setUp(self):
super(EBSVolumeTestCase, self).setUp()
ebs.CONF.AWS.region_name = 'us-east-1'
@ -62,7 +62,7 @@ class EBSVolumeTestCase(test.TestCase):
ss['display_name'] = kwargs.get('display_name', 'snapshot_007')
return ss
@mock_ec2
@mock_ec2_deprecated
def test_availability_zone_config(self):
ebs.CONF.AWS.az = 'hgkjhgkd'
driver = ebs.EBSDriver()
@ -70,11 +70,11 @@ class EBSVolumeTestCase(test.TestCase):
self.assertRaises(AvailabilityZoneNotFound, driver.do_setup, ctxt)
ebs.CONF.AWS.az = 'us-east-1a'
@mock_ec2
@mock_ec2_deprecated
def test_volume_create_success(self):
self.assertIsNone(self._driver.create_volume(self._stub_volume()))
@mock_ec2
@mock_ec2_deprecated
@mock.patch('cinder.volume.drivers.aws.ebs.EBSDriver._wait_for_create')
def test_volume_create_fails(self, mock_wait):
def wait(*args):
@ -88,34 +88,34 @@ class EBSVolumeTestCase(test.TestCase):
self.assertRaises(APITimeout, self._driver.create_volume,
self._stub_volume())
@mock_ec2
@mock_ec2_deprecated
def test_volume_deletion(self):
vol = self._stub_volume()
self._driver.create_volume(vol)
self.assertIsNone(self._driver.delete_volume(vol))
@mock_ec2
@mock_ec2_deprecated
@mock.patch('cinder.volume.drivers.aws.ebs.EBSDriver._find')
def test_volume_deletion_not_found(self, mock_find):
vol = self._stub_volume()
mock_find.side_effect = NotFound
self.assertIsNone(self._driver.delete_volume(vol))
@mock_ec2
@mock_ec2_deprecated
def test_snapshot(self):
vol = self._stub_volume()
snapshot = self._stub_snapshot()
self._driver.create_volume(vol)
self.assertIsNone(self._driver.create_snapshot(snapshot))
@mock_ec2
@mock_ec2_deprecated
@mock.patch('cinder.volume.drivers.aws.ebs.EBSDriver._find')
def test_snapshot_volume_not_found(self, mock_find):
mock_find.side_effect = NotFound
ss = self._stub_snapshot()
self.assertRaises(VolumeNotFound, self._driver.create_snapshot, ss)
@mock_ec2
@mock_ec2_deprecated
@mock.patch('cinder.volume.drivers.aws.ebs.EBSDriver._wait_for_snapshot')
def test_snapshot_create_fails(self, mock_wait):
def wait(*args):
@ -130,7 +130,7 @@ class EBSVolumeTestCase(test.TestCase):
self._driver.create_volume(ss['volume'])
self.assertRaises(APITimeout, self._driver.create_snapshot, ss)
@mock_ec2
@mock_ec2_deprecated
def test_volume_from_snapshot(self):
snapshot = self._stub_snapshot()
volume = self._stub_volume()
@ -139,7 +139,7 @@ class EBSVolumeTestCase(test.TestCase):
self.assertIsNone(
self._driver.create_volume_from_snapshot(volume, snapshot))
@mock_ec2
@mock_ec2_deprecated
def test_volume_from_non_existing_snapshot(self):
self.assertRaises(NotFound, self._driver.create_volume_from_snapshot,
self._stub_volume(), self._stub_snapshot())

5
neutron/neutron/services/l3_router/aws_router_plugin.py
View File

@ -12,11 +12,11 @@ under the License.
"""
import neutron_lib
from neutron_lib import constants as const
from distutils.version import LooseVersion
from neutron.common.aws_utils import AwsException
from neutron.common.aws_utils import AwsUtils
from neutron.common import constants as n_const
from neutron.db import common_db_mixin
from neutron.db import extraroute_db
from neutron.db import l3_db
@ -83,6 +83,7 @@ class AwsRouterPlugin(
# FLOATING IP FEATURES
def create_floatingip(self, context, floatingip):
public_ip_allocated = None
try:
response = self.aws_utils.allocate_elastic_ip()
public_ip_allocated = response['PublicIp']
@ -107,7 +108,7 @@ class AwsRouterPlugin(
try:
res = super(AwsRouterPlugin, self).create_floatingip(
context, floatingip,
initial_status=n_const.FLOATINGIP_STATUS_DOWN)
initial_status=const.FLOATINGIP_STATUS_DOWN)
except Exception as e:
LOG.error("Error when adding floating ip in openstack. "
"Deleting Elastic IP: %s" % public_ip_allocated)

291
neutron/tests/services/l3_router/test_ec2.py Normal file
View File

@ -0,0 +1,291 @@
"""
Copyright (c) 2017 Platform9 Systems Inc.
Licensed under the Apache License, Version 2.0 (the "License"); you may
not use this file except in compliance with the License. You may obtain
a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
WARRANTIES OR CONDITIONS OF ANY KIND, either expressed or implied. See the
License for the specific language governing permissions and limitations
under the License.
"""
import mock
from moto import mock_ec2
from neutron.common.aws_utils import AwsException
from neutron.common.aws_utils import AwsUtils
from neutron.common.aws_utils import cfg
from neutron.common import exceptions
from neutron.services.l3_router.aws_router_plugin import AwsRouterPlugin
from neutron.tests import base
from neutron.tests.unit.extensions import test_securitygroup as test_sg
from neutron_lib import constants as const
L3_NAT_DBONLY_MIXIN = 'neutron.db.l3_db.L3_NAT_dbonly_mixin'
L3_NAT_WITH_DVR_DB_MIXIN = 'neutron.db.l3_dvr_db.L3_NAT_with_dvr_db_mixin'
AWS_ROUTER = 'neutron.services.l3_router.aws_router_plugin.AwsRouterPlugin'
class AWSRouterPluginTests(test_sg.SecurityGroupsTestCase, base.BaseTestCase):
@mock_ec2
def setUp(self):
super(AWSRouterPluginTests, self).setUp()
cfg.CONF.AWS.secret_key = 'aws_access_key'
cfg.CONF.AWS.access_key = 'aws_secret_key'
cfg.CONF.AWS.region_name = 'us-east-1'
self._driver = AwsRouterPlugin()
self.context = self._create_fake_context()
def _create_fake_context(self):
context = mock.Mock()
context.current = {}
context.current['id'] = "fake_id_1234"
context.current['cidr'] = "192.168.1.0/24"
context.current['network_id'] = "fake_network_id_1234"
return context
def _get_fake_tags(self):
tags = []
tags.append({'Value': 'fake_name', 'Key': 'Name'})
tags.append({'Value': 'fake_id', 'Key': 'openstack_network_id'})
tags.append({'Value': 'fake_tenant_id', 'Key': 'openstack_tenant_id'})
return tags
def _create_router(self, mock_create):
mock_create.return_value = {'id': 'fake_id'}
router = {'router': {'name': 'fake_name'}}
response = self._driver.create_router(self.context, router)
mock_create.assert_called_once_with(self.context, router)
return response
@mock_ec2
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.create_floatingip')
@mock.patch(AWS_ROUTER + '._associate_floatingip_to_port')
def test_create_floatingip_with_port(self, mock_assoc, mock_create):
floatingip = {'floatingip': {'port_id': 'fake_port_id'}}
mock_assoc.return_value = None
mock_create.return_value = None
self.assertIsNone(self._driver.create_floatingip(self.context,
floatingip))
self.assertTrue(mock_assoc.called)
mock_create.assert_called_once_with(
self.context, floatingip,
initial_status=const.FLOATINGIP_STATUS_DOWN)
@mock_ec2
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.create_floatingip')
def test_create_floatingip_without_port(self, mock_create):
floatingip = {'floatingip': {}}
mock_create.return_value = None
self.assertIsNone(self._driver.create_floatingip(self.context,
floatingip))
mock_create.assert_called_once_with(
self.context, floatingip,
initial_status=const.FLOATINGIP_STATUS_DOWN)
@mock_ec2
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
def test_create_floatingip_with_failure_in_associating(self, mock_get):
floatingip = {'floatingip': {'port_id': 'fake_port_id'}}
port = {'fixed_ips': []}
mock_get.return_value = port
self.assertRaises(AwsException, self._driver.create_floatingip,
self.context, floatingip)
@mock_ec2
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.create_floatingip')
@mock.patch(AWS_ROUTER + '._associate_floatingip_to_port')
def test_create_floatingip_with_failure_in_creating(
self, mock_assoc, mock_create):
floatingip = {'floatingip': {'port_id': 'fake_port_id'}}
mock_create.side_effect = exceptions.PhysicalNetworkNameError()
mock_assoc.return_value = None
self.assertRaises(exceptions.PhysicalNetworkNameError,
self._driver.create_floatingip, self.context,
floatingip)
@mock_ec2
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.update_floatingip')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
def test_update_floatingip_without_port(self, mock_get, mock_update):
fake_id = 'fake_id'
floatingip = {'floatingip': {}}
mock_get.return_value = {}
mock_update.return_value = None
self.assertIsNone(self._driver.update_floatingip(self.context, fake_id,
floatingip))
self.assertTrue(mock_get.called)
mock_get.assert_called_once_with(self.context, fake_id)
mock_update.assert_called_once_with(self.context, fake_id, floatingip)
@mock_ec2
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.delete_floatingip')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
def test_delete_floatingip_success(self, mock_get, mock_delete):
fake_id = 'fake_id'
mock_get.return_value = {'floating_ip_address': '192.169.10.1'}
mock_delete.return_value = None
self.assertIsNone(self._driver.delete_floatingip(self.context,
fake_id))
self.assertTrue(mock_get.called)
mock_get.assert_called_once_with(self.context, fake_id)
mock_delete.assert_called_once_with(self.context, fake_id)
@mock_ec2
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.delete_floatingip')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
def test_delete_floatingip_failure(self, mock_get, mock_delete):
fake_id = 'fake_id'
mock_get.return_value = {'floating_ip_address': '192.169.10.1'}
mock_delete.side_effect = exceptions.PhysicalNetworkNameError()
self.assertRaises(exceptions.PhysicalNetworkNameError,
self._driver.delete_floatingip,
self.context, fake_id)
self.assertTrue(mock_get.called)
mock_get.assert_called_once_with(self.context, fake_id)
@mock_ec2
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.delete_floatingip')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
def test_delete_floatingip_aws_failure(self, mock_get, mock_delete):
fake_id = 'fake_id'
mock_get.return_value = {'floating_ip_address': None}
mock_delete.side_effect = {}
self.assertRaises(AwsException, self._driver.delete_floatingip,
self.context, fake_id)
self.assertTrue(mock_get.called)
mock_get.assert_called_once_with(self.context, fake_id)
@mock_ec2
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.update_floatingip')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
def test_update_floatingip_with_port(self, mock_get, mock_update):
fake_id = 'fake_id'
floatingip = {'floatingip': {'port_id': None}}
mock_get.return_value = {'floating_ip_address': '192.169.10.1'}
mock_update.return_value = None
self.assertIsNone(self._driver.update_floatingip(self.context, fake_id,
floatingip))
self.assertTrue(mock_get.called)
mock_get.assert_called_once_with(self.context, fake_id)
mock_update.assert_called_once_with(self.context, fake_id, floatingip)
@mock_ec2
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.update_floatingip')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
def test_update_floatingip_failure_without_port(self, mock_get,
mock_update):
fake_id = 'fake_id'
floatingip = {'floatingip': {}}
mock_get.return_value = {'floating_ip_address': '192.169.10.1'}
mock_update.side_effect = exceptions.PhysicalNetworkNameError()
self.assertRaises(exceptions.PhysicalNetworkNameError,
self._driver.update_floatingip, self.context,
fake_id, floatingip)
self.assertTrue(mock_get.called)
mock_get.assert_called_once_with(self.context, fake_id)
@mock_ec2
@mock.patch(L3_NAT_WITH_DVR_DB_MIXIN + '.update_floatingip')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.get_floatingip')
def test_update_floatingip_failure_with_port(self, mock_get, mock_update):
fake_id = 'fake_id'
floatingip = {'floatingip': {'port': 'fake_port_id'}}
mock_get.return_value = {'floating_ip_address': '192.169.10.1'}
mock_update.side_effect = exceptions.PhysicalNetworkNameError()
self.assertRaises(exceptions.PhysicalNetworkNameError,
self._driver.update_floatingip, self.context,
fake_id, floatingip)
self.assertTrue(mock_get.called)
mock_get.assert_called_once_with(self.context, fake_id)
@mock_ec2
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
def test_create_router_success(self, mock_create):
response = self._create_router(mock_create)
self.assertIsInstance(response, dict)
@mock_ec2
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
def test_create_router_failure(self, mock_create):
mock_create.side_effect = exceptions.PhysicalNetworkNameError()
router = {'router': {'name': 'fake_name'}}
self.assertRaises(exceptions.PhysicalNetworkNameError,
self._driver.create_router, self.context, router)
@mock_ec2
@mock.patch(L3_NAT_DBONLY_MIXIN + '.delete_router')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
def test_delete_router_success(self, mock_create, mock_delete):
mock_delete.return_value = None
response = self._create_router(mock_create)
self.assertIsNone(self._driver.delete_router(self.context,
response['id']))
self.assertTrue(mock_delete.called)
mock_delete.assert_called_once_with(self.context, response['id'])
@mock_ec2
@mock.patch(L3_NAT_DBONLY_MIXIN + '.delete_router')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
def test_delete_router_failure(self, mock_create, mock_delete):
mock_delete.side_effect = exceptions.PhysicalNetworkNameError()
response = self._create_router(mock_create)
self.assertRaises(
exceptions.PhysicalNetworkNameError, self._driver.delete_router,
self.context, response['id'])
@mock_ec2
@mock.patch(L3_NAT_DBONLY_MIXIN + '.update_router')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
def test_update_router_success(self, mock_create, mock_update):
mock_update.return_value = {'id': "fake_id"}
response = self._create_router(mock_create)
router = {'router': {'name': 'fake_name'}}
response = self._driver.update_router(
self.context, response['id'], router)
self.assertIsInstance(response, dict)
mock_update.assert_called_once_with(
self.context, response['id'], router)
@mock_ec2
@mock.patch(L3_NAT_DBONLY_MIXIN + '.add_router_interface')
@mock.patch(
'neutron.common.aws_utils.AwsUtils.get_vpc_from_neutron_network_id')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
def test_add_router_interface(self, mock_create, mock_get, mock_vpc,
mock_add):
aws_obj = AwsUtils()
vpc_id = aws_obj.create_vpc_and_tags(self.context.current['cidr'],
self._get_fake_tags())
interface_info = {'subnet_id': 'fake_subnet_id'}
response = self._create_router(mock_create)
router_id = response['id']
mock_get.return_value = {'network_id': 'fake_network_id'}
# We need to mock 'get_vpc_from_neutron_network_id' from aws_utils,
# because we need a valid vpc_id when attaching internet gateway.
mock_vpc.return_value = vpc_id
mock_add.return_value = {'id': 'fake_id',
'subnet_id': 'fake_subnet_id'}
response = self._driver.add_router_interface(
self.context, router_id, interface_info)
self.assertIsInstance(response, dict)
mock_add.assert_called_once_with(self.context, router_id,
interface_info)
@mock_ec2
@mock.patch(L3_NAT_DBONLY_MIXIN + '.remove_router_interface')
@mock.patch(L3_NAT_DBONLY_MIXIN + '.create_router')
def test_remove_router_interface(self, mock_create, mock_remove):
response = self._create_router(mock_create)
router_id = response['id']
interface_info = {'port_id': 'fake_port_id'}
mock_remove.return_value = {'id': 'fake_id',
'subnet_id': 'fake_subnet_id'}
response = self._driver.remove_router_interface(
self.context, router_id, interface_info)
self.assertIsInstance(response, dict)
mock_remove.assert_called_once_with(self.context, router_id,
interface_info)

251
nova/tests/unit/virt/ec2/test_ec2.py
View File

@ -18,7 +18,7 @@ import contextlib
import boto
import mock
from moto import mock_cloudwatch
from moto import mock_ec2
from moto import mock_ec2_deprecated
from oslo_utils import uuidutils
from nova.compute import task_states
@ -33,7 +33,7 @@ from nova.virt.ec2 import EC2Driver
class EC2DriverTestCase(test.NoDBTestCase):
@mock_ec2
@mock_ec2_deprecated
@mock_cloudwatch
def setUp(self):
super(EC2DriverTestCase, self).setUp()
@ -72,94 +72,16 @@ class EC2DriverTestCase(test.NoDBTestCase):
instance_id_list = [
x.id for x in instance_list if x.state != 'terminated'
]
self.conn.ec2_conn.stop_instances(
instance_ids=instance_id_list, force=True)
self.conn.ec2_conn.terminate_instances(instance_ids=instance_id_list)
if len(instance_id_list) > 0:
self.conn.ec2_conn.stop_instances(
instance_ids=instance_id_list, force=True)
self.conn.ec2_conn.terminate_instances(
instance_ids=instance_id_list)
self.type_data = None
self.instance = None
self.uuid = None
self.instance_node = None
@mock_ec2
def test_list_instances(self):
for _ in range(0, 5):
self.conn.ec2_conn.run_instances('ami-1234abc')
fake_list = self.conn.list_instances()
self.assertEqual(5, len(fake_list))
self.reset()
@mock_ec2
def test_add_ssh_keys_key_exists(self):
fake_key = 'fake_key'
fake_key_data = 'abcdefgh'
self.conn.ec2_conn.import_key_pair(fake_key, fake_key_data)
with contextlib.nested(
mock.patch.object(boto.ec2.EC2Connection, 'get_key_pair'),
mock.patch.object(boto.ec2.EC2Connection, 'import_key_pair'),
) as (fake_get, fake_import):
fake_get.return_value = True
self.conn._add_ssh_keys(fake_key, fake_key_data)
fake_get.assert_called_once_with(fake_key)
fake_import.assert_not_called()
@mock_ec2
def test_add_ssh_keys_key_absent(self):
fake_key = 'fake_key'
fake_key_data = 'abcdefgh'
with contextlib.nested(
mock.patch.object(boto.ec2.EC2Connection, 'get_key_pair'),
mock.patch.object(boto.ec2.EC2Connection, 'import_key_pair'),
) as (fake_get, fake_import):
fake_get.return_value = False
self.conn._add_ssh_keys(fake_key, fake_key_data)
fake_get.assert_called_once_with(fake_key)
fake_import.assert_called_once_with(fake_key, fake_key_data)
def test_process_network_info(self):
fake_network_info = [{
'profile': {},
'ovs_interfaceid': None,
'preserve_on_delete': False,
'network': {
'bridge': None,
'subnets': [{
'ips': [{'meta': {},
'version': 4,
'type': 'fixed',
'floating_ips': [],
'address': u'192.168.100.5'}],
'version': 4,
'meta': {},
'dns': [],
'routes': [],
'cidr': u'192.168.100.0/24',
'gateway': {'meta': {},
'version': 4,
'type': 'gateway',
'address': u'192.168.100.1'}}],
'meta': {'injected': True,
'tenant_id': '135b1a036a51414ea1f989ab59fefde5'},
'id': '4f8ad58d-de60-4b52-94ba-8b988a9b7f33',
'label': 'test'},
'devname': 'tapa9a90cf6-62',
'vnic_type': 'normal',
'qbh_params': None,
'meta': {},
'details': '{"subnet_id": "subnet-0107db5a",'
' "ip_address": "192.168.100.5"}',
'address': 'fa:16:3e:23:65:2c',
'active': True,
'type': 'vip_type_a',
'id': 'a9a90cf6-627c-46f3-829d-c5a2ae07aaf0',
'qbg_params': None
}]
aws_subnet_id, aws_fixed_ip, port_id, network_id = \
self.conn._process_network_info(fake_network_info)
self.assertEqual(aws_subnet_id, 'subnet-0107db5a')
self.assertEqual(aws_fixed_ip, '192.168.100.5')
self.assertEqual(port_id, 'a9a90cf6-627c-46f3-829d-c5a2ae07aaf0')
self.assertEqual(network_id, '4f8ad58d-de60-4b52-94ba-8b988a9b7f33')
def _get_instance_flavor_details(self):
return {'memory_mb': 2048.0,
'root_gb': 0,
@ -218,7 +140,102 @@ class EC2DriverTestCase(test.NoDBTestCase):
admin_password=None, network_info=None,
block_device_info=None)
@mock_ec2
@mock_ec2_deprecated
def _create_vm_in_aws_nova(self):
self._create_instance()
self._create_network()
with contextlib.nested(
mock.patch.object(EC2Driver, '_get_image_ami_id_from_meta'),
mock.patch.object(EC2Driver, '_process_network_info'),
mock.patch.object(EC2Driver, '_get_instance_sec_grps'),
) as (mock_image, mock_network, mock_secgrp):
mock_image.return_value = 'ami-1234abc'
mock_network.return_value = (self.subnet_id, '192.168.10.5', None,
None)
mock_secgrp.return_value = []
self._create_nova_vm()
@mock_ec2_deprecated
def test_list_instances(self):
for _ in range(0, 5):
self.conn.ec2_conn.run_instances('ami-1234abc')
fake_list = self.conn.list_instances()
self.assertEqual(5, len(fake_list))
self.reset()
@mock_ec2_deprecated
def test_add_ssh_keys_key_exists(self):
fake_key = 'fake_key'
fake_key_data = 'abcdefgh'
self.conn.ec2_conn.import_key_pair(fake_key, fake_key_data)
with contextlib.nested(
mock.patch.object(boto.ec2.EC2Connection, 'get_key_pair'),
mock.patch.object(boto.ec2.EC2Connection, 'import_key_pair'),
) as (fake_get, fake_import):
fake_get.return_value = True
self.conn._add_ssh_keys(fake_key, fake_key_data)
fake_get.assert_called_once_with(fake_key)
fake_import.assert_not_called()
@mock_ec2_deprecated
def test_add_ssh_keys_key_absent(self):
fake_key = 'fake_key'
fake_key_data = 'abcdefgh'
with contextlib.nested(
mock.patch.object(boto.ec2.EC2Connection, 'get_key_pair'),
mock.patch.object(boto.ec2.EC2Connection, 'import_key_pair'),
) as (fake_get, fake_import):
fake_get.return_value = False
self.conn._add_ssh_keys(fake_key, fake_key_data)
fake_get.assert_called_once_with(fake_key)
fake_import.assert_called_once_with(fake_key, fake_key_data)
def test_process_network_info(self):
fake_network_info = [{
'profile': {},
'ovs_interfaceid': None,
'preserve_on_delete': False,
'network': {
'bridge': None,
'subnets': [{
'ips': [{'meta': {},
'version': 4,
'type': 'fixed',
'floating_ips': [],
'address': u'192.168.100.5'}],
'version': 4,
'meta': {},
'dns': [],
'routes': [],
'cidr': u'192.168.100.0/24',
'gateway': {'meta': {},
'version': 4,
'type': 'gateway',
'address': u'192.168.100.1'}}],
'meta': {'injected': True,
'tenant_id': '135b1a036a51414ea1f989ab59fefde5'},
'id': '4f8ad58d-de60-4b52-94ba-8b988a9b7f33',
'label': 'test'},
'devname': 'tapa9a90cf6-62',
'vnic_type': 'normal',
'qbh_params': None,
'meta': {},
'details': '{"subnet_id": "subnet-0107db5a",'
' "ip_address": "192.168.100.5"}',
'address': 'fa:16:3e:23:65:2c',
'active': True,
'type': 'vip_type_a',
'id': 'a9a90cf6-627c-46f3-829d-c5a2ae07aaf0',
'qbg_params': None
}]
aws_subnet_id, aws_fixed_ip, port_id, network_id = \
self.conn._process_network_info(fake_network_info)
self.assertEqual(aws_subnet_id, 'subnet-0107db5a')
self.assertEqual(aws_fixed_ip, '192.168.100.5')
self.assertEqual(port_id, 'a9a90cf6-627c-46f3-829d-c5a2ae07aaf0')
self.assertEqual(network_id, '4f8ad58d-de60-4b52-94ba-8b988a9b7f33')
@mock_ec2_deprecated
def test_spawn(self):
self._create_instance()
self._create_network()
@ -245,7 +262,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
self.assertEqual(inst.instance_type, 't2.small')
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_spawn_with_key(self):
self._create_instance(key_name='fake_key', key_data='fake_key_data')
self._create_network()
@ -265,12 +282,12 @@ class EC2DriverTestCase(test.NoDBTestCase):
self.assertEqual(inst.key_name, 'fake_key')
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_spawn_with_userdata(self):
userdata = """
userdata = '''
#cloud-config
password: password
"""
'''
b64encoded = base64.b64encode(userdata)
self._create_instance(user_data=b64encoded)
self._create_network()
@ -298,7 +315,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
security_group_ids=[])
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_spawn_with_network_error(self):
self._create_instance()
with contextlib.nested(
@ -313,7 +330,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
self._create_nova_vm)
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_spawn_with_network_error_from_aws(self):
self._create_instance()
with contextlib.nested(
@ -322,14 +339,13 @@ class EC2DriverTestCase(test.NoDBTestCase):
mock.patch.object(EC2Driver, '_get_instance_sec_grps'),
) as (mock_image, mock_network, mock_secgrp):
mock_image.return_value = 'ami-1234abc'
mock_network.return_value = ('subnet-1234abc', '192.168.10.5',
None, None)
mock_network.return_value = (None, '192.168.10.5', None, None)
mock_secgrp.return_value = []
self.assertRaises(exception.BuildAbortException,
self._create_nova_vm)
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_spawn_with_image_error(self):
self._create_instance()
self._create_network()
@ -346,22 +362,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
self._create_nova_vm)
self.reset()
@mock_ec2
def _create_vm_in_aws_nova(self):
self._create_instance()
self._create_network()
with contextlib.nested(
mock.patch.object(EC2Driver, '_get_image_ami_id_from_meta'),
mock.patch.object(EC2Driver, '_process_network_info'),
mock.patch.object(EC2Driver, '_get_instance_sec_grps'),
) as (mock_image, mock_network, mock_secgrp):
mock_image.return_value = 'ami-1234abc'
mock_network.return_value = (self.subnet_id, '192.168.10.5', None,
None)
mock_secgrp.return_value = []
self._create_nova_vm()
@mock_ec2
@mock_ec2_deprecated
def test_snapshot(self):
self._create_vm_in_aws_nova()
GlanceImageServiceV2.update = mock.Mock()
@ -382,7 +383,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
self.assertEqual(aws_img.id, metadata['properties']['ec2_image_id'])
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_snapshot_instance_not_found(self):
boto.ec2.EC2Connection.create_image = mock.Mock()
self._create_instance()
@ -398,7 +399,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
boto.ec2.EC2Connection.create_image.assert_not_called()
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_reboot_soft(self):
boto.ec2.EC2Connection.reboot_instances = mock.Mock()
self._create_vm_in_aws_nova()
@ -408,7 +409,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
instance_ids=[fake_inst.id], dry_run=False)
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_reboot_hard(self):
self._create_vm_in_aws_nova()
fake_inst = self.fake_ec2_conn.get_only_instances()[0]
@ -428,7 +429,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
self.assertEqual(fake_inst.id, wait_state_calls[0][0][1])
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_reboot_instance_not_found(self):
self._create_instance()
boto.ec2.EC2Connection.stop_instances = mock.Mock()
@ -438,7 +439,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
boto.ec2.EC2Connection.stop_instances.assert_not_called()
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_power_off(self):
self._create_vm_in_aws_nova()
fake_inst = self.fake_ec2_conn.get_only_instances()[0]
@ -448,14 +449,14 @@ class EC2DriverTestCase(test.NoDBTestCase):
self.assertEqual(fake_inst.state, 'stopped')
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_power_off_instance_not_found(self):
self._create_instance()
self.assertRaises(exception.InstanceNotFound, self.conn.power_off,
self.instance)
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_power_on(self):
self._create_vm_in_aws_nova()
fake_inst = self.fake_ec2_conn.get_only_instances()[0]
@ -465,14 +466,14 @@ class EC2DriverTestCase(test.NoDBTestCase):
self.assertEqual(fake_inst.state, 'running')
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_power_on_instance_not_found(self):
self._create_instance()
self.assertRaises(exception.InstanceNotFound, self.conn.power_on,
self.context, self.instance, None, None)
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_destroy(self):
self._create_vm_in_aws_nova()
self.conn.destroy(self.context, self.instance, None, None)
@ -480,7 +481,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
self.assertEqual('terminated', fake_instance.state)
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_destroy_instance_not_found(self):
self._create_instance()
with contextlib.nested(
@ -494,7 +495,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
fake_wait.assert_not_called()
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_destory_instance_terminated_on_aws(self):
self._create_vm_in_aws_nova()
fake_instances = self.fake_ec2_conn.get_only_instances()
@ -512,7 +513,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
fake_wait.assert_not_called()
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_destroy_instance_shut_down_on_aws(self):
self._create_vm_in_aws_nova()
fake_instances = self.fake_ec2_conn.get_only_instances()
@ -528,7 +529,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
instance_ids=[fake_instances[0].id])
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_get_info(self):
self._create_vm_in_aws_nova()
vm_info = self.conn.get_info(self.instance)
@ -536,7 +537,7 @@ class EC2DriverTestCase(test.NoDBTestCase):
self.assertEqual(self.instance.id, vm_info.id)
self.reset()
@mock_ec2
@mock_ec2_deprecated
def test_get_info_instance_not_found(self):
self._create_instance()
self.assertRaises(exception.InstanceNotFound, self.conn.get_info,

2
omni-requirements.txt
View File

@ -1,5 +1,5 @@
google-api-python-client>=1.4.2 # Apache-2.0
moto==0.4.27
moto>=1.0.1
boto>=2.32.1 # MIT
ipaddr
google_compute_engine

2
run_tests.sh
View File

@ -88,7 +88,7 @@ echo "============Running tests============"
run_tests cinder "$GCE_TEST|$AWS_TEST" &
run_tests nova "$GCE_TEST|$AWS_NOVA_TEST" &
run_tests glance_store "$GCE_TEST" &
run_tests neutron "$GCE_TEST" &
run_tests neutron "$GCE_TEST|$AWS_TEST" &
wait
check_results cinder