props = {}
props['ip'] = CONF.my_ip
props['host'] = socket.gethostname()
- props['initiator'] = iscsi.get_initiator()
- props['wwpns'] = fc.get_fc_wwpns()
+ initiator = iscsi.get_initiator()
+ if initiator:
+ props['initiator'] = initiator
+ wwpns = fc.get_fc_wwpns()
+ if wwpns:
+ props['wwpns'] = wwpns
return props
if l.startswith('InitiatorName='):
return l[l.index('=') + 1:].strip()
except exception.ProcessExecutionError:
- raise exception.FileNotFound(file_path=file_path)
+ return None
def _run_iscsiadm(self, connection_properties, iscsi_command, **kwargs):
check_exit_code = kwargs.pop('check_exit_code', 0)
LOG.warn(_("systool is not installed"))
return []
+ # No FC HBAs were found
if out is None:
- raise RuntimeError(_("Cannot find any Fibre Channel HBAs"))
+ return []
lines = out.split('\n')
# ignore the first 2 lines
# Note(walter-boring) modern linux kernels contain the FC HBA's in /sys
# and are obtainable via the systool app
hbas = self.get_fc_hbas()
+ if not hbas:
+ return []
+
hbas_info = []
for hba in hbas:
wwpn = hba['port_name'].replace('0x', '')
# Note(walter-boring) modern linux kernels contain the FC HBA's in /sys
# and are obtainable via the systool app
hbas = self.get_fc_hbas()
+ if not hbas:
+ return []
wwnns = []
if hbas:
}
}
+ def test_get_initiator(self):
+ def initiator_no_file(*args, **kwargs):
+ raise exception.ProcessExecutionError('No file')
+
+ def initiator_get_text(*arg, **kwargs):
+ text = ('## DO NOT EDIT OR REMOVE THIS FILE!\n'
+ '## If you remove this file, the iSCSI daemon '
+ 'will not start.\n'
+ '## If you change the InitiatorName, existing '
+ 'access control lists\n'
+ '## may reject this initiator. The InitiatorName must '
+ 'be unique\n'
+ '## for each iSCSI initiator. Do NOT duplicate iSCSI '
+ 'InitiatorNames.\n'
+ 'InitiatorName=iqn.1234-56.foo.bar:01:23456789abc')
+ return text, None
+
+ self.stubs.Set(self.connector, '_execute', initiator_no_file)
+ initiator = self.connector.get_initiator()
+ self.assertEquals(initiator, None)
+ self.stubs.Set(self.connector, '_execute', initiator_get_text)
+ initiator = self.connector.get_initiator()
+ self.assertEquals(initiator, 'iqn.1234-56.foo.bar:01:23456789abc')
+
@test.testtools.skipUnless(os.path.exists('/dev/disk/by-path'),
'Test requires /dev/disk/by-path')
def test_connect_volume(self):
'tee -a /sys/class/scsi_host/bar/scan']
self.assertEquals(expected_commands, self.cmds)
+ def test_get_fc_hbas_fail(self):
+ def fake_exec1(a, b, c, d, run_as_root=True, root_helper='sudo'):
+ raise OSError
+
+ def fake_exec2(a, b, c, d, run_as_root=True, root_helper='sudo'):
+ return None, 'None found'
+
+ self.stubs.Set(self.lfc, "_execute", fake_exec1)
+ hbas = self.lfc.get_fc_hbas()
+ self.assertEquals(0, len(hbas))
+ self.stubs.Set(self.lfc, "_execute", fake_exec2)
+ hbas = self.lfc.get_fc_hbas()
+ self.assertEquals(0, len(hbas))
+
def test_get_fc_hbas(self):
def fake_exec(a, b, c, d, run_as_root=True, root_helper='sudo'):
return SYSTOOL_FC, None