class TestFCZoneManager(test.TestCase):
- def setUp(self):
+ @mock.patch('oslo_config.cfg._is_opt_registered', return_value=False)
+ def setUp(self, opt_mock):
super(TestFCZoneManager, self).setUp()
config = conf.Configuration(None)
config.fc_fabric_names = fabric_name
self.driver = Mock(FCZoneDriver)
def __init__(self, *args, **kwargs):
- test.TestCase.__init__(self, *args, **kwargs)
+ super(TestFCZoneManager, self).__init__(*args, **kwargs)
- def test_add_connection(self):
+ @mock.patch('oslo_config.cfg._is_opt_registered', return_value=False)
+ def test_add_connection(self, opt_mock):
with mock.patch.object(self.zm.driver, 'add_connection')\
as add_connection_mock:
self.zm.driver.get_san_context.return_value = fabric_map
add_connection_mock.assert_called_once_with(fabric_name,
init_target_map)
- def test_add_connection_error(self):
+ @mock.patch('oslo_config.cfg._is_opt_registered', return_value=False)
+ def test_add_connection_error(self, opt_mock):
with mock.patch.object(self.zm.driver, 'add_connection')\
as add_connection_mock:
add_connection_mock.side_effect = exception.FCZoneDriverException
self.assertRaises(exception.ZoneManagerException,
self.zm.add_connection, init_target_map)
- def test_delete_connection(self):
+ @mock.patch('oslo_config.cfg._is_opt_registered', return_value=False)
+ def test_delete_connection(self, opt_mock):
with mock.patch.object(self.zm.driver, 'delete_connection')\
as delete_connection_mock:
self.zm.driver.get_san_context.return_value = fabric_map
delete_connection_mock.assert_called_once_with(fabric_name,
init_target_map)
- def test_delete_connection_error(self):
+ @mock.patch('oslo_config.cfg._is_opt_registered', return_value=False)
+ def test_delete_connection_error(self, opt_mock):
with mock.patch.object(self.zm.driver, 'delete_connection')\
as del_connection_mock:
del_connection_mock.side_effect = exception.FCZoneDriverException
self.driver = None
def __init__(self, *args, **kwargs):
- test.TestCase.__init__(self, *args, **kwargs)
+ super(TestVolumeDriver, self).__init__(*args, **kwargs)
+ @mock.patch('oslo_config.cfg._is_opt_registered', return_value=False)
@mock.patch.object(utils, 'require_driver_initialized')
- def test_initialize_connection_with_decorator(self, utils_mock):
+ def test_initialize_connection_with_decorator(self, utils_mock, opt_mock):
utils_mock.return_value = True
with mock.patch.object(fc_zone_manager.ZoneManager, 'add_connection')\
as add_zone_mock:
self.driver.no_zone_initialize_connection(None, None)
assert not add_zone_mock.called
+ @mock.patch('oslo_config.cfg._is_opt_registered', return_value=False)
@mock.patch.object(utils, 'require_driver_initialized')
- def test_terminate_connection_with_decorator(self, utils_mock):
+ def test_terminate_connection_with_decorator(self, utils_mock, opt_mock):
utils_mock.return_value = True
with mock.patch.object(fc_zone_manager.ZoneManager,
'delete_connection') as remove_zone_mock:
lookup_service, configuration=self.configuration)
else:
msg = _("Lookup service not configured. Config option for "
- "fc_san_lookup_service need to specify a concrete "
- "implementation of lookup service")
+ "fc_san_lookup_service needs to specify a concrete "
+ "implementation of the lookup service.")
LOG.error(msg)
raise exception.FCSanLookupServiceException(msg)
try:
help='FC Zone Driver responsible for zone management'),
cfg.StrOpt('zoning_policy',
default='initiator-target',
- help='Zoning policy configured by user'),
+ help='Zoning policy configured by user; valid values include '
+ '"initiator-target" or "initiator"'),
cfg.StrOpt('fc_fabric_names',
default=None,
- help='Comma separated list of fibre channel fabric names.'
+ help='Comma separated list of Fibre Channel fabric names.'
' This list of names is used to retrieve other SAN credentials'
' for connecting to each SAN fabric'),
cfg.StrOpt('fc_san_lookup_service',
default='cinder.zonemanager.drivers.brocade'
'.brcd_fc_san_lookup_service.BrcdFCSanLookupService',
- help='FC San Lookup Service'),
+ help='FC SAN Lookup Service'),
]
CONF = cfg.CONF
"""Load the driver from the one specified in args, or from flags."""
super(ZoneManager, self).__init__(**kwargs)
- self.configuration = kwargs.get('configuration', None)
+ self.configuration = kwargs.get('configuration')
if self.configuration:
self.configuration.append_config_values(zone_manager_opts)
try:
for initiator in initiator_target_map.keys():
target_list = initiator_target_map[initiator]
- LOG.debug("Target List :%s", {initiator: target_list})
+ LOG.debug("Target List: %s", target_list)
# get SAN context for the target list
fabric_map = self.get_san_context(target_list)
- LOG.debug("Fabric Map after context lookup:%s", fabric_map)
+ LOG.debug("Fabric Map after context lookup: %s", fabric_map)
# iterate over each SAN and apply connection control
for fabric in fabric_map.keys():
connected_fabric = fabric
valid_i_t_map = self.get_valid_initiator_target_map(
i_t_map, True)
LOG.info(_LI("Final filtered map for fabric: %s"),
- {fabric: valid_i_t_map})
+ valid_i_t_map)
# Call driver to add connection control
self.driver.add_connection(fabric, valid_i_t_map)
"over all target list"))
except Exception as e:
msg = _("Failed adding connection for fabric=%(fabric)s: "
- "Error:%(err)s") % {'fabric': connected_fabric,
- 'err': e}
+ "Error: %(err)s") % {'fabric': connected_fabric,
+ 'err': e}
LOG.error(msg)
raise exception.ZoneManagerException(reason=msg)
try:
for initiator in initiator_target_map.keys():
target_list = initiator_target_map[initiator]
- LOG.info(_LI("Delete connection Target List:%s"),
- {initiator: target_list})
+ LOG.info(_LI("Delete connection Target List: %s"),
+ target_list)
# get SAN context for the target list
fabric_map = self.get_san_context(target_list)
" target list")
except Exception as e:
msg = _("Failed removing connection for fabric=%(fabric)s: "
- "Error:%(err)s") % {'fabric': connected_fabric,
- 'err': e}
+ "Error: %(err)s") % {'fabric': connected_fabric,
+ 'err': e}
LOG.error(msg)
raise exception.ZoneManagerException(reason=msg)
to list of target WWNs visible to the fabric.
"""
fabric_map = self.driver.get_san_context(target_wwn_list)
- LOG.debug("Got SAN context:%s", fabric_map)
+ LOG.debug("Got SAN context: %s", fabric_map)
return fabric_map
def get_valid_initiator_target_map(self, initiator_target_map,
def create_zone_manager():
"""If zoning is enabled, build the Zone Manager."""
config = Configuration(manager.volume_manager_opts)
- LOG.debug("zoning mode %s" % config.safe_get('zoning_mode'))
+ LOG.debug("Zoning mode: %s", config.safe_get('zoning_mode'))
if config.safe_get('zoning_mode') == 'fabric':
LOG.debug("FC Zone Manager enabled.")
zm = fc_zone_manager.ZoneManager(configuration=config)
LOG.info(_LI("Using FC Zone Manager %(zm_version)s,"
- " Driver %(drv_name)s %(drv_version)s.") %
+ " Driver %(drv_name)s %(drv_version)s."),
{'zm_version': zm.get_version(),
'drv_name': zm.driver.__class__.__name__,
'drv_version': zm.driver.get_version()})
def create_lookup_service():
config = Configuration(manager.volume_manager_opts)
- LOG.debug("zoning mode %s" % config.safe_get('zoning_mode'))
+ LOG.debug("Zoning mode: %s", config.safe_get('zoning_mode'))
if config.safe_get('zoning_mode') == 'fabric':
LOG.debug("FC Lookup Service enabled.")
lookup = fc_san_lookup_service.FCSanLookupService(configuration=config)
- LOG.info(_LI("Using FC lookup service %s") % lookup.lookup_service)
+ LOG.info(_LI("Using FC lookup service %s"), lookup.lookup_service)
return lookup
else:
LOG.debug("FC Lookup Service not enabled in cinder.conf.")
init_target_map = conn_info['data']['initiator_target_map']
zm = create_zone_manager()
if zm:
- LOG.debug("Add FC Zone for mapping '%s'." %
+ LOG.debug("Add FC Zone for mapping '%s'.",
init_target_map)
zm.add_connection(init_target_map)
init_target_map = conn_info['data']['initiator_target_map']
zm = create_zone_manager()
if zm:
- LOG.debug("Remove FC Zone for mapping '%s'." %
+ LOG.debug("Remove FC Zone for mapping '%s'.",
init_target_map)
zm.delete_connection(init_target_map)