Shared filesystem management project for OpenStack.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

2410 lines
110 KiB

# Copyright (c) 2014 NetApp, Inc.
# Copyright (c) 2015 Mirantis, Inc.
# All Rights Reserved.
#
# 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 express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Unit tests for the instance module."""
import os
import time
from unittest import mock
import ddt
import netaddr
from oslo_config import cfg
from oslo_utils import importutils
import six
from manila import exception
from manila.share import configuration
from manila.share import driver # noqa
from manila.share.drivers import service_instance
from manila import test
from manila.tests import fake_compute
from manila.tests import fake_network
from manila.tests import utils as test_utils
CONF = cfg.CONF
def fake_get_config_option(key):
if key == 'driver_handles_share_servers':
return True
elif key == 'service_instance_password':
return None
elif key == 'service_instance_user':
return 'fake_user'
elif key == 'service_network_name':
return 'fake_service_network_name'
elif key == 'service_instance_flavor_id':
return '100'
elif key == 'service_instance_name_template':
return 'fake_manila_service_instance_%s'
elif key == 'service_image_name':
return 'fake_service_image_name'
elif key == 'manila_service_keypair_name':
return 'fake_manila_service_keypair_name'
elif key == 'path_to_private_key':
return 'fake_path_to_private_key'
elif key == 'path_to_public_key':
return 'fake_path_to_public_key'
elif key == 'max_time_to_build_instance':
return 500
elif key == 'connect_share_server_to_tenant_network':
return False
elif key == 'service_network_cidr':
return '99.254.0.0/24'
elif key == 'service_network_division_mask':
return 27
elif key == 'service_network_name':
return 'fake_service_network_name'
elif key == 'interface_driver':
return 'i.am.fake.VifDriver'
elif key == 'admin_network_id':
return None
elif key == 'admin_subnet_id':
return None
elif key == 'backend_availability_zone':
return None
else:
return mock.Mock()
class FakeServiceInstance(object):
def __init__(self, driver_config=None):
super(FakeServiceInstance, self).__init__()
self.compute_api = service_instance.compute.API()
self.admin_context = service_instance.context.get_admin_context()
self.driver_config = driver_config
def get_config_option(self, key):
return fake_get_config_option(key)
class FakeNetworkHelper(service_instance.BaseNetworkhelper):
@property
def NAME(self):
return service_instance.NEUTRON_NAME
@property
def neutron_api(self):
if not hasattr(self, '_neutron_api'):
self._neutron_api = mock.Mock()
return self._neutron_api
def __init__(self, service_instance_manager):
self.get_config_option = service_instance_manager.get_config_option
def get_network_name(self, network_info):
"""Return name of network."""
return 'fake_network_name'
def setup_connectivity_with_service_instances(self):
"""Nothing to do in fake network helper."""
def setup_network(self, network_info):
"""Combine fake network data."""
return dict()
def teardown_network(self, server_details):
"""Nothing to do in fake network helper."""
@ddt.ddt
class ServiceInstanceManagerTestCase(test.TestCase):
"""Test suite for service instance manager."""
def setUp(self):
super(ServiceInstanceManagerTestCase, self).setUp()
self.instance_id = 'fake_instance_id'
self.config = configuration.Configuration(None)
self.config.safe_get = mock.Mock(side_effect=fake_get_config_option)
self.mock_object(service_instance.compute, 'API', fake_compute.API)
self.mock_object(
service_instance.os.path, 'exists', mock.Mock(return_value=True))
self.mock_object(service_instance, 'NeutronNetworkHelper',
mock.Mock(side_effect=FakeNetworkHelper))
self._manager = service_instance.ServiceInstanceManager(self.config)
self._manager._execute = mock.Mock(return_value=('', ''))
self.mock_object(time, 'sleep')
def test_get_config_option_from_driver_config(self):
username1 = 'fake_username_1_%s' % self.id()
username2 = 'fake_username_2_%s' % self.id()
config_data = dict(
DEFAULT=dict(service_instance_user=username1),
CUSTOM=dict(service_instance_user=username2))
with test_utils.create_temp_config_with_opts(config_data):
self.config = configuration.Configuration(
service_instance.common_opts, config_group='CUSTOM')
self._manager = service_instance.ServiceInstanceManager(
self.config)
result = self._manager.get_config_option('service_instance_user')
self.assertEqual(username2, result)
def test_get_config_option_from_common_config(self):
username = 'fake_username_%s' % self.id()
config_data = dict(DEFAULT=dict(service_instance_user=username))
with test_utils.create_temp_config_with_opts(config_data):
self._manager = service_instance.ServiceInstanceManager()
result = self._manager.get_config_option('service_instance_user')
self.assertEqual(username, result)
def test_get_neutron_network_helper(self):
# Mock it again, because it was called in setUp method.
self.mock_object(service_instance, 'NeutronNetworkHelper')
config_data = dict(DEFAULT=dict(service_instance_user='fake_username',
driver_handles_share_servers=True))
with test_utils.create_temp_config_with_opts(config_data):
self._manager = service_instance.ServiceInstanceManager()
self._manager.network_helper
service_instance.NeutronNetworkHelper.assert_called_once_with(
self._manager)
def test_init_with_driver_config_and_handling_of_share_servers(self):
self.mock_object(service_instance, 'NeutronNetworkHelper')
config_data = dict(CUSTOM=dict(
driver_handles_share_servers=True,
service_instance_user='fake_user'))
opts = service_instance.common_opts + driver.share_opts
with test_utils.create_temp_config_with_opts(config_data):
self.config = configuration.Configuration(opts, 'CUSTOM')
self._manager = service_instance.ServiceInstanceManager(
self.config)
self.assertTrue(
self._manager.get_config_option("driver_handles_share_servers"))
self.assertIsNotNone(self._manager.driver_config)
self.assertTrue(hasattr(self._manager, 'network_helper'))
self.assertTrue(service_instance.NeutronNetworkHelper.called)
def test_init_with_driver_config_and_wo_handling_of_share_servers(self):
self.mock_object(service_instance, 'NeutronNetworkHelper')
config_data = dict(CUSTOM=dict(
driver_handles_share_servers=False,
service_instance_user='fake_user'))
opts = service_instance.common_opts + driver.share_opts
with test_utils.create_temp_config_with_opts(config_data):
self.config = configuration.Configuration(opts, 'CUSTOM')
self._manager = service_instance.ServiceInstanceManager(
self.config)
self.assertIsNotNone(self._manager.driver_config)
self.assertFalse(hasattr(self._manager, 'network_helper'))
self.assertFalse(service_instance.NeutronNetworkHelper.called)
def test_init_with_common_config_and_handling_of_share_servers(self):
self.mock_object(service_instance, 'NeutronNetworkHelper')
config_data = dict(DEFAULT=dict(
service_instance_user='fake_username',
driver_handles_share_servers=True))
with test_utils.create_temp_config_with_opts(config_data):
self._manager = service_instance.ServiceInstanceManager()
self.assertTrue(
self._manager.get_config_option("driver_handles_share_servers"))
self.assertIsNone(self._manager.driver_config)
self.assertTrue(hasattr(self._manager, 'network_helper'))
self.assertTrue(service_instance.NeutronNetworkHelper.called)
def test_init_with_common_config_and_wo_handling_of_share_servers(self):
self.mock_object(service_instance, 'NeutronNetworkHelper')
config_data = dict(DEFAULT=dict(
service_instance_user='fake_username',
driver_handles_share_servers=False))
with test_utils.create_temp_config_with_opts(config_data):
self._manager = service_instance.ServiceInstanceManager()
self.assertEqual(
False,
self._manager.get_config_option("driver_handles_share_servers"))
self.assertIsNone(self._manager.driver_config)
self.assertFalse(hasattr(self._manager, 'network_helper'))
self.assertFalse(service_instance.NeutronNetworkHelper.called)
def test_no_service_user_defined(self):
group_name = 'GROUP_%s' % self.id()
config_data = {group_name: dict()}
with test_utils.create_temp_config_with_opts(config_data):
config = configuration.Configuration(
service_instance.common_opts, config_group=group_name)
self.assertRaises(
exception.ServiceInstanceException,
service_instance.ServiceInstanceManager, config)
def test_get_service_instance_name_using_driver_config(self):
fake_server_id = 'fake_share_server_id_%s' % self.id()
self.mock_object(service_instance, 'NeutronNetworkHelper')
config_data = dict(CUSTOM=dict(
driver_handles_share_servers=True,
service_instance_user='fake_user'))
opts = service_instance.common_opts + driver.share_opts
with test_utils.create_temp_config_with_opts(config_data):
self.config = configuration.Configuration(opts, 'CUSTOM')
self._manager = service_instance.ServiceInstanceManager(
self.config)
result = self._manager._get_service_instance_name(fake_server_id)
self.assertIsNotNone(self._manager.driver_config)
self.assertEqual(
self._manager.get_config_option(
"service_instance_name_template") % "%s_%s" % (
self._manager.driver_config.config_group, fake_server_id),
result)
self.assertTrue(
self._manager.get_config_option("driver_handles_share_servers"))
self.assertTrue(hasattr(self._manager, 'network_helper'))
self.assertTrue(service_instance.NeutronNetworkHelper.called)
def test_get_service_instance_name_using_default_config(self):
fake_server_id = 'fake_share_server_id_%s' % self.id()
config_data = dict(CUSTOM=dict(
service_instance_user='fake_user'))
with test_utils.create_temp_config_with_opts(config_data):
self._manager = service_instance.ServiceInstanceManager()
result = self._manager._get_service_instance_name(fake_server_id)
self.assertIsNone(self._manager.driver_config)
self.assertEqual(
self._manager.get_config_option(
"service_instance_name_template") % fake_server_id, result)
def test__check_server_availability_available_from_start(self):
fake_server = dict(id='fake_server', ip='127.0.0.1')
self.mock_object(service_instance.socket.socket, 'connect')
self.mock_object(service_instance.time, 'sleep')
self.mock_object(service_instance.time, 'time',
mock.Mock(return_value=0))
result = self._manager._check_server_availability(fake_server)
self.assertTrue(result)
service_instance.socket.socket.connect.assert_called_once_with(
(fake_server['ip'], 22))
service_instance.time.time.assert_has_calls([
mock.call(), mock.call()])
service_instance.time.time.assert_has_calls([])
@ddt.data(True, False)
def test__check_server_availability_with_recall(self, is_ok):
fake_server = dict(id='fake_server', ip='fake_ip_address')
self.fake_time = 0
def fake_connect(addr):
if not(is_ok and self.fake_time > 1):
raise service_instance.socket.error
def fake_time():
return self.fake_time
def fake_sleep(time):
self.fake_time += 5
self.mock_object(service_instance.time, 'sleep',
mock.Mock(side_effect=fake_sleep))
self.mock_object(service_instance.socket.socket, 'connect',
mock.Mock(side_effect=fake_connect))
self.mock_object(service_instance.time, 'time',
mock.Mock(side_effect=fake_time))
self._manager.max_time_to_build_instance = 6
result = self._manager._check_server_availability(fake_server)
if is_ok:
self.assertTrue(result)
else:
self.assertFalse(result)
service_instance.socket.socket.connect.assert_has_calls([
mock.call((fake_server['ip'], 22)),
mock.call((fake_server['ip'], 22))])
service_instance.time.time.assert_has_calls([
mock.call(), mock.call(), mock.call()])
service_instance.time.time.assert_has_calls([mock.call()])
def test_get_server_ip_found_in_networks_section(self):
ip = '10.0.0.1'
net_name = self._manager.get_config_option('service_network_name')
fake_server = dict(networks={net_name: [ip]})
result = self._manager._get_server_ip(fake_server, net_name)
self.assertEqual(ip, result)
def test_get_server_ip_found_in_addresses_section(self):
ip = '10.0.0.1'
net_name = self._manager.get_config_option('service_network_name')
fake_server = dict(addresses={net_name: [dict(addr=ip, version=4)]})
result = self._manager._get_server_ip(fake_server, net_name)
self.assertEqual(ip, result)
@ddt.data(
{},
{'networks': {fake_get_config_option('service_network_name'): []}},
{'addresses': {fake_get_config_option('service_network_name'): []}})
def test_get_server_ip_not_found(self, data):
self.assertRaises(
exception.ManilaException,
self._manager._get_server_ip, data,
fake_get_config_option('service_network_name'))
def test_security_group_name_not_specified(self):
self.mock_object(self._manager, 'get_config_option',
mock.Mock(return_value=None))
result = self._manager._get_or_create_security_groups(
self._manager.admin_context)
self.assertIsNone(result)
self._manager.get_config_option.assert_called_once_with(
'service_instance_security_group')
def test_security_group_name_from_config_and_sg_exist(self):
name = "fake_sg_name_from_config"
desc = "fake_sg_description"
fake_secgroup = {'id': 'fake_sg_id', 'name': name, 'description': desc}
self.mock_object(self._manager, 'get_config_option',
mock.Mock(return_value=name))
neutron_api = self._manager.network_helper.neutron_api
neutron_api.security_group_list.return_value = {
'security_groups': [fake_secgroup]}
result = self._manager._get_or_create_security_groups(
self._manager.admin_context)
self.assertEqual([fake_secgroup, ], result)
self._manager.get_config_option.assert_called_once_with(
'service_instance_security_group')
neutron_api.security_group_list.assert_called_once_with({"name": name})
@ddt.data(None, 'fake_name')
def test_security_group_creation_with_name_from_config(self, name):
config_name = "fake_sg_name_from_config"
desc = "fake_sg_description"
fake_secgroup = {'id': 'fake_sg_id', 'name': name, 'description': desc}
self.mock_object(self._manager, 'get_config_option',
mock.Mock(return_value=name or config_name))
neutron_api = self._manager.network_helper.neutron_api
neutron_api.security_group_list.return_value = {'security_groups': []}
neutron_api.security_group_create.return_value = {
'security_group': fake_secgroup,
}
result = self._manager._get_or_create_security_groups(
context=self._manager.admin_context,
name=name,
description=desc,
)
self.assertEqual([fake_secgroup, ], result)
if not name:
self._manager.get_config_option.assert_called_once_with(
'service_instance_security_group')
neutron_api.security_group_list.assert_called_once_with(
{"name": name or config_name})
neutron_api.security_group_create.assert_called_once_with(
name or config_name, desc)
@ddt.data(None, 'fake_name')
def test_security_group_creation_with_name_from_conf_allow_ssh(self, name):
def fake_secgroup(*args, **kwargs):
return {'security_group': {'id': 'fake_sg_id', 'name': args[0],
'description': args[1]}}
config_name = "fake_sg_name_from_config"
desc = "fake_sg_description"
self.mock_object(self._manager, 'get_config_option',
mock.Mock(return_value=name or config_name))
neutron_api = self._manager.network_helper.neutron_api
neutron_api.security_group_list.return_value = {'security_groups': []}
self.mock_object(neutron_api, 'security_group_create',
mock.Mock(side_effect=fake_secgroup))
fake_ssh_allow_subnet = dict(cidr="10.254.0.1/24",
id='allow_subnet_id')
ssh_sg_name = 'manila-service-subnet-{}'.format(
fake_ssh_allow_subnet['id'])
result = self._manager._get_or_create_security_groups(
context=self._manager.admin_context,
name=name,
description=desc,
allow_ssh_subnet=fake_ssh_allow_subnet
)
self.assertEqual([fake_secgroup(name if name else config_name,
desc)['security_group'],
fake_secgroup(ssh_sg_name, desc)['security_group']],
result)
if not name:
self._manager.get_config_option.assert_called_with(
'service_instance_security_group')
neutron_api.security_group_list.assert_has_calls([
mock.call({"name": name or config_name}),
mock.call({"name": ssh_sg_name})])
neutron_api.security_group_create.assert_has_calls([
mock.call(name or config_name, desc),
mock.call(ssh_sg_name, desc)])
def test_security_group_limit_ssh_invalid_subnet(self):
def fake_secgroup(*args, **kwargs):
return {'security_group': {'id': 'fake_sg_id', 'name': args[0],
'description': args[1]}}
config_name = "fake_sg_name_from_config"
desc = "fake_sg_description"
self.mock_object(self._manager, 'get_config_option',
mock.Mock(config_name))
neutron_api = self._manager.network_helper.neutron_api
neutron_api.security_group_list.return_value = {'security_groups': []}
self.mock_object(neutron_api, 'security_group_create',
mock.Mock(side_effect=fake_secgroup))
fake_ssh_allow_subnet = dict(id='allow_subnet_id')
self.assertRaises(exception.ManilaException,
self._manager._get_or_create_security_groups,
context=self._manager.admin_context,
name=None,
description=desc,
allow_ssh_subnet=fake_ssh_allow_subnet)
def test_security_group_two_sg_in_list(self):
name = "fake_name"
fake_secgroup1 = {'id': 'fake_sg_id1', 'name': name}
fake_secgroup2 = {'id': 'fake_sg_id2', 'name': name}
neutron_api = self._manager.network_helper.neutron_api
neutron_api.security_group_list.return_value = {
'security_groups': [fake_secgroup1, fake_secgroup2]}
self.assertRaises(exception.ServiceInstanceException,
self._manager._get_or_create_security_groups,
self._manager.admin_context,
name)
neutron_api.security_group_list.assert_called_once_with(
{"name": name})
@ddt.data(
dict(),
dict(service_port_id='fake_service_port_id'),
dict(public_port_id='fake_public_port_id'),
dict(service_port_id='fake_service_port_id',
public_port_id='fake_public_port_id'),
)
def test_set_up_service_instance(self, update_data):
fake_network_info = {'foo': 'bar', 'server_id': 'fake_server_id'}
fake_server = {
'id': 'fake', 'ip': '1.2.3.4', 'public_address': '1.2.3.4',
'pk_path': None, 'subnet_id': 'fake-subnet-id',
'router_id': 'fake-router-id',
'username': self._manager.get_config_option(
'service_instance_user'),
'admin_ip': 'admin_ip'}
fake_server.update(update_data)
expected_details = fake_server.copy()
expected_details.pop('pk_path')
expected_details['instance_id'] = expected_details.pop('id')
expected_instance_name = self._manager._get_service_instance_name(
fake_network_info['server_id'])
self.mock_object(self._manager, '_create_service_instance',
mock.Mock(return_value=fake_server))
self.mock_object(self._manager, '_check_server_availability')
result = self._manager.set_up_service_instance(
self._manager.admin_context, fake_network_info)
self._manager._create_service_instance.assert_called_once_with(
self._manager.admin_context,
expected_instance_name, fake_network_info)
self._manager._check_server_availability.assert_called_once_with(
expected_details)
self.assertEqual(expected_details, result)
def test_set_up_service_instance_not_available(self):
fake_network_info = {'foo': 'bar', 'server_id': 'fake_server_id'}
fake_server = {
'id': 'fake', 'ip': '1.2.3.4', 'public_address': '1.2.3.4',
'pk_path': None, 'subnet_id': 'fake-subnet-id',
'router_id': 'fake-router-id',
'username': self._manager.get_config_option(
'service_instance_user'),
'admin_ip': 'admin_ip'}
expected_details = fake_server.copy()
expected_details.pop('pk_path')
expected_details['instance_id'] = expected_details.pop('id')
expected_instance_name = self._manager._get_service_instance_name(
fake_network_info['server_id'])
self.mock_object(self._manager, '_create_service_instance',
mock.Mock(return_value=fake_server))
self.mock_object(self._manager, '_check_server_availability',
mock.Mock(return_value=False))
result = self.assertRaises(
exception.ServiceInstanceException,
self._manager.set_up_service_instance,
self._manager.admin_context, fake_network_info)
self.assertTrue(hasattr(result, 'detail_data'))
self.assertEqual(
{'server_details': expected_details}, result.detail_data)
self._manager._create_service_instance.assert_called_once_with(
self._manager.admin_context,
expected_instance_name, fake_network_info)
self._manager._check_server_availability.assert_called_once_with(
expected_details)
def test_ensure_server(self):
server_details = {'instance_id': 'fake_inst_id', 'ip': '1.2.3.4'}
fake_server = fake_compute.FakeServer()
self.mock_object(self._manager, '_check_server_availability',
mock.Mock(return_value=True))
self.mock_object(self._manager.compute_api, 'server_get',
mock.Mock(return_value=fake_server))
result = self._manager.ensure_service_instance(
self._manager.admin_context, server_details)
self._manager.compute_api.server_get.assert_called_once_with(
self._manager.admin_context, server_details['instance_id'])
self._manager._check_server_availability.assert_called_once_with(
server_details)
self.assertTrue(result)
def test_ensure_server_not_exists(self):
server_details = {'instance_id': 'fake_inst_id', 'ip': '1.2.3.4'}
self.mock_object(self._manager, '_check_server_availability',
mock.Mock(return_value=True))
self.mock_object(self._manager.compute_api, 'server_get',
mock.Mock(side_effect=exception.InstanceNotFound(
instance_id=server_details['instance_id'])))
result = self._manager.ensure_service_instance(
self._manager.admin_context, server_details)
self._manager.compute_api.server_get.assert_called_once_with(
self._manager.admin_context, server_details['instance_id'])
self.assertFalse(self._manager._check_server_availability.called)
self.assertFalse(result)
def test_ensure_server_exception(self):
server_details = {'instance_id': 'fake_inst_id', 'ip': '1.2.3.4'}
self.mock_object(self._manager, '_check_server_availability',
mock.Mock(return_value=True))
self.mock_object(self._manager.compute_api, 'server_get',
mock.Mock(side_effect=exception.ManilaException))
self.assertRaises(exception.ManilaException,
self._manager.ensure_service_instance,
self._manager.admin_context,
server_details)
self._manager.compute_api.server_get.assert_called_once_with(
self._manager.admin_context, server_details['instance_id'])
self.assertFalse(self._manager._check_server_availability.called)
def test_ensure_server_non_active(self):
server_details = {'instance_id': 'fake_inst_id', 'ip': '1.2.3.4'}
fake_server = fake_compute.FakeServer(status='ERROR')
self.mock_object(self._manager.compute_api, 'server_get',
mock.Mock(return_value=fake_server))
self.mock_object(self._manager, '_check_server_availability',
mock.Mock(return_value=True))
result = self._manager.ensure_service_instance(
self._manager.admin_context, server_details)
self.assertFalse(self._manager._check_server_availability.called)
self.assertFalse(result)
def test_ensure_server_no_instance_id(self):
# Tests that we avoid a KeyError if the share details don't have an
# instance_id key set (so we can't find the share instance).
self.assertFalse(self._manager.ensure_service_instance(
self._manager.admin_context, {'ip': '1.2.3.4'}))
def test_get_key_create_new(self):
keypair_name = self._manager.get_config_option(
'manila_service_keypair_name')
fake_keypair = fake_compute.FakeKeypair(name=keypair_name)
self.mock_object(self._manager.compute_api, 'keypair_list',
mock.Mock(return_value=[]))
self.mock_object(self._manager.compute_api, 'keypair_import',
mock.Mock(return_value=fake_keypair))
result = self._manager._get_key(self._manager.admin_context)
self.assertEqual(
(fake_keypair.name,
os.path.expanduser(self._manager.get_config_option(
'path_to_private_key'))),
result)
self._manager.compute_api.keypair_list.assert_called_once_with(
self._manager.admin_context)
self._manager.compute_api.keypair_import.assert_called_once_with(
self._manager.admin_context, keypair_name, '')
def test_get_key_exists(self):
fake_keypair = fake_compute.FakeKeypair(
name=self._manager.get_config_option(
'manila_service_keypair_name'),
public_key='fake_public_key')
self.mock_object(self._manager.compute_api, 'keypair_list',
mock.Mock(return_value=[fake_keypair]))
self.mock_object(self._manager.compute_api, 'keypair_import',
mock.Mock(return_value=fake_keypair))
self.mock_object(self._manager, '_execute',
mock.Mock(return_value=('fake_public_key', '')))
result = self._manager._get_key(self._manager.admin_context)
self._manager.compute_api.keypair_list.assert_called_once_with(
self._manager.admin_context)
self.assertFalse(self._manager.compute_api.keypair_import.called)
self.assertEqual(
(fake_keypair.name,
os.path.expanduser(self._manager.get_config_option(
'path_to_private_key'))),
result)
def test_get_key_exists_recreate(self):
fake_keypair = fake_compute.FakeKeypair(
name=self._manager.get_config_option(
'manila_service_keypair_name'),
public_key='fake_public_key1')
self.mock_object(self._manager.compute_api, 'keypair_list',
mock.Mock(return_value=[fake_keypair]))
self.mock_object(self._manager.compute_api, 'keypair_import',
mock.Mock(return_value=fake_keypair))
self.mock_object(self._manager.compute_api, 'keypair_delete')
self.mock_object(self._manager, '_execute',
mock.Mock(return_value=('fake_public_key2', '')))
result = self._manager._get_key(self._manager.admin_context)
self._manager.compute_api.keypair_list.assert_called_once_with(
self._manager.admin_context)
self._manager.compute_api.keypair_delete.assert_called_once_with(
self._manager.admin_context, fake_keypair.id)
self._manager.compute_api.keypair_import.assert_called_once_with(
self._manager.admin_context, fake_keypair.name, 'fake_public_key2')
self.assertEqual(
(fake_keypair.name,
os.path.expanduser(self._manager.get_config_option(
'path_to_private_key'))),
result)
def test_get_key_more_than_one_exist(self):
fake_keypair = fake_compute.FakeKeypair(
name=self._manager.get_config_option(
'manila_service_keypair_name'),
public_key='fake_public_key1')
self.mock_object(self._manager.compute_api, 'keypair_list',
mock.Mock(return_value=[fake_keypair, fake_keypair]))
self.assertRaises(
exception.ServiceInstanceException,
self._manager._get_key, self._manager.admin_context)
self._manager.compute_api.keypair_list.assert_called_once_with(
self._manager.admin_context)
def test_get_key_keypath_to_public_not_set(self):
self._manager.path_to_public_key = None
result = self._manager._get_key(self._manager.admin_context)
self.assertEqual((None, None), result)
def test_get_key_keypath_to_private_not_set(self):
self._manager.path_to_private_key = None
result = self._manager._get_key(self._manager.admin_context)
self.assertEqual((None, None), result)
def test_get_key_incorrect_keypath_to_public(self):
def exists_side_effect(path):
return False if path == 'fake_path' else True
self._manager.path_to_public_key = 'fake_path'
os_path_exists_mock = mock.Mock(side_effect=exists_side_effect)
with mock.patch.object(os.path, 'exists', os_path_exists_mock):
with mock.patch.object(os.path, 'expanduser',
mock.Mock(side_effect=lambda value: value)):
result = self._manager._get_key(self._manager.admin_context)
self.assertEqual((None, None), result)
def test_get_key_incorrect_keypath_to_private(self):
def exists_side_effect(path):
return False if path == 'fake_path' else True
self._manager.path_to_private_key = 'fake_path'
os_path_exists_mock = mock.Mock(side_effect=exists_side_effect)
with mock.patch.object(os.path, 'exists', os_path_exists_mock):
with mock.patch.object(os.path, 'expanduser',
mock.Mock(side_effect=lambda value: value)):
result = self._manager._get_key(self._manager.admin_context)
self.assertEqual((None, None), result)
def test_get_service_image(self):
fake_image1 = fake_compute.FakeImage(
name=self._manager.get_config_option('service_image_name'),
status='active')
fake_image2 = fake_compute.FakeImage(
name='service_image_name',
status='error')
fake_image3 = fake_compute.FakeImage(
name='another-image',
status='active')
self.mock_object(self._manager.image_api, 'image_list',
mock.Mock(return_value=[fake_image1,
fake_image2,
fake_image3]))
result = self._manager._get_service_image(self._manager.admin_context)
self.assertEqual(fake_image1.id, result)
def test_get_service_image_not_found(self):
self.mock_object(self._manager.image_api, 'image_list',
mock.Mock(return_value=[]))
self.assertRaises(
exception.ServiceInstanceException,
self._manager._get_service_image, self._manager.admin_context)
def test_get_service_image_not_active(self):
fake_error_image = fake_compute.FakeImage(
name='service_image_name',
status='error')
self.mock_object(self._manager.image_api, 'image_list',
mock.Mock(return_value=[fake_error_image]))
self.assertRaises(
exception.ServiceInstanceException,
self._manager._get_service_image, self._manager.admin_context)
def test_get_service_image_ambiguous(self):
fake_image = fake_compute.FakeImage(
name=fake_get_config_option('service_image_name'),
status='active')
fake_images = [fake_image, fake_image]
self.mock_object(self._manager.image_api, 'image_list',
mock.Mock(return_value=fake_images))
self.assertRaises(
exception.ServiceInstanceException,
self._manager._get_service_image, self._manager.admin_context)
def test__delete_server_not_found(self):
self.mock_object(self._manager.compute_api, 'server_delete')
self.mock_object(
self._manager.compute_api, 'server_get',
mock.Mock(side_effect=exception.InstanceNotFound(
instance_id=self.instance_id)))
self._manager._delete_server(
self._manager.admin_context, self.instance_id)
self.assertFalse(self._manager.compute_api.server_delete.called)
self._manager.compute_api.server_get.assert_called_once_with(
self._manager.admin_context, self.instance_id)
def test__delete_server(self):
def fake_server_get(*args, **kwargs):
ctx = args[0]
if not hasattr(ctx, 'called'):
ctx.called = True
return
else:
raise exception.InstanceNotFound(instance_id=self.instance_id)
self.mock_object(self._manager.compute_api, 'server_delete')
self.mock_object(self._manager.compute_api, 'server_get',
mock.Mock(side_effect=fake_server_get))
self._manager._delete_server(
self._manager.admin_context, self.instance_id)
self._manager.compute_api.server_delete.assert_called_once_with(
self._manager.admin_context, self.instance_id)
self._manager.compute_api.server_get.assert_has_calls([
mock.call(self._manager.admin_context, self.instance_id),
mock.call(self._manager.admin_context, self.instance_id)])
def test__delete_server_found_always(self):
self.fake_time = 0
def fake_time():
return self.fake_time
def fake_sleep(time):
self.fake_time += 1
server_details = {'instance_id': 'fake_inst_id', 'status': 'ACTIVE'}
self.mock_object(self._manager.compute_api, 'server_delete')
self.mock_object(self._manager.compute_api, 'server_get',
mock.Mock(return_value=server_details))
self.mock_object(service_instance, 'time')
self.mock_object(
service_instance.time, 'time', mock.Mock(side_effect=fake_time))
self.mock_object(
service_instance.time, 'sleep', mock.Mock(side_effect=fake_sleep))
self.mock_object(self._manager, 'max_time_to_build_instance', 2)
self.assertRaises(
exception.ServiceInstanceException, self._manager._delete_server,
self._manager.admin_context, self.instance_id)
self._manager.compute_api.server_delete.assert_called_once_with(
self._manager.admin_context, self.instance_id)
service_instance.time.sleep.assert_has_calls(
[mock.call(mock.ANY) for i in range(2)])
service_instance.time.time.assert_has_calls(
[mock.call() for i in range(4)])
self._manager.compute_api.server_get.assert_has_calls(
[mock.call(self._manager.admin_context,
self.instance_id) for i in range(3)])
def test_delete_server_soft_deleted(self):
server_details = {'instance_id': 'fake_inst_id',
'status': 'SOFT_DELETED'}
self.mock_object(self._manager.compute_api, 'server_delete')
self.mock_object(self._manager.compute_api, 'server_get',
mock.Mock(return_value=server_details))
self._manager._delete_server(
self._manager.admin_context, self.instance_id)
self._manager.compute_api.server_delete.assert_called_once_with(
self._manager.admin_context, self.instance_id)
self._manager.compute_api.server_get.assert_has_calls([
mock.call(self._manager.admin_context, self.instance_id),
mock.call(self._manager.admin_context, self.instance_id)])
def test_delete_service_instance(self):
fake_server_details = dict(
router_id='foo', subnet_id='bar', instance_id='quuz')
self.mock_object(self._manager, '_delete_server')
self.mock_object(self._manager.network_helper, 'teardown_network')
self._manager.delete_service_instance(
self._manager.admin_context, fake_server_details)
self._manager._delete_server.assert_called_once_with(
self._manager.admin_context, fake_server_details['instance_id'])
self._manager.network_helper.teardown_network.assert_called_once_with(
fake_server_details)
@ddt.data(
*[{'service_config': service_config,
'tenant_config': tenant_config,
'server': server}
for service_config, tenant_config in (
('fake_net_s', 'fake_net_t'),
('fake_net_s', '12.34.56.78'),
('98.76.54.123', 'fake_net_t'),
('98.76.54.123', '12.34.56.78'))
for server in (
{'networks': {
'fake_net_s': ['foo', '98.76.54.123', 'bar'],
'fake_net_t': ['baar', '12.34.56.78', 'quuz']}},
{'addresses': {
'fake_net_s': [
{'addr': 'fake1'},
{'addr': '98.76.54.123'},
{'addr': 'fake2'}],
'fake_net_t': [
{'addr': 'fake3'},
{'addr': '12.34.56.78'},
{'addr': 'fake4'}],
}})])
@ddt.unpack
def test_get_common_server_valid_cases(self, service_config,
tenant_config, server):
self._get_common_server(service_config, tenant_config, server,
'98.76.54.123', '12.34.56.78', True)
@ddt.data(
*[{'service_config': service_config,
'tenant_config': tenant_config,
'server': server}
for service_config, tenant_config in (
('fake_net_s', 'fake'),
('fake', 'fake_net_t'),
('fake', 'fake'),
('98.76.54.123', '12.12.12.1212'),
('12.12.12.1212', '12.34.56.78'),
('12.12.12.1212', '12.12.12.1212'),
('1001::1001', '1001::100G'),
('1001::10G1', '1001::1001'),
)
for server in (
{'networks': {
'fake_net_s': ['foo', '98.76.54.123', 'bar'],
'fake_net_t': ['baar', '12.34.56.78', 'quuz']}},
{'addresses': {
'fake_net_s': [
{'addr': 'fake1'},
{'addr': '98.76.54.123'},
{'addr': 'fake2'}],
'fake_net_t': [
{'addr': 'fake3'},
{'addr': '12.34.56.78'},
{'addr': 'fake4'}],
}})])
@ddt.unpack
def test_get_common_server_invalid_cases(self, service_config,
tenant_config, server):
self._get_common_server(service_config, tenant_config, server,
'98.76.54.123', '12.34.56.78', False)
@ddt.data(
*[{'service_config': service_config,
'tenant_config': tenant_config,
'server': server}
for service_config, tenant_config in (
('fake_net_s', '1001::1002'),
('1001::1001', 'fake_net_t'),
('1001::1001', '1001::1002'))
for server in (
{'networks': {
'fake_net_s': ['foo', '1001::1001'],
'fake_net_t': ['bar', '1001::1002']}},
{'addresses': {
'fake_net_s': [{'addr': 'foo'}, {'addr': '1001::1001'}],
'fake_net_t': [{'addr': 'bar'}, {'addr': '1001::1002'}]}})])
@ddt.unpack
def test_get_common_server_valid_ipv6_address(self, service_config,
tenant_config, server):
self._get_common_server(service_config, tenant_config, server,
'1001::1001', '1001::1002', True)
def _get_common_server(self, service_config, tenant_config,
server, service_address, network_address,
is_valid=True):
fake_instance_id = 'fake_instance_id'
fake_user = 'fake_user'
fake_pass = 'fake_pass'
fake_server = {'id': fake_instance_id}
fake_server.update(server)
expected = {
'backend_details': {
'username': fake_user,
'password': fake_pass,
'pk_path': self._manager.path_to_private_key,
'ip': service_address,
'public_address': network_address,
'instance_id': fake_instance_id,
}
}
def fake_get_config_option(attr):
if attr == 'service_net_name_or_ip':
return service_config
elif attr == 'tenant_net_name_or_ip':
return tenant_config
elif attr == 'service_instance_name_or_id':
return fake_instance_id
elif attr == 'service_instance_user':
return fake_user
elif attr == 'service_instance_password':
return fake_pass
else:
raise exception.ManilaException("Wrong test data provided.")
self.mock_object(
self._manager.compute_api, 'server_get_by_name_or_id',
mock.Mock(return_value=fake_server))
self.mock_object(
self._manager, 'get_config_option',
mock.Mock(side_effect=fake_get_config_option))
if is_valid:
actual = self._manager.get_common_server()
self.assertEqual(expected, actual)
else:
self.assertRaises(
exception.ManilaException,
self._manager.get_common_server)
self.assertTrue(
self._manager.compute_api.server_get_by_name_or_id.called)
def test___create_service_instance_with_sg_success(self):
self.mock_object(service_instance, 'NeutronNetworkHelper',
mock.Mock(side_effect=FakeNetworkHelper))
config_data = dict(DEFAULT=dict(
driver_handles_share_servers=True,
service_instance_user='fake_user',
limit_ssh_access=True))
with test_utils.create_temp_config_with_opts(config_data):
self._manager = service_instance.ServiceInstanceManager()
server_create = dict(id='fakeid', status='CREATING', networks=dict())
net_name = self._manager.get_config_option("service_network_name")
sg = [{'id': 'fakeid', 'name': 'fakename'}, ]
ip_address = 'fake_ip_address'
service_image_id = 'fake_service_image_id'
key_data = 'fake_key_name', 'fake_key_path'
instance_name = 'fake_instance_name'
network_info = dict()
network_data = {'nics': ['fake_nic1', 'fake_nic2']}
network_data['router'] = dict(id='fake_router_id')
server_get = dict(
id='fakeid', status='ACTIVE', networks={net_name: [ip_address]})
network_data.update(dict(
router_id='fake_router_id', subnet_id='fake_subnet_id',
public_port=dict(id='fake_public_port',
fixed_ips=[dict(ip_address=ip_address)]),
service_port=dict(id='fake_service_port',
fixed_ips=[{'ip_address': ip_address}]),
admin_port={'id': 'fake_admin_port',
'fixed_ips': [{'ip_address': ip_address}]},
service_subnet={'id': 'fake_subnet_id',
'cidr': '10.254.0.0/28'})
)
self.mock_object(service_instance.time, 'time',
mock.Mock(return_value=5))
self.mock_object(self._manager.network_helper, 'setup_network',
mock.Mock(return_value=network_data))
self.mock_object(self._manager.network_helper, 'get_network_name',
mock.Mock(return_value=net_name))
self.mock_object(self._manager, '_get_service_image',
mock.Mock(return_value=service_image_id))
self.mock_object(self._manager, '_get_key',
mock.Mock(return_value=key_data))
self.mock_object(self._manager, '_get_or_create_security_groups',
mock.Mock(return_value=sg))
self.mock_object(self._manager.compute_api, 'server_create',
mock.Mock(return_value=server_create))
self.mock_object(self._manager.compute_api, 'server_get',
mock.Mock(return_value=server_get))
self.mock_object(self._manager.compute_api,
'add_security_group_to_server')
expected = {
'id': server_get['id'],
'status': server_get['status'],
'pk_path': key_data[1],
'public_address': ip_address,
'router_id': network_data.get('router_id'),
'subnet_id': network_data.get('subnet_id'),
'instance_id': server_get['id'],
'ip': ip_address,
'networks': server_get['networks'],
'public_port_id': 'fake_public_port',
'service_port_id': 'fake_service_port',
'admin_port_id': 'fake_admin_port',
'admin_ip': 'fake_ip_address',
}
result = self._manager._create_service_instance(
self._manager.admin_context, instance_name, network_info)
self.assertEqual(expected, result)
self.assertTrue(service_instance.time.time.called)
self._manager.network_helper.setup_network.assert_called_once_with(
network_info)
self._manager._get_service_image.assert_called_once_with(
self._manager.admin_context)
self._manager._get_key.assert_called_once_with(
self._manager.admin_context)
self._manager._get_or_create_security_groups.assert_called_once_with(
self._manager.admin_context,
allow_ssh_subnet=network_data['service_subnet'])
self._manager.compute_api.server_create.assert_called_once_with(
self._manager.admin_context, name=instance_name,
image=service_image_id, flavor='100',
key_name=key_data[0], nics=network_data['nics'],
availability_zone=service_instance.CONF.storage_availability_zone)
self._manager.compute_api.server_get.assert_called_once_with(
self._manager.admin_context, server_create['id'])
(self._manager.compute_api.add_security_group_to_server.
assert_called_once_with(
self._manager.admin_context,
server_get['id'],
sg[0]['id']))
self._manager.network_helper.get_network_name.assert_has_calls([])
def test___create_service_instance_neutron_no_admin_ip(self):
self.mock_object(service_instance, 'NeutronNetworkHelper',
mock.Mock(side_effect=FakeNetworkHelper))
config_data = {'DEFAULT': {
'driver_handles_share_servers': True,
'service_instance_user': 'fake_user',
'limit_ssh_access': True}}
with test_utils.create_temp_config_with_opts(config_data):
self._manager = service_instance.ServiceInstanceManager()
server_create = {'id': 'fakeid', 'status': 'CREATING', 'networks': {}}
net_name = self._manager.get_config_option("service_network_name")
sg = {'id': 'fakeid', 'name': 'fakename'}
ip_address = 'fake_ip_address'
service_image_id = 'fake_service_image_id'
key_data = 'fake_key_name', 'fake_key_path'
instance_name = 'fake_instance_name'
network_info = {}
network_data = {
'nics': ['fake_nic1', 'fake_nic2'],
'router_id': 'fake_router_id', 'subnet_id': 'fake_subnet_id',
'public_port': {'id': 'fake_public_port',
'fixed_ips': [{'ip_address': ip_address}]},
'service_port': {'id': 'fake_service_port',
'fixed_ips': [{'ip_address': ip_address}]},
'admin_port': {'id': 'fake_admin_port',
'fixed_ips': []},
'router': {'id': 'fake_router_id'},
'service_subnet': {'id': 'fake_id',
'cidr': '10.254.0.0/28'}
}
server_get = {
'id': 'fakeid', 'status': 'ACTIVE', 'networks':
{net_name: [ip_address]}}
self.mock_object(service_instance.time, 'time',
mock.Mock(return_value=5))
self.mock_object(self._manager.network_helper, 'setup_network',
mock.Mock(return_value=network_data))
self.mock_object(self._manager.network_helper, 'get_network_name',
mock.Mock(return_value=net_name))
self.mock_object(self._manager, '_get_service_image',
mock.Mock(return_value=service_image_id))
self.mock_object(self._manager, '_get_key',
mock.Mock(return_value=key_data))
self.mock_object(self._manager, '_get_or_create_security_groups',
mock.Mock(return_value=[sg, ]))
self.mock_object(self._manager.compute_api, 'server_create',
mock.Mock(return_value=server_create))
self.mock_object(self._manager.compute_api, 'server_get',
mock.Mock(return_value=server_get))
self.mock_object(self._manager.compute_api,
'add_security_group_to_server')
self.assertRaises(
exception.AdminIPNotFound, self._manager._create_service_instance,
self._manager.admin_context, instance_name, network_info)
self.assertTrue(service_instance.time.time.called)
self._manager.network_helper.setup_network.assert_called_once_with(
network_info)
self._manager._get_service_image.assert_called_once_with(
self._manager.admin_context)
self._manager._get_key.assert_called_once_with(
self._manager.admin_context)
self._manager._get_or_create_security_groups.assert_called_once_with(
self._manager.admin_context,
allow_ssh_subnet=network_data['service_subnet'])
self._manager.compute_api.server_create.assert_called_once_with(
self._manager.admin_context, name=instance_name,
image=service_image_id, flavor='100',
key_name=key_data[0], nics=network_data['nics'],
availability_zone=service_instance.CONF.storage_availability_zone)
self._manager.compute_api.server_get.assert_called_once_with(
self._manager.admin_context, server_create['id'])
(self._manager.compute_api.add_security_group_to_server.
assert_called_once_with(
self._manager.admin_context, server_get['id'], sg['id']))
self._manager.network_helper.get_network_name.assert_has_calls([])
@ddt.data(
dict(
instance_id_included=False,
mockobj=mock.Mock(side_effect=exception.ServiceInstanceException)),
dict(
instance_id_included=True,
mockobj=mock.Mock(return_value=dict(id='fakeid', status='ERROR'))))
@ddt.unpack
def test___create_service_instance_failed_to_create(
self, instance_id_included, mockobj):
service_image_id = 'fake_service_image_id'
key_data = 'fake_key_name', 'fake_key_path'
instance_name = 'fake_instance_name'
network_info = dict()
network_data = dict(
nics=['fake_nic1', 'fake_nic2'],
router_id='fake_router_id', subnet_id='fake_subnet_id')
self.mock_object(self._manager.network_helper, 'setup_network',
mock.Mock(return_value=network_data))
self.mock_object(self._manager, '_get_service_image',
mock.Mock(return_value=service_image_id))
self.mock_object(self._manager, '_get_key',
mock.Mock(return_value=key_data))
self.mock_object(
self._manager.compute_api, 'server_create', mockobj)
self.mock_object(
self._manager, 'wait_for_instance_to_be_active',
mock.Mock(side_effect=exception.ServiceInstanceException))
try:
self._manager._create_service_instance(
self._manager.admin_context, instance_name, network_info)
except exception.ServiceInstanceException as e:
expected = dict(server_details=dict(
subnet_id=network_data['subnet_id'],
router_id=network_data['router_id']))
if instance_id_included:
expected['server_details']['instance_id'] = 'fakeid'
self.assertEqual(expected, e.detail_data)
else:
raise exception.ManilaException('Expected error was not raised.')
self._manager.network_helper.setup_network.assert_called_once_with(
network_info)
self._manager._get_service_image.assert_called_once_with(
self._manager.admin_context)
self._manager._get_key.assert_called_once_with(
self._manager.admin_context)
self._manager.compute_api.server_create.assert_called_once_with(
self._manager.admin_context, name=instance_name,
image=service_image_id, flavor='100',
key_name=key_data[0], nics=network_data['nics'],
availability_zone=service_instance.CONF.storage_availability_zone)
def test___create_service_instance_limit_ssh_no_service_subnet(self):
self.mock_object(service_instance, 'NeutronNetworkHelper',
mock.Mock(side_effect=FakeNetworkHelper))
config_data = dict(DEFAULT=dict(
driver_handles_share_servers=True,
service_instance_user='fake_user',
limit_ssh_access=True))
with test_utils.create_temp_config_with_opts(config_data):
self._manager = service_instance.ServiceInstanceManager()
server_create = dict(id='fakeid', status='CREATING', networks=dict())
net_name = self._manager.get_config_option("service_network_name")
ip_address = 'fake_ip_address'
service_image_id = 'fake_service_image_id'
key_data = 'fake_key_name', 'fake_key_path'
instance_name = 'fake_instance_name'
network_info = dict()
network_data = {'nics': ['fake_nic1', 'fake_nic2']}
network_data['router'] = dict(id='fake_router_id')
server_get = dict(
id='fakeid', status='ACTIVE', networks={net_name: [ip_address]})
network_data.update(dict(
router_id='fake_router_id', subnet_id='fake_subnet_id',
public_port=dict(id='fake_public_port',
fixed_ips=[dict(ip_address=ip_address)]),
service_port=dict(id='fake_service_port',
fixed_ips=[{'ip_address': ip_address}]),
admin_port={'id': 'fake_admin_port',
'fixed_ips': [{'ip_address': ip_address}]},)
)
self.mock_object(service_instance.time, 'time',
mock.Mock(return_value=5))
self.mock_object(self._manager.network_helper, 'setup_network',
mock.Mock(return_value=network_data))
self.mock_object(self._manager.network_helper, 'get_network_name',
mock.Mock(return_value=net_name))
self.mock_object(self._manager, '_get_service_image',
mock.Mock(return_value=service_image_id))
self.mock_object(self._manager, '_get_key',
mock.Mock(return_value=key_data))
self.mock_object(self._manager.compute_api, 'server_create',
mock.Mock(return_value=server_create))
self.mock_object(self._manager.compute_api, 'server_get',
mock.Mock(return_value=server_get))
self.assertRaises(exception.ManilaException,
self._manager._create_service_instance,
self._manager.admin_context, instance_name,
network_info)
def test___create_service_instance_failed_to_build(self):
server_create = dict(id='fakeid', status='CREATING', networks=dict())
service_image_id = 'fake_service_image_id'
key_data = 'fake_key_name', 'fake_key_path'
instance_name = 'fake_instance_name'
network_info = dict()
network_data = dict(
nics=['fake_nic1', 'fake_nic2'],
router_id='fake_router_id', subnet_id='fake_subnet_id')
self.mock_object(self._manager.network_helper, 'setup_network',
mock.Mock(return_value=network_data))
self.mock_object(self._manager, '_get_service_image',
mock.Mock(return_value=service_image_id))
self.mock_object(self._manager, '_get_key',
mock.Mock(return_value=key_data))
self.mock_object(self._manager.compute_api, 'server_create',
mock.Mock(return_value=server_create))
self.mock_object(
self._manager, 'wait_for_instance_to_be_active',
mock.Mock(side_effect=exception.ServiceInstanceException))
try:
self._manager._create_service_instance(
self._manager.admin_context, instance_name, network_info)
except exception.ServiceInstanceException as e:
self.assertEqual(
dict(server_details=dict(subnet_id=network_data['subnet_id'],
router_id=network_data['router_id'],
instance_id=server_create['id'])),
e.detail_data)
else:
raise exception.ManilaException('Expected error was not raised.')
self._manager.network_helper.setup_network.assert_called_once_with(
network_info)
self._manager._get_service_image.assert_called_once_with(
self._manager.admin_context)
self._manager._get_key.assert_called_once_with(
self._manager.admin_context)
self._manager.compute_api.server_create.assert_called_once_with(
self._manager.admin_context, name=instance_name,
image=service_image_id, flavor='100',
key_name=key_data[0], nics=network_data['nics'],
availability_zone=service_instance.CONF.storage_availability_zone)
@ddt.data(
dict(name=None, path=None),
dict(name=None, path='/tmp'))
@ddt.unpack
def test__create_service_instance_no_key_and_no_path(self, name, path):
key_data = name, path
self.mock_object(self._manager, '_get_service_image')
self.mock_object(self._manager, '_get_key',
mock.Mock(return_value=key_data))
self.assertRaises(
exception.ServiceInstanceException,
self._manager._create_service_instance,
self._manager.admin_context, 'fake_instance_name', dict())
self._manager._get_service_image.assert_called_once_with(
self._manager.admin_context)
self._manager._get_key.assert_called_once_with(
self._manager.admin_context)
@mock.patch('time.sleep')
@mock.patch('time.time')
def _test_wait_for_instance(self, mock_time, mock_sleep,
server_get_side_eff=None,
expected_try_count=1,
expected_sleep_count=0,
expected_ret_val=None,
expected_exc=None):
mock_server_get = mock.Mock(side_effect=server_get_side_eff)
self.mock_object(self._manager.compute_api, 'server_get',
mock_server_get)
self.fake_time = 0
def fake_time():
return self.fake_time
def fake_sleep(sleep_time):
self.fake_time += sleep_time
# Note(lpetrut): LOG methods can call time.time
mock_time.side_effect = fake_time
mock_sleep.side_effect = fake_sleep
timeout = 3
if expected_exc:
self.assertRaises(
expected_exc,
self._manager.wait_for_instance_to_be_active,
instance_id=mock.sentinel.instance_id,
timeout=timeout)
else:
instance = self._manager.wait_for_instance_to_be_active(
instance_id=mock.sentinel.instance_id,
timeout=timeout)
self.assertEqual(expected_ret_val, instance)
mock_server_get.assert_has_calls(
[mock.call(self._manager.admin_context,
mock.sentinel.instance_id)] * expected_try_count)
mock_sleep.assert_has_calls([mock.call(1)] * expected_sleep_count)
def test_wait_for_instance_timeout(self):
server_get_side_eff = [
exception.InstanceNotFound(
instance_id=mock.sentinel.instance_id),
{'status': 'BUILDING'},
{'status': 'ACTIVE'}]
# Note that in this case, although the status is active, the
# 'networks' field is missing.
self._test_wait_for_instance( # pylint: disable=no-value-for-parameter
server_get_side_eff=server_get_side_eff,
expected_exc=exception.ServiceInstanceException,
expected_try_count=3,
expected_sleep_count=3)
def test_wait_for_instance_error_state(self):
mock_instance = {'status': 'ERROR'}
self._test_wait_for_instance( # pylint: disable=no-value-for-parameter
server_get_side_eff=[mock_instance],
expected_exc=exception.ServiceInstanceException,
expected_try_count=1)
def test_wait_for_instance_available(self):
mock_instance = {'status': 'ACTIVE',
'networks': mock.sentinel.networks}
self._test_wait_for_instance( # pylint: disable=no-value-for-parameter
server_get_side_eff=[mock_instance],
expected_try_count=1,
expected_ret_val=mock_instance)
def test_reboot_server(self):
fake_server = {'instance_id': mock.sentinel.instance_id}
soft_reboot = True
mock_reboot = mock.Mock()
self.mock_object(self._manager.compute_api, 'server_reboot',
mock_reboot)
self._manager.reboot_server(fake_server, soft_reboot)
mock_reboot.assert_called_once_with(self._manager.admin_context,
fake_server['instance_id'],
soft_reboot)
class BaseNetworkHelperTestCase(test.TestCase):
"""Tests Base network helper for service instance."""
def test_instantiate_valid(self):
class FakeNetworkHelper(service_instance.BaseNetworkhelper):
@property
def NAME(self):
return 'fake_NAME'
def __init__(self, service_instance_manager):
self.fake_init = 'fake_init_value'
def get_network_name(self, network_info):
return 'fake_network_name'
def setup_connectivity_with_service_instances(self):
return 'fake_setup_connectivity_with_service_instances'
def setup_network(self, network_info):
return 'fake_setup_network'
def teardown_network(self, server_details):
return 'fake_teardown_network'
instance = FakeNetworkHelper('fake')
attrs = [
'fake_init', 'NAME', 'get_network_name', 'teardown_network',
'setup_connectivity_with_service_instances', 'setup_network',
]
for attr in attrs:
self.assertTrue(hasattr(instance, attr))
self.assertEqual('fake_init_value', instance.fake_init)
self.assertEqual('fake_NAME', instance.NAME)
self.assertEqual(
'fake_network_name', instance.get_network_name('fake'))
self.assertEqual(
'fake_setup_connectivity_with_service_instances',
instance.setup_connectivity_with_service_instances())
self.assertEqual('fake_setup_network', instance.setup_network('fake'))
self.assertEqual(
'fake_teardown_network', instance.teardown_network('fake'))
def test_instantiate_invalid(self):
self.assertRaises(
TypeError, service_instance.BaseNetworkhelper, 'fake')
@ddt.ddt
class NeutronNetworkHelperTestCase(test.TestCase):
"""Tests Neutron network helper for service instance."""
def setUp(self):
super(NeutronNetworkHelperTestCase, self).setUp()
self.mock_object(importutils, 'import_class')
self.fake_manager = FakeServiceInstance()
def _init_neutron_network_plugin(self):
self.mock_object(
service_instance.NeutronNetworkHelper, '_get_service_network_id',
mock.Mock(return_value='fake_service_network_id'))
return service_instance.NeutronNetworkHelper(self.fake_manager)
def test_init_neutron_network_plugin(self):
instance = self._init_neutron_network_plugin()
self.assertEqual(service_instance.NEUTRON_NAME, instance.NAME)
attrs = [
'neutron_api', 'vif_driver', 'service_network_id',
'connect_share_server_to_tenant_network', 'get_config_option']
for attr in attrs:
self.assertTrue(hasattr(instance, attr), "No attr '%s'" % attr)
(service_instance.NeutronNetworkHelper._get_service_network_id.
assert_called_once_with())
self.assertEqual('DEFAULT', instance.neutron_api.config_group_name)
def test_init_neutron_network_plugin_with_driver_config_group(self):
self.fake_manager.driver_config = mock.Mock()
self.fake_manager.driver_config.config_group = (
'fake_config_group')
self.fake_manager.driver_config.network_config_group = None
instance = self._init_neutron_network_plugin()
self.assertEqual('fake_config_group',
instance.neutron_api.config_group_name)
def test_init_neutron_network_plugin_with_network_config_group(self):
self.fake_manager.driver_config = mock.Mock()
self.fake_manager.driver_config.config_group = (
"fake_config_group")
self.fake_manager.driver_config.network_config_group = (
"fake_network_config_group")
instance = self._init_neutron_network_plugin()
self.assertEqual('fake_network_config_group',
instance.neutron_api.config_group_name)
def test_admin_project_id(self):
instance = self._init_neutron_network_plugin()
admin_project_id = 'fake_admin_project_id'
self.mock_class('manila.network.neutron.api.API', mock.Mock())
instance.neutron_api.admin_project_id = admin_project_id
self.assertEqual(admin_project_id, instance.admin_project_id)
def test_get_network_name(self):
network_info = dict(neutron_net_id='fake_neutron_net_id')
network = dict(name='fake_network_name')
instance = self._init_neutron_network_plugin()
self.mock_object(
instance.neutron_api, 'get_network',
mock.Mock(return_value=network))
result = instance.get_network_name(network_info)
self.assertEqual(network['name'], result)
instance.neutron_api.get_network.assert_called_once_with(
network_info['neutron_net_id'])
def test_get_service_network_id_none_exist(self):
service_network_name = fake_get_config_option('service_network_name')
network = dict(id='fake_network_id')
admin_project_id = 'fake_admin_project_id'
self.mock_object(
service_instance.neutron.API, 'get_all_admin_project_networks',
mock.Mock(return_value=[]))
self.mock_object(
service_instance.neutron.API, 'admin_project_id',
mock.Mock(return_value=admin_project_id))
self.mock_object(
service_instance.neutron.API, 'network_create',
mock.Mock(return_value=network))
instance = service_instance.NeutronNetworkHelper(self.fake_manager)
result = instance._get_service_network_id()
self.assertEqual(network['id'], result)
self.assertTrue(service_instance.neutron.API.
get_all_admin_project_networks.called)
service_instance.neutron.API.network_create.assert_has_calls([
mock.call(instance.admin_project_id, service_network_name)])
def test_get_service_network_id_one_exist(self):
service_network_name = fake_get_config_option('service_network_name')
network = dict(id='fake_network_id', name=service_network_name)
admin_project_id = 'fake_admin_project_id'
self.mock_object(
service_instance.neutron.API, 'get_all_admin_project_networks',
mock.Mock(return_value=[network]))
self.mock_object(
service_instance.neutron.API, 'admin_project_id',
mock.Mock(return_value=admin_project_id))
instance = service_instance.NeutronNetworkHelper(self.fake_manager)
result = instance._get_service_network_id()
self.assertEqual(network['id'], result)
self.assertTrue(service_instance.neutron.API.
get_all_admin_project_networks.called)
def test_get_service_network_id_two_exist(self):
service_network_name = fake_get_config_option('service_network_name')
network = dict(id='fake_network_id', name=service_network_name)
self.mock_object(
service_instance.neutron.API, 'get_all_admin_project_networks',
mock.Mock(return_value=[network, network]))
helper = service_instance.NeutronNetworkHelper(self.fake_manager)
self.assertRaises(exception.ManilaException,
lambda: helper.service_network_id)
(service_instance.neutron.API.get_all_admin_project_networks.
assert_has_calls([mock.call()]))
@ddt.data(dict(), dict(subnet_id='foo'), dict(router_id='bar'))
def test_teardown_network_no_service_data(self, server_details):
instance = self._init_neutron_network_plugin()
self.mock_object(
service_instance.neutron.API, 'router_remove_interface')
instance.teardown_network(server_details)
self.assertFalse(
service_instance.neutron.API.router_remove_interface.called)
@ddt.data(
*[dict(server_details=sd, fail=f) for f in (True, False)
for sd in (dict(service_port_id='fake_service_port_id'),
dict(public_port_id='fake_public_port_id'),
dict(service_port_id='fake_service_port_id',
public_port_id='fake_public_port_id'))]
)
@ddt.unpack
def test_teardown_network_with_ports(self, server_details, fail):
instance = self._init_neutron_network_plugin()
self.mock_object(
service_instance.neutron.API, 'router_remove_interface')
if fail:
delete_port_mock = mock.Mock(
side_effect=exception.NetworkException(code=404))
else:
delete_port_mock = mock.Mock()
self.mock_object(instance.neutron_api, 'delete_port', delete_port_mock)
self.mock_object(service_instance.LOG, 'debug')
instance.teardown_network(server_details)
self.assertFalse(instance.neutron_api.router_remove_interface.called)
self.assertEqual(
len(server_details),
len(instance.neutron_api.delete_port.mock_calls))
for k, v in server_details.items():
self.assertIn(
mock.call(v), instance.neutron_api.delete_port.mock_calls)
if fail:
service_instance.LOG.debug.assert_has_calls([
mock.call(mock.ANY, mock.ANY) for sd in server_details
])
else:
service_instance.LOG.debug.assert_has_calls([])
@ddt.data(
dict(service_port_id='fake_service_port_id'),
dict(public_port_id='fake_public_port_id'),
dict(service_port_id='fake_service_port_id',
public_port_id='fake_public_port_id'),
)
def test_teardown_network_with_ports_unhandled_exception(self,
server_details):
instance = self._init_neutron_network_plugin()
self.mock_object(
service_instance.neutron.API, 'router_remove_interface')
delete_port_mock = mock.Mock(
side_effect=exception.NetworkException(code=500))
self.mock_object(
service_instance.neutron.API, 'delete_port', delete_port_mock)
self.mock_object(service_instance.LOG, 'debug')
self.assertRaises(
exception.NetworkException,
instance.teardown_network,
server_details,
)
self.assertFalse(
service_instance.neutron.API.router_remove_interface.called)
service_instance.neutron.API.delete_port.assert_called_once_with(
mock.ANY)
service_instance.LOG.debug.assert_has_calls([])
def test_teardown_network_with_wrong_ports(self):
instance = self._init_neutron_network_plugin()
self.mock_object(
service_instance.neutron.API, 'router_remove_interface')
self.mock_object(
service_instance.neutron.API, 'delete_port')
self.mock_object(service_instance.LOG, 'debug')
instance.teardown_network(dict(foo_id='fake_service_port_id'))
service_instance.neutron.API.router_remove_interface.assert_has_calls(
[])
service_instance.neutron.API.delete_port.assert_has_calls([])
service_instance.LOG.debug.assert_has_calls([])
def test_teardown_network_subnet_is_used(self):
server_details = dict(subnet_id='foo', router_id='bar')
fake_ports = [
{'device_id': 'fake_device_id',
'device_owner': 'compute:foo'},
]
instance = self._init_neutron_network_plugin()
self.mock_object(
service_instance.neutron.API, 'router_remove_interface')
self.mock_object(
service_instance.neutron.API, 'update_subnet')
self.mock_object(
service_instance.neutron.API, 'list_ports',
mock.Mock(return_value=fake_ports))
instance.teardown_network(server_details)
self.assertFalse(
service_instance.neutron.API.router_remove_interface.called)
self.assertFalse(service_instance.neutron.API.update_subnet.called)
service_instance.neutron.API.list_ports.assert_called_once_with(
fields=['device_id', 'device_owner'], fixed_ips=['subnet_id=foo'])
def test_teardown_network_subnet_not_used(self):
server_details = dict(subnet_id='foo', router_id='bar')
fake_ports = [
{'device_id': 'fake_device_id',
'device_owner': 'network:router_interface'},
{'device_id': 'fake_device_id',
'device_owner': 'compute'},
{'device_id': '',
'device_owner': 'compute'},
]
instance = self._init_neutron_network_plugin()
self.mock_object(
service_instance.neutron.API, 'router_remove_interface')
self.mock_object(
service_instance.neutron.API, 'update_subnet')
self.mock_object(
service_instance.neutron.API, 'list_ports',
mock.Mock(return_value=fake_ports))
instance.teardown_network(server_details)
(service_instance.neutron.API.router_remove_interface.
assert_called_once_with('bar', 'foo'))
(service_instance.neutron.API.update_subnet.
assert_called_once_with('foo', ''))
service_instance.neutron.API.list_ports.assert_called_once_with(
fields=['device_id', 'device_owner'], fixed_ips=['subnet_id=foo'])
def test_teardown_network_subnet_not_used_and_get_error_404(self):
server_details = dict(subnet_id='foo', router_id='bar')
fake_ports = [
{'device_id': 'fake_device_id',
'device_owner': 'fake'},
]
instance = self._init_neutron_network_plugin()
self.mock_object(
service_instance.neutron.API, 'router_remove_interface',
mock.Mock(side_effect=exception.NetworkException(code=404)))
self.mock_object(
service_instance.neutron.API, 'update_subnet')
self.mock_object(
service_instance.neutron.API, 'list_ports',
mock.Mock(return_value=fake_ports))
instance.teardown_network(server_details)
(service_instance.neutron.API.router_remove_interface.
assert_called_once_with('bar', 'foo'))
(service_instance.neutron.API.update_subnet.
assert_called_once_with('foo', ''))
service_instance.neutron.API.list_ports.assert_called_once_with(
fields=['device_id', 'device_owner'], fixed_ips=['subnet_id=foo'])
def test_teardown_network_subnet_not_used_get_unhandled_error(self):
server_details = dict(subnet_id='foo', router_id='bar')
fake_ports = [
{'device_id': 'fake_device_id',
'device_owner': 'fake'},
]
instance = self._init_neutron_network_plugin()
self.mock_object(
service_instance.neutron.API, 'router_remove_interface',
mock.Mock(side_effect=exception.NetworkException(code=500)))
self.mock_object(
service_instance.neutron.API, 'update_subnet')
self.mock_object(
service_instance.neutron.API, 'list_ports',
mock.Mock(return_value=fake_ports))
self.assertRaises(
exception.NetworkException,
instance.teardown_network, server_details)
(service_instance.neutron.API.router_remove_interface.
assert_called_once_with('bar', 'foo'))
self.assertFalse(service_instance.neutron.API.update_subnet.called)
service_instance.neutron.API.list_ports.assert_called_once_with(
fields=['device_id', 'device_owner'], fixed_ips=['subnet_id=foo'])
def test_setup_network_and_connect_share_server_to_tenant_net(self):
def fake_create_port(*aargs, **kwargs):
if aargs[1] == 'fake_service_network_id':
return self.service_port
elif aargs[1] == 'fake_tenant_network_id':
return self.public_port
else:
raise exception.ManilaException('Got unexpected data')
admin_project_id = 'fake_admin_project_id'
network_info = dict(
neutron_net_id='fake_tenant_network_id',
neutron_subnet_id='fake_tenant_subnet_id')
cidr = '13.0.0.0/24'
self.service_port = dict(
id='fake_service_port_id',
fixed_ips=[dict(ip_address='fake_service_port_ip_address')])
self.public_port = dict(
id='fake_tenant_port_id',
fixed_ips=[dict(ip_address='fake_public_port_ip_address')])
service_subnet = dict(id='fake_service_subnet')
instance = self._init_neutron_network_plugin()
instance.connect_share_server_to_tenant_network = True
self.mock_object(instance, '_get_service_network_id',
mock.Mock(return_value='fake_service_network_id'))
self.mock_object(
service_instance.neutron.API, 'admin_project_id',
mock.Mock(return_value=admin_project_id))
self.mock_object(
service_instance.neutron.API, 'create_port',
mock.Mock(side_effect=fake_create_port))
self.mock_object(
service_instance.neutron.API, 'subnet_create',
mock.Mock(return_value=service_subnet))
self.mock_object(
instance, 'setup_connectivity_with_service_instances',
mock.Mock(return_value=service_subnet))
self.mock_object(
instance, '_get_cidr_for_subnet', mock.Mock(return_value=cidr))
self.mock_object(
instance, '_get_service_subnet', mock.Mock(return_value=None))
expected = {
'ip_address': self.public_port['fixed_ips'][0]['ip_address'],
'public_port': self.public_port,
'service_port': self.service_port,
'service_subnet': service_subnet,
'ports': [self.public_port, self.service_port],
'nics': [{'port-id': self.public_port['id']},
{'port-id': self.service_port['id']}]}
result = instance.setup_network(network_info)
self.assertEqual(expected, result)
(instance.setup_connectivity_with_service_instances.
assert_called_once_with())
instance._get_service_subnet.assert_called_once_with(mock.ANY)
instance._get_cidr_for_subnet.assert_called_once_with()
self.assertTrue(service_instance.neutron.API.subnet_create.called)
self.assertTrue(service_instance.neutron.API.create_port.called)
def test_setup_network_and_connect_share_server_to_tenant_net_admin(self):
def fake_create_port(*aargs, **kwargs):
if aargs[1] == 'fake_admin_network_id':
return self.admin_port
elif aargs[1] == 'fake_tenant_network_id':
return self.public_port
else:
raise exception.ManilaException('Got unexpected data')
admin_project_id = 'fake_admin_project_id'
network_info = {
'neutron_net_id': 'fake_tenant_network_id',
'neutron_subnet_id': 'fake_tenant_subnet_id'}
self.admin_port = {
'id': 'fake_admin_port_id',
'fixed_ips': [{'ip_address': 'fake_admin_port_ip_address'}]}
self.public_port = {
'id': 'fake_tenant_port_id',
'fixed_ips': [{'ip_address': 'fake_public_port_ip_address'}]}
instance = self._init_neutron_network_plugin()
instance.use_admin_port = True
instance.use_service_network = False
instance.admin_network_id = 'fake_admin_network_id'
instance.admin_subnet_id = 'fake_admin_subnet_id'
instance.connect_share_server_to_tenant_network = True
self.mock_object(
service_instance.neutron.API, 'admin_project_id',
mock.Mock(return_value=admin_project_id))
self.mock_object(
service_instance.neutron.API, 'create_port',
mock.Mock(side_effect=fake_create_port))
self.mock_object(
instance, 'setup_connectivity_with_service_instances')
expected = {
'ip_address': self.public_port['fixed_ips'][0]['ip_address'],
'public_port': self.public_port,
'admin_port': self.admin_port,
'ports': [self.public_port, self.admin_port],
'nics': [{'port-id': self.public_port['id']},
{'port-id': self.admin_port['id']}]}
result = instance.setup_network(network_info)
self.assertEqual(expected, result)
(instance.setup_connectivity_with_service_instances.
assert_called_once_with())
self.assertTrue(service_instance.neutron.API.create_port.called)
@ddt.data(None, exception.NetworkException(code=400))
def test_setup_network_using_router_success(self, return_obj):
admin_project_id = 'fake_admin_project_id'
network_info = dict(
neutron_net_id='fake_tenant_network_id',
neutron_subnet_id='fake_tenant_subnet_id')
cidr = '13.0.0.0/24'
self.admin_port = {
'id': 'fake_admin_port_id',
'fixed_ips': [{'ip_address': 'fake_admin_port_ip_address'}]}
self.service_port = dict(
id='fake_service_port_id',
fixed_ips=[dict(ip_address='fake_service_port_ip_address')])
service_subnet = dict(id='fake_service_subnet')
instance = self._init_neutron_network_plugin()
instance.use_admin_port = True
instance.admin_network_id = 'fake_admin_network_id'
instance.admin_subnet_id = 'fake_admin_subnet_id'
instance.connect_share_server_to_tenant_network = False
self.mock_object(instance, '_get_service_network_id',
mock.Mock(return_value='fake_service_network_id'))
router = dict(id='fake_router_id')
self.mock_object(
service_instance.neutron.API, 'admin_project_id',
mock.Mock(return_value=admin_project_id))
self.mock_object(
service_instance.neutron.API, 'create_port',
mock.Mock(side_effect=[self.service_port, self.admin_port]))
self.mock_object(
service_instance.neutron.API, 'subnet_create',
mock.Mock(return_value=service_subnet))
self.mock_object(
instance, '_get_private_router', mock.Mock(return_value=router))
self.mock_object(
service_instance.neutron.API, 'router_add_interface',
mock.Mock(side_effect=return_obj))
self.mock_object(instance, 'setup_connectivity_with_service_instances')
self.mock_object(
instance, '_get_cidr_for_subnet', mock.Mock(return_value=cidr))
self.mock_object(
instance, '_get_service_subnet', mock.Mock(return_value=None))
expected = {
'ip_address': self.service_port['fixed_ips'][0]['ip_address'],
'service_port': self.service_port,
'service_subnet': service_subnet,
'admin_port': self.admin_port, 'router': router,
'ports': [self.service_port, self.admin_port],
'nics': [{'port-id': self.service_port['id']},
{'port-id': self.admin_port['id']}]}
result = instance.setup_network(network_info)
self.assertEqual(expected, result)
(instance.setup_connectivity_with_service_instances.
assert_called_once_with())
instance._get_service_subnet.assert_called_once_with(mock.ANY)
instance._get_cidr_for_subnet.assert_called_once_with()
self.assertTrue(service_instance.neutron.API.subnet_create.called)
self.assertTrue(service_instance.neutron.API.create_port.called)
instance._get_private_router.assert_called_once_with(
network_info['neutron_net_id'], network_info['neutron_subnet_id'])
(service_instance.neutron.API.router_add_interface.
assert_called_once_with(router['id'], service_subnet['id']))
def test_setup_network_using_router_addon_of_interface_failed(self):
network_info = dict(
neutron_net_id='fake_tenant_network_id',
neutron_subnet_id='fake_tenant_subnet_id')
service_subnet = dict(id='fake_service_subnet')
instance = self._init_neutron_network_plugin()
instance.connect_share_server_to_tenant_network = False
self.mock_object(instance, '_get_service_network_id',
mock.Mock(return_value='fake_service_network_id'))
router = dict(id='fake_router_id')
self.mock_object(
instance, '_get_private_router', mock.Mock(return_value=router))
self.mock_object(
service_instance.neutron.API, 'router_add_interface',
mock.Mock(side_effect=exception.NetworkException(code=500)))
self.mock_object(
instance, '_get_service_subnet',
mock.Mock(return_value=service_subnet))
self.assertRaises(
exception.NetworkException,
instance.setup_network, network_info)
instance._get_service_subnet.assert_called_once_with(mock.ANY)
instance._get_private_router.assert_called_once_with(
network_info['neutron_net_id'], network_info['neutron_subnet_id'])
(service_instance.neutron.API.router_add_interface.
assert_called_once_with(router['id'], service_subnet['id']))
def test_setup_network_using_router_connectivity_verification_fail(self):
admin_project_id = 'fake_admin_project_id'
network_info = dict