<s> import os <EOL> import mock <EOL> from oslo_concurrency import processutils <EOL> from nova . tests . unit . virt . libvirt . volume import test_volume <EOL> from nova import utils <EOL> from nova . virt . libvirt import utils as libvirt_utils <EOL> from nova . virt . libvirt . volume import glusterfs <EOL> class LibvirtGlusterfsVolumeDriverTestCase ( <EOL> test_volume . LibvirtVolumeBaseTestCase ) : <EOL> def test_libvirt_glusterfs_driver ( self ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( glusterfs_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = glusterfs . LibvirtGlusterfsVolumeDriver ( self . fake_conn ) <EOL> self . stubs . Set ( libvirt_utils , '<STR_LIT>' , lambda x , d : False ) <EOL> export_string = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( export_string ) ) <EOL> connection_info = { '<STR_LIT:data>' : { '<STR_LIT>' : export_string , <EOL> '<STR_LIT:name>' : self . name } } <EOL> libvirt_driver . connect_volume ( connection_info , self . disk_info ) <EOL> libvirt_driver . disconnect_volume ( connection_info , "<STR_LIT>" ) <EOL> device_path = os . path . join ( export_mnt_base , <EOL> connection_info [ '<STR_LIT:data>' ] [ '<STR_LIT:name>' ] ) <EOL> self . assertEqual ( connection_info [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] , device_path ) <EOL> expected_commands = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , export_mnt_base ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , export_string , export_mnt_base ) , <EOL> ( '<STR_LIT>' , export_mnt_base ) ] <EOL> self . assertEqual ( expected_commands , self . executes ) <EOL> def test_libvirt_glusterfs_driver_get_config ( self ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( glusterfs_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = glusterfs . LibvirtGlusterfsVolumeDriver ( self . fake_conn ) <EOL> export_string = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( export_string ) ) <EOL> file_path = os . path . join ( export_mnt_base , self . name ) <EOL> connection_info = { '<STR_LIT:data>' : { '<STR_LIT>' : export_string , <EOL> '<STR_LIT:name>' : self . name , <EOL> '<STR_LIT>' : file_path } } <EOL> conf = libvirt_driver . get_config ( connection_info , self . disk_info ) <EOL> tree = conf . format_dom ( ) <EOL> self . _assertFileTypeEquals ( tree , file_path ) <EOL> self . assertEqual ( '<STR_LIT>' , tree . find ( '<STR_LIT>' ) . get ( '<STR_LIT:type>' ) ) <EOL> connection_info = { '<STR_LIT:data>' : { '<STR_LIT>' : export_string , <EOL> '<STR_LIT:name>' : self . name , <EOL> '<STR_LIT>' : file_path , <EOL> '<STR_LIT>' : '<STR_LIT>' } } <EOL> conf = libvirt_driver . get_config ( connection_info , self . disk_info ) <EOL> tree = conf . format_dom ( ) <EOL> self . _assertFileTypeEquals ( tree , file_path ) <EOL> self . assertEqual ( '<STR_LIT>' , tree . find ( '<STR_LIT>' ) . get ( '<STR_LIT:type>' ) ) <EOL> def test_libvirt_glusterfs_driver_already_mounted ( self ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( glusterfs_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = glusterfs . LibvirtGlusterfsVolumeDriver ( self . fake_conn ) <EOL> export_string = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( export_string ) ) <EOL> connection_info = { '<STR_LIT:data>' : { '<STR_LIT>' : export_string , <EOL> '<STR_LIT:name>' : self . name } } <EOL> libvirt_driver . connect_volume ( connection_info , self . disk_info ) <EOL> libvirt_driver . disconnect_volume ( connection_info , "<STR_LIT>" ) <EOL> expected_commands = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , export_mnt_base , <EOL> '<STR_LIT>' , export_string ) , <EOL> ( '<STR_LIT>' , export_mnt_base ) ] <EOL> self . assertEqual ( expected_commands , self . executes ) <EOL> @ mock . patch . object ( glusterfs . utils , '<STR_LIT>' ) <EOL> @ mock . patch . object ( glusterfs . LOG , '<STR_LIT>' ) <EOL> @ mock . patch . object ( glusterfs . LOG , '<STR_LIT>' ) <EOL> def test_libvirt_glusterfs_driver_umount_error ( self , mock_LOG_exception , <EOL> mock_LOG_debug , mock_utils_exe ) : <EOL> export_string = '<STR_LIT>' <EOL> connection_info = { '<STR_LIT:data>' : { '<STR_LIT>' : export_string , <EOL> '<STR_LIT:name>' : self . name } } <EOL> libvirt_driver = glusterfs . LibvirtGlusterfsVolumeDriver ( self . fake_conn ) <EOL> mock_utils_exe . side_effect = processutils . ProcessExecutionError ( <EOL> None , None , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> libvirt_driver . disconnect_volume ( connection_info , "<STR_LIT>" ) <EOL> self . assertTrue ( mock_LOG_debug . called ) <EOL> def test_libvirt_glusterfs_driver_with_opts ( self ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( glusterfs_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = glusterfs . LibvirtGlusterfsVolumeDriver ( self . fake_conn ) <EOL> self . stubs . Set ( libvirt_utils , '<STR_LIT>' , lambda x , d : False ) <EOL> export_string = '<STR_LIT>' <EOL> options = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( export_string ) ) <EOL> connection_info = { '<STR_LIT:data>' : { '<STR_LIT>' : export_string , <EOL> '<STR_LIT:name>' : self . name , <EOL> '<STR_LIT>' : options } } <EOL> libvirt_driver . connect_volume ( connection_info , self . disk_info ) <EOL> libvirt_driver . disconnect_volume ( connection_info , "<STR_LIT>" ) <EOL> expected_commands = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , export_mnt_base ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> export_string , export_mnt_base ) , <EOL> ( '<STR_LIT>' , export_mnt_base ) , <EOL> ] <EOL> self . assertEqual ( expected_commands , self . executes ) <EOL> def test_libvirt_glusterfs_libgfapi ( self ) : <EOL> self . flags ( qemu_allowed_storage_drivers = [ '<STR_LIT>' ] , group = '<STR_LIT>' ) <EOL> libvirt_driver = glusterfs . LibvirtGlusterfsVolumeDriver ( self . fake_conn ) <EOL> self . stubs . Set ( libvirt_utils , '<STR_LIT>' , lambda x , d : False ) <EOL> export_string = '<STR_LIT>' <EOL> name = '<STR_LIT>' <EOL> connection_info = { '<STR_LIT:data>' : { '<STR_LIT>' : export_string , '<STR_LIT:name>' : name } } <EOL> disk_info = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> libvirt_driver . connect_volume ( connection_info , disk_info ) <EOL> conf = libvirt_driver . get_config ( connection_info , disk_info ) <EOL> tree = conf . format_dom ( ) <EOL> self . assertEqual ( '<STR_LIT>' , tree . get ( '<STR_LIT:type>' ) ) <EOL> self . assertEqual ( '<STR_LIT>' , tree . find ( '<STR_LIT>' ) . get ( '<STR_LIT:type>' ) ) <EOL> source = tree . find ( '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , source . get ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( '<STR_LIT>' , source . get ( '<STR_LIT:name>' ) ) <EOL> self . assertEqual ( '<STR_LIT>' , source . find ( '<STR_LIT>' ) . get ( '<STR_LIT:name>' ) ) <EOL> self . assertEqual ( '<STR_LIT>' , source . find ( '<STR_LIT>' ) . get ( '<STR_LIT:port>' ) ) <EOL> libvirt_driver . disconnect_volume ( connection_info , "<STR_LIT>" ) </s>
