@mock.patch('os.path.exists', return_value=True)
def test_libvirt_aoe_driver(self, exists):
- libvirt_driver = aoe.LibvirtAOEVolumeDriver(self.fake_conn)
+ libvirt_driver = aoe.LibvirtAOEVolumeDriver(self.fake_host)
self.assertIsInstance(libvirt_driver.connector,
connector.AoEConnector)
def test_libvirt_disco_driver(self):
libvirt_driver = disco.LibvirtDISCOVolumeDriver(
- self.fake_conn)
+ self.fake_host)
self.assertIsInstance(libvirt_driver.connector,
connector.DISCOConnector)
def test_libvirt_disco_driver_connect(self):
- dcon = disco.LibvirtDISCOVolumeDriver(self.fake_conn)
+ dcon = disco.LibvirtDISCOVolumeDriver(self.fake_host)
conf = {'server_ip': '127.0.0.1', 'server_port': 9898}
disk_info = {'disco_id': '1234567',
'name': 'aDiscoVolume',
conn['data']['device_path'])
def test_libvirt_disco_driver_get_config(self):
- dcon = disco.LibvirtDISCOVolumeDriver(self.fake_conn)
+ dcon = disco.LibvirtDISCOVolumeDriver(self.fake_host)
disk_info = {'path': '/dev/dms1234567', 'name': 'aDiscoVolume',
'type': 'raw', 'dev': 'vda1', 'bus': 'pci0',
self.assertEqual('disco', conf.source_protocol)
def test_libvirt_disco_driver_disconnect(self):
- dcon = disco.LibvirtDISCOVolumeDriver(self.fake_conn)
+ dcon = disco.LibvirtDISCOVolumeDriver(self.fake_host)
dcon.connector.disconnect_volume = mock.MagicMock()
disk_info = {'path': '/dev/dms1234567', 'name': 'aDiscoVolume',
'type': 'raw', 'dev': 'vda1', 'bus': 'pci0',
def test_libvirt_fibrechan_driver(self):
libvirt_driver = fibrechannel.LibvirtFibreChannelVolumeDriver(
- self.fake_conn)
+ self.fake_host)
self.assertIsInstance(libvirt_driver.connector,
connector.FibreChannelConnector)
def _test_libvirt_fibrechan_driver_s390(self):
libvirt_driver = fibrechannel.LibvirtFibreChannelVolumeDriver(
- self.fake_conn)
+ self.fake_host)
self.assertIsInstance(libvirt_driver.connector,
connector.FibreChannelConnectorS390X)
def test_libvirt_fibrechan_driver_get_config(self):
libvirt_driver = fibrechannel.LibvirtFibreChannelVolumeDriver(
- self.fake_conn)
+ self.fake_host)
device_path = '/dev/fake-dev'
connection_info = {'data': {'device_path': device_path}}
mnt_base = '/mnt'
self.flags(glusterfs_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_conn)
+ libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_host)
export_string = '192.168.1.1:/volume-00001'
export_mnt_base = os.path.join(mnt_base,
utils.get_hash_str(export_string))
mnt_base = '/mnt'
self.flags(glusterfs_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_conn)
+ libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_host)
export_string = '192.168.1.1:/volume-00001'
export_mnt_base = os.path.join(mnt_base,
utils.get_hash_str(export_string))
mnt_base = '/mnt'
self.flags(glusterfs_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_conn)
+ libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_host)
export_string = '192.168.1.1:/volume-00001'
export_mnt_base = os.path.join(mnt_base,
utils.get_hash_str(export_string))
export_string = '192.168.1.1:/volume-00001'
connection_info = {'data': {'export': export_string,
'name': self.name}}
- libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_conn)
+ libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_host)
mock_utils_exe.side_effect = processutils.ProcessExecutionError(
None, None, None, 'umount', 'umount: target is busy.')
libvirt_driver.disconnect_volume(connection_info, "vde")
mnt_base = '/mnt'
self.flags(glusterfs_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_conn)
+ libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_host)
export_string = '192.168.1.1:/volume-00001'
options = '-o backupvolfile-server=192.168.1.2'
export_mnt_base = os.path.join(mnt_base,
@mock.patch.object(libvirt_utils, 'is_mounted', return_value=False)
def test_libvirt_glusterfs_libgfapi(self, mock_is_mounted):
self.flags(qemu_allowed_storage_drivers=['gluster'], group='libvirt')
- libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_conn)
+ libvirt_driver = glusterfs.LibvirtGlusterfsVolumeDriver(self.fake_host)
export_string = '192.168.1.1:/volume-00001'
name = 'volume-00001'
class LibvirtGPFSVolumeDriverTestCase(test_volume.LibvirtVolumeBaseTestCase):
def test_libvirt_gpfs_driver_get_config(self):
- libvirt_driver = gpfs.LibvirtGPFSVolumeDriver(self.fake_conn)
+ libvirt_driver = gpfs.LibvirtGPFSVolumeDriver(self.fake_host)
connection_info = {
'driver_volume_type': 'gpfs',
'data': {
class LibvirtHGSTVolumeDriverTestCase(test_volume.LibvirtVolumeBaseTestCase):
def test_libvirt_hgst_driver_type(self):
- drvr = hgst.LibvirtHGSTVolumeDriver(self.fake_conn)
+ drvr = hgst.LibvirtHGSTVolumeDriver(self.fake_host)
self.assertIsInstance(drvr.connector, connector.HGSTConnector)
def test_libvirt_hgst_driver_connect(self):
def brick_conn_vol(data):
return {'path': '/dev/space01'}
- drvr = hgst.LibvirtHGSTVolumeDriver(self.fake_conn)
+ drvr = hgst.LibvirtHGSTVolumeDriver(self.fake_host)
drvr.connector.connect_volume = brick_conn_vol
di = {'path': '/dev/space01', 'name': 'space01'}
ci = {'data': di}
ci['data']['device_path'])
def test_libvirt_hgst_driver_get_config(self):
- drvr = hgst.LibvirtHGSTVolumeDriver(self.fake_conn)
+ drvr = hgst.LibvirtHGSTVolumeDriver(self.fake_host)
di = {'path': '/dev/space01', 'name': 'space01', 'type': 'raw',
'dev': 'vda1', 'bus': 'pci0', 'device_path': '/dev/space01'}
ci = {'data': di}
self.assertEqual('/dev/space01', conf.source_path)
def test_libvirt_hgst_driver_disconnect(self):
- drvr = hgst.LibvirtHGSTVolumeDriver(self.fake_conn)
+ drvr = hgst.LibvirtHGSTVolumeDriver(self.fake_host)
drvr.connector.disconnect_volume = mock.MagicMock()
di = {'path': '/dev/space01', 'name': 'space01', 'type': 'raw',
'dev': 'vda1', 'bus': 'pci0', 'device_path': '/dev/space01'}
self.assertEqual(targets, out)
def test_libvirt_iscsi_driver(self, transport=None):
- libvirt_driver = iscsi.LibvirtISCSIVolumeDriver(self.fake_conn)
+ libvirt_driver = iscsi.LibvirtISCSIVolumeDriver(self.fake_host)
self.assertIsInstance(libvirt_driver.connector,
connector.ISCSIConnector)
def test_libvirt_iscsi_driver_get_config(self):
- libvirt_driver = iscsi.LibvirtISCSIVolumeDriver(self.fake_conn)
+ libvirt_driver = iscsi.LibvirtISCSIVolumeDriver(self.fake_host)
device_path = '/dev/fake-dev'
connection_info = {'data': {'device_path': device_path}}
device_path = '/dev/fake-dev'
connection_info = {'data': {'device_path': device_path}}
- libvirt_driver = iscsi.LibvirtISCSIVolumeDriver(self.fake_conn)
+ libvirt_driver = iscsi.LibvirtISCSIVolumeDriver(self.fake_host)
libvirt_driver.connector.disconnect_volume = mock.MagicMock(
side_effect=os_brick_exception.VolumeDeviceNotFound(
device=device_path))
"""Tests the libvirt iSER volume driver."""
def test_get_transport(self):
- driver = iser.LibvirtISERVolumeDriver(self.fake_conn)
+ driver = iser.LibvirtISERVolumeDriver(self.fake_host)
self.assertEqual('iser', driver._get_transport())
}
def test_libvirt_sheepdog_driver(self):
- libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_conn)
+ libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_host)
connection_info = self.sheepdog_connection(self.vol)
conf = libvirt_driver.get_config(connection_info, self.disk_info)
tree = conf.format_dom()
}
def test_libvirt_rbd_driver(self):
- libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_conn)
+ libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_host)
connection_info = self.rbd_connection(self.vol)
conf = libvirt_driver.get_config(connection_info, self.disk_info)
tree = conf.format_dom()
libvirt_driver.disconnect_volume(connection_info, "vde")
def test_libvirt_rbd_driver_hosts(self):
- libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_conn)
+ libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_host)
connection_info = self.rbd_connection(self.vol)
hosts = ['example.com', '1.2.3.4', '::1']
ports = [None, '6790', '6791']
libvirt_driver.disconnect_volume(connection_info, "vde")
def test_libvirt_rbd_driver_auth_enabled(self):
- libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_conn)
+ libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_host)
connection_info = self.rbd_connection(self.vol)
secret_type = 'ceph'
connection_info['data']['auth_enabled'] = True
libvirt_driver.disconnect_volume(connection_info, "vde")
def test_libvirt_rbd_driver_auth_enabled_flags_override(self):
- libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_conn)
+ libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_host)
connection_info = self.rbd_connection(self.vol)
secret_type = 'ceph'
connection_info['data']['auth_enabled'] = True
libvirt_driver.disconnect_volume(connection_info, "vde")
def test_libvirt_rbd_driver_auth_disabled(self):
- libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_conn)
+ libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_host)
connection_info = self.rbd_connection(self.vol)
secret_type = 'ceph'
connection_info['data']['auth_enabled'] = False
libvirt_driver.disconnect_volume(connection_info, "vde")
def test_libvirt_rbd_driver_auth_disabled_flags_override(self):
- libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_conn)
+ libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_host)
connection_info = self.rbd_connection(self.vol)
secret_type = 'ceph'
connection_info['data']['auth_enabled'] = False
mock_find):
mock_find.return_value = test_volume.FakeSecret()
mock_create.return_value = test_volume.FakeSecret()
- libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_conn)
+ libvirt_driver = net.LibvirtNetVolumeDriver(self.fake_host)
connection_info = self.iscsi_connection(self.vol, self.location,
self.iqn, auth=True)
secret_type = 'iscsi'
@mock.patch.object(libvirt_utils, 'is_mounted', return_value=False)
def test_libvirt_nfs_driver(self, mock_is_mounted):
- libvirt_driver = nfs.LibvirtNFSVolumeDriver(self.fake_conn)
+ libvirt_driver = nfs.LibvirtNFSVolumeDriver(self.fake_host)
export_string = '192.168.1.1:/nfs/share1'
export_mnt_base = os.path.join(self.mnt_base,
export_string = '192.168.1.1:/nfs/share1'
connection_info = {'data': {'export': export_string,
'name': self.name}}
- libvirt_driver = nfs.LibvirtNFSVolumeDriver(self.fake_conn)
+ libvirt_driver = nfs.LibvirtNFSVolumeDriver(self.fake_host)
mock_utils_exe.side_effect = processutils.ProcessExecutionError(
None, None, None, 'umount', 'umount: device is busy.')
libvirt_driver.disconnect_volume(connection_info, "vde")
self.assertTrue(mock_LOG_exception.called)
def test_libvirt_nfs_driver_get_config(self):
- libvirt_driver = nfs.LibvirtNFSVolumeDriver(self.fake_conn)
+ libvirt_driver = nfs.LibvirtNFSVolumeDriver(self.fake_host)
export_string = '192.168.1.1:/nfs/share1'
export_mnt_base = os.path.join(self.mnt_base,
utils.get_hash_str(export_string))
self.assertEqual('native', tree.find('./driver').get('io'))
def test_libvirt_nfs_driver_already_mounted(self):
- libvirt_driver = nfs.LibvirtNFSVolumeDriver(self.fake_conn)
+ libvirt_driver = nfs.LibvirtNFSVolumeDriver(self.fake_host)
export_string = '192.168.1.1:/nfs/share1'
export_mnt_base = os.path.join(self.mnt_base,
@mock.patch.object(libvirt_utils, 'is_mounted', return_value=False)
def test_libvirt_nfs_driver_with_opts(self, mock_is_mounted):
- libvirt_driver = nfs.LibvirtNFSVolumeDriver(self.fake_conn)
+ libvirt_driver = nfs.LibvirtNFSVolumeDriver(self.fake_host)
export_string = '192.168.1.1:/nfs/share1'
options = '-o intr,nfsvers=3'
export_mnt_base = os.path.join(self.mnt_base,
mnt_base = '/mnt'
self.flags(quobyte_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_conn)
+ libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_host)
export_string = 'quobyte://192.168.1.1/volume-00001'
quobyte_volume = '192.168.1.1/volume-00001'
export_mnt_base = os.path.join(mnt_base,
mnt_base = '/mnt'
self.flags(quobyte_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_conn)
+ libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_host)
export_string = 'quobyte://192.168.1.1/volume-00001'
quobyte_volume = '192.168.1.1/volume-00001'
export_mnt_base = os.path.join(mnt_base,
mnt_base = '/mnt'
self.flags(quobyte_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_conn)
+ libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_host)
export_string = 'quobyte://192.168.1.1/volume-00001'
quobyte_volume = '192.168.1.1/volume-00001'
export_mnt_base = os.path.join(mnt_base,
):
mnt_base = '/mnt'
self.flags(quobyte_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_conn)
+ libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_host)
export_string = 'quobyte://192.168.1.1/volume-00001'
name = 'volume-00001'
image_format = 'qcow2'
mnt_base = '/mnt'
self.flags(quobyte_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_conn)
+ libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_host)
export_string = 'quobyte://192.168.1.1/volume-00001'
connection_info = {'data': {'export': export_string,
mnt_base = '/mnt'
self.flags(quobyte_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_conn)
+ libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_host)
export_string = 'quobyte://192.168.1.1/volume-00001'
self.assertEqual("192.168.1.1/volume-00001",
libvirt_driver._normalize_export(export_string))
mnt_base = '/mnt'
self.flags(quobyte_mount_point_base=mnt_base, group='libvirt')
- libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_conn)
+ libvirt_driver = quobyte.LibvirtQuobyteVolumeDriver(self.fake_host)
export_string = '192.168.1.1/volume-00001'
self.assertEqual("192.168.1.1/volume-00001",
libvirt_driver._normalize_export(export_string))
def test_libvirt_scaleio_driver(self):
libvirt_driver = scaleio.LibvirtScaleIOVolumeDriver(
- self.fake_conn)
+ self.fake_host)
self.assertIsInstance(libvirt_driver.connector,
connector.ScaleIOConnector)
def brick_conn_vol(data):
return {'path': '/dev/vol01'}
- sio = scaleio.LibvirtScaleIOVolumeDriver(self.fake_conn)
+ sio = scaleio.LibvirtScaleIOVolumeDriver(self.fake_host)
sio.connector.connect_volume = brick_conn_vol
disk_info = {'path': '/dev/vol01', 'name': 'vol01'}
conn = {'data': disk_info}
conn['data']['device_path'])
def test_libvirt_scaleio_driver_get_config(self):
- sio = scaleio.LibvirtScaleIOVolumeDriver(self.fake_conn)
+ sio = scaleio.LibvirtScaleIOVolumeDriver(self.fake_host)
disk_info = {'path': '/dev/vol01', 'name': 'vol01', 'type': 'raw',
'dev': 'vda1', 'bus': 'pci0', 'device_path': '/dev/vol01'}
conn = {'data': disk_info}
self.assertEqual('/dev/vol01', conf.source_path)
def test_libvirt_scaleio_driver_disconnect(self):
- sio = scaleio.LibvirtScaleIOVolumeDriver(self.fake_conn)
+ sio = scaleio.LibvirtScaleIOVolumeDriver(self.fake_host)
sio.connector.disconnect_volume = mock.MagicMock()
disk_info = {'path': '/dev/vol01', 'name': 'vol01', 'type': 'raw',
'dev': 'vda1', 'bus': 'pci0', 'device_path': '/dev/vol01'}
scality_sofs_mount_point=self.scality_sofs_mount_point,
group='libvirt')
- self.drv = scality.LibvirtScalityVolumeDriver(self.fake_conn)
+ self.drv = scality.LibvirtScalityVolumeDriver(self.fake_host)
@mock.patch('six.moves.urllib.request.urlopen')
def test_connect_volume(self, mock_urlopen):
def test_libvirt_smbfs_driver(self, mock_is_mounted):
mock_is_mounted.return_value = False
- libvirt_driver = smbfs.LibvirtSMBFSVolumeDriver(self.fake_conn)
+ libvirt_driver = smbfs.LibvirtSMBFSVolumeDriver(self.fake_host)
export_string = '//192.168.1.1/volumes'
export_mnt_base = os.path.join(self.mnt_base,
utils.get_hash_str(export_string))
self.assertEqual(expected_commands, self.executes)
def test_libvirt_smbfs_driver_already_mounted(self):
- libvirt_driver = smbfs.LibvirtSMBFSVolumeDriver(self.fake_conn)
+ libvirt_driver = smbfs.LibvirtSMBFSVolumeDriver(self.fake_host)
export_string = '//192.168.1.1/volumes'
export_mnt_base = os.path.join(self.mnt_base,
utils.get_hash_str(export_string))
self.assertEqual(expected_commands, self.executes)
def test_libvirt_smbfs_driver_get_config(self):
- libvirt_driver = smbfs.LibvirtSMBFSVolumeDriver(self.fake_conn)
+ libvirt_driver = smbfs.LibvirtSMBFSVolumeDriver(self.fake_host)
export_string = '//192.168.1.1/volumes'
export_mnt_base = os.path.join(self.mnt_base,
utils.get_hash_str(export_string))
def test_libvirt_smbfs_driver_with_opts(self, mock_is_mounted):
mock_is_mounted.return_value = False
- libvirt_driver = smbfs.LibvirtSMBFSVolumeDriver(self.fake_conn)
+ libvirt_driver = smbfs.LibvirtSMBFSVolumeDriver(self.fake_host)
export_string = '//192.168.1.1/volumes'
options = '-o user=guest,uid=107,gid=105'
export_mnt_base = os.path.join(self.mnt_base,
self.useFixture(fakelibvirt.FakeLibvirtFixture())
- class FakeLibvirtDriver(object):
- def __init__(self):
- self._host = host.Host("qemu:///system")
+ self.fake_host = host.Host("qemu:///system")
- def _get_all_block_devices(self):
- return []
-
- self.fake_conn = FakeLibvirtDriver()
self.connr = {
'ip': '127.0.0.1',
'initiator': 'fake_initiator',
self.assertEqual(disk_info['dev'], tree.find('./target').get('dev'))
def _test_libvirt_volume_driver_disk_info(self):
- libvirt_driver = volume.LibvirtVolumeDriver(self.fake_conn)
+ libvirt_driver = volume.LibvirtVolumeDriver(self.fake_host)
connection_info = {
'driver_volume_type': 'fake',
'data': {
self._test_libvirt_volume_driver_disk_info()
def test_libvirt_volume_driver_serial(self):
- libvirt_driver = volume.LibvirtVolumeDriver(self.fake_conn)
+ libvirt_driver = volume.LibvirtVolumeDriver(self.fake_host)
connection_info = {
'driver_volume_type': 'fake',
'data': {
self.assertIsNone(tree.find("driver[@discard]"))
def test_libvirt_volume_driver_blockio(self):
- libvirt_driver = volume.LibvirtVolumeDriver(self.fake_conn)
+ libvirt_driver = volume.LibvirtVolumeDriver(self.fake_host)
connection_info = {
'driver_volume_type': 'fake',
'data': {
self.assertEqual('4096', blockio.get('physical_block_size'))
def test_libvirt_volume_driver_iotune(self):
- libvirt_driver = volume.LibvirtVolumeDriver(self.fake_conn)
+ libvirt_driver = volume.LibvirtVolumeDriver(self.fake_host)
connection_info = {
'driver_volume_type': 'fake',
'data': {
self.assertEqual('200', tree.find('./iotune/write_iops_sec').text)
def test_libvirt_volume_driver_readonly(self):
- libvirt_driver = volume.LibvirtVolumeDriver(self.fake_conn)
+ libvirt_driver = volume.LibvirtVolumeDriver(self.fake_host)
connection_info = {
'driver_volume_type': 'fake',
'data': {
def test_libvirt_volume_driver_discard_true(self, mock_has_min_version):
# Check the discard attrib is present in driver section
mock_has_min_version.return_value = True
- libvirt_driver = volume.LibvirtVolumeDriver(self.fake_conn)
+ libvirt_driver = volume.LibvirtVolumeDriver(self.fake_host)
connection_info = {
'driver_volume_type': 'fake',
'data': {
def test_libvirt_volume_driver_discard_false(self):
# Check the discard attrib is not present in driver section
- libvirt_driver = volume.LibvirtVolumeDriver(self.fake_conn)
+ libvirt_driver = volume.LibvirtVolumeDriver(self.fake_host)
connection_info = {
'driver_volume_type': 'fake',
'data': {
self, mock_has_min_version):
# Check the discard attrib is not present in driver section
mock_has_min_version.return_value = False
- libvirt_driver = volume.LibvirtVolumeDriver(self.fake_conn)
+ libvirt_driver = volume.LibvirtVolumeDriver(self.fake_host)
connection_info = {
'driver_volume_type': 'fake',
'data': {
group='libvirt')
def test_libvirt_vzstorage_driver(self):
- libvirt_driver = vzstorage.LibvirtVZStorageVolumeDriver(self.fake_conn)
+ libvirt_driver = vzstorage.LibvirtVZStorageVolumeDriver(self.fake_host)
self.assertIsInstance(libvirt_driver.connector,
connector.RemoteFsConnector)
self.flags(vzstorage_mount_opts=opts, group='libvirt')
self.assertRaises(exception.NovaException,
vzstorage.LibvirtVZStorageVolumeDriver,
- self.fake_conn)
+ self.fake_host)
def test_libvirt_vzstorage_driver_share_fmt_neg(self):
- drv = vzstorage.LibvirtVZStorageVolumeDriver(self.fake_conn)
+ drv = vzstorage.LibvirtVZStorageVolumeDriver(self.fake_host)
wrong_export_string = 'mds1, mds2:/testcluster:passwd12111'
connection_info = {'data': {'export': wrong_export_string,
def brick_conn_vol(data):
return {'path': 'vstorage://testcluster'}
- drv = vzstorage.LibvirtVZStorageVolumeDriver(self.fake_conn)
+ drv = vzstorage.LibvirtVZStorageVolumeDriver(self.fake_host)
drv.connector.connect_volume = brick_conn_vol
export_string = 'testcluster'
connection_info['data']['options'])
def test_libvirt_vzstorage_driver_disconnect(self):
- drv = vzstorage.LibvirtVZStorageVolumeDriver(self.fake_conn)
+ drv = vzstorage.LibvirtVZStorageVolumeDriver(self.fake_host)
drv.connector.disconnect_volume = mock.MagicMock()
conn = {'data': self.disk_info}
drv.disconnect_volume(conn, self.disk_info)
self.disk_info, None)
def test_libvirt_vzstorage_driver_get_config(self):
- libvirt_driver = vzstorage.LibvirtVZStorageVolumeDriver(self.fake_conn)
+ libvirt_driver = vzstorage.LibvirtVZStorageVolumeDriver(self.fake_host)
export_string = 'vzstorage'
export_mnt_base = os.path.join(self.mnt_base,
utils.get_hash_str(export_string))
self.vif_driver = libvirt_vif.LibvirtGenericVIFDriver()
self.volume_drivers = driver.driver_dict_from_config(
- self._get_volume_drivers(), self)
+ self._get_volume_drivers(), self._host)
self._disk_cachemode = None
self.image_cache_manager = imagecache.ImageCacheManager()
class LibvirtAOEVolumeDriver(libvirt_volume.LibvirtBaseVolumeDriver):
"""Driver to attach AoE volumes to libvirt."""
- def __init__(self, connection):
+ def __init__(self, host):
super(LibvirtAOEVolumeDriver,
- self).__init__(connection, is_block_dev=True)
+ self).__init__(host, is_block_dev=True)
# Call the factory here so we can support
# more than x86 architectures.
Uses the DISCO connector from the os-brick projects.
"""
- def __init__(self, connection):
+ def __init__(self, host):
"""Init DISCO connector for LibVirt."""
- super(LibvirtDISCOVolumeDriver, self).__init__(connection,
+ super(LibvirtDISCOVolumeDriver, self).__init__(host,
is_block_dev=False)
self.connector = connector.InitiatorConnector.factory(
'DISCO', utils.get_root_helper(),
class LibvirtFibreChannelVolumeDriver(libvirt_volume.LibvirtBaseVolumeDriver):
"""Driver to attach Fibre Channel Network volumes to libvirt."""
- def __init__(self, connection):
+ def __init__(self, host):
super(LibvirtFibreChannelVolumeDriver,
- self).__init__(connection, is_block_dev=False)
+ self).__init__(host, is_block_dev=False)
# Call the factory here so we can support
# more than x86 architectures.
libvirt_volume.LibvirtBaseVolumeDriver):
"""The base class for file system type volume drivers"""
- def __init__(self, connection):
+ def __init__(self, host):
super(LibvirtBaseFileSystemVolumeDriver,
- self).__init__(connection, is_block_dev=False)
+ self).__init__(host, is_block_dev=False)
@abc.abstractmethod
def _get_mount_point_base(self):
class LibvirtHGSTVolumeDriver(libvirt_volume.LibvirtBaseVolumeDriver):
"""Driver to attach HGST volumes to libvirt."""
- def __init__(self, connection):
+ def __init__(self, host):
super(LibvirtHGSTVolumeDriver,
- self).__init__(connection, is_block_dev=True)
+ self).__init__(host, is_block_dev=True)
self.connector = connector.InitiatorConnector.factory(
'HGST', utils.get_root_helper(),
device_scan_attempts=CONF.libvirt.num_iscsi_scan_tries)
class LibvirtISCSIVolumeDriver(libvirt_volume.LibvirtBaseVolumeDriver):
"""Driver to attach Network volumes to libvirt."""
- def __init__(self, connection):
- super(LibvirtISCSIVolumeDriver, self).__init__(connection,
+ def __init__(self, host):
+ super(LibvirtISCSIVolumeDriver, self).__init__(host,
is_block_dev=True)
# Call the factory here so we can support
class LibvirtNetVolumeDriver(libvirt_volume.LibvirtBaseVolumeDriver):
"""Driver to attach Network volumes to libvirt."""
- def __init__(self, connection):
+ def __init__(self, host):
super(LibvirtNetVolumeDriver,
- self).__init__(connection, is_block_dev=False)
+ self).__init__(host, is_block_dev=False)
def _get_secret_uuid(self, conf, password=None):
# TODO(mriedem): Add delegation methods to connection (LibvirtDriver)
# to call through for these secret CRUD operations so the volume driver
# doesn't need to know the internal attributes of the connection
# object.
- secret = self.connection._host.find_secret(conf.source_protocol,
+ secret = self.host.find_secret(conf.source_protocol,
conf.source_name)
if secret is None:
- secret = self.connection._host.create_secret(conf.source_protocol,
+ secret = self.host.create_secret(conf.source_protocol,
conf.source_name,
password)
return secret.UUIDString()
usage_type = 'iscsi'
usage_name = ("%(target_iqn)s/%(target_lun)s" %
netdisk_properties)
- self.connection._host.delete_secret(usage_type, usage_name)
+ self.host.delete_secret(usage_type, usage_name)
def get_config(self, connection_info, disk_info):
"""Returns xml for libvirt."""
Implements Libvirt part of volume driver for ScaleIO cinder driver.
Uses the ScaleIO connector from the os-brick projects
"""
- def __init__(self, connection):
- super(LibvirtScaleIOVolumeDriver, self).__init__(connection,
+ def __init__(self, host):
+ super(LibvirtScaleIOVolumeDriver, self).__init__(host,
is_block_dev=False)
self.connector = connector.InitiatorConnector.factory(
'SCALEIO', utils.get_root_helper(),
class LibvirtBaseVolumeDriver(object):
"""Base class for volume drivers."""
- def __init__(self, connection, is_block_dev):
- self.connection = connection
+ def __init__(self, host, is_block_dev):
+ self.host = host
self.is_block_dev = is_block_dev
def get_config(self, connection_info, disk_info):
"""Returns xml for libvirt."""
conf = vconfig.LibvirtConfigGuestDisk()
conf.driver_name = libvirt_utils.pick_disk_driver_name(
- self.connection._host.get_version(),
+ self.host.get_version(),
self.is_block_dev
)
# Configure usage of discard
if data.get('discard', False) is True:
min_qemu = nova.virt.libvirt.driver.MIN_QEMU_DISCARD_VERSION
- if self.connection._host.has_min_version(
+ if self.host.has_min_version(
hv_ver=min_qemu,
hv_type=host.HV_DRIVER_QEMU):
conf.driver_discard = 'unmap'
class LibvirtVolumeDriver(LibvirtBaseVolumeDriver):
"""Class for volumes backed by local file."""
- def __init__(self, connection):
+ def __init__(self, host):
super(LibvirtVolumeDriver,
- self).__init__(connection, is_block_dev=True)
+ self).__init__(host, is_block_dev=True)
def get_config(self, connection_info, disk_info):
"""Returns xml for libvirt."""
class LibvirtFakeVolumeDriver(LibvirtBaseVolumeDriver):
"""Driver to attach fake volumes to libvirt."""
- def __init__(self, connection):
+ def __init__(self, host):
super(LibvirtFakeVolumeDriver,
- self).__init__(connection, is_block_dev=True)
+ self).__init__(host, is_block_dev=True)
def get_config(self, connection_info, disk_info):
"""Returns xml for libvirt."""
projects / osstest / openstack-nova.git / commitdiff