<s> """<STR_LIT>""" <EOL> import mock <EOL> import os <EOL> from oslo_concurrency import processutils <EOL> from oslo_utils import fileutils <EOL> from nova import exception <EOL> from nova import test <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 quobyte <EOL> class QuobyteTestCase ( test . NoDBTestCase ) : <EOL> """<STR_LIT>""" <EOL> @ mock . patch . object ( fileutils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( utils , "<STR_LIT>" ) <EOL> def test_quobyte_mount_volume ( self , mock_execute , mock_ensure_tree ) : <EOL> mnt_base = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <EOL> quobyte . mount_volume ( quobyte_volume , export_mnt_base ) <EOL> mock_ensure_tree . assert_called_once_with ( export_mnt_base ) <EOL> expected_commands = [ mock . call ( '<STR_LIT>' , <EOL> quobyte_volume , <EOL> export_mnt_base , <EOL> check_exit_code = [ <NUM_LIT:0> , <NUM_LIT:4> ] ) <EOL> ] <EOL> mock_execute . assert_has_calls ( expected_commands ) <EOL> @ mock . patch . object ( fileutils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( utils , "<STR_LIT>" ) <EOL> def test_quobyte_mount_volume_with_config ( self , <EOL> mock_execute , <EOL> mock_ensure_tree ) : <EOL> mnt_base = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <EOL> config_file_dummy = "<STR_LIT>" <EOL> quobyte . mount_volume ( quobyte_volume , <EOL> export_mnt_base , <EOL> config_file_dummy ) <EOL> mock_ensure_tree . assert_called_once_with ( export_mnt_base ) <EOL> expected_commands = [ mock . call ( '<STR_LIT>' , <EOL> quobyte_volume , <EOL> export_mnt_base , <EOL> '<STR_LIT:-c>' , <EOL> config_file_dummy , <EOL> check_exit_code = [ <NUM_LIT:0> , <NUM_LIT:4> ] ) <EOL> ] <EOL> mock_execute . assert_has_calls ( expected_commands ) <EOL> @ mock . patch . object ( fileutils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( utils , "<STR_LIT>" , <EOL> side_effect = ( processutils . <EOL> ProcessExecutionError ) ) <EOL> def test_quobyte_mount_volume_fails ( self , mock_execute , mock_ensure_tree ) : <EOL> mnt_base = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <EOL> self . assertRaises ( processutils . ProcessExecutionError , <EOL> quobyte . mount_volume , <EOL> quobyte_volume , <EOL> export_mnt_base ) <EOL> @ mock . patch . object ( utils , "<STR_LIT>" ) <EOL> def test_quobyte_umount_volume ( self , mock_execute ) : <EOL> mnt_base = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <EOL> quobyte . umount_volume ( export_mnt_base ) <EOL> mock_execute . assert_called_once_with ( '<STR_LIT>' , <EOL> export_mnt_base ) <EOL> @ mock . patch . object ( quobyte . LOG , "<STR_LIT:error>" ) <EOL> @ mock . patch . object ( utils , "<STR_LIT>" ) <EOL> def test_quobyte_umount_volume_warns ( self , <EOL> mock_execute , <EOL> mock_debug ) : <EOL> mnt_base = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <EOL> def exec_side_effect ( * cmd , ** kwargs ) : <EOL> exerror = processutils . ProcessExecutionError ( <EOL> "<STR_LIT>" ) <EOL> raise exerror <EOL> mock_execute . side_effect = exec_side_effect <EOL> quobyte . umount_volume ( export_mnt_base ) <EOL> ( mock_debug . <EOL> assert_called_once_with ( "<STR_LIT>" , <EOL> export_mnt_base ) ) <EOL> @ mock . patch . object ( quobyte . LOG , "<STR_LIT>" ) <EOL> @ mock . patch . object ( utils , "<STR_LIT>" , <EOL> side_effect = ( processutils . ProcessExecutionError ) ) <EOL> def test_quobyte_umount_volume_fails ( self , <EOL> mock_execute , <EOL> mock_exception ) : <EOL> mnt_base = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <EOL> quobyte . umount_volume ( export_mnt_base ) <EOL> ( mock_exception . <EOL> assert_called_once_with ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> export_mnt_base ) ) <EOL> @ mock . patch . object ( os , "<STR_LIT>" , return_value = True ) <EOL> @ mock . patch . object ( utils , "<STR_LIT>" ) <EOL> def test_quobyte_is_valid_volume ( self , mock_execute , mock_access ) : <EOL> mnt_base = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <EOL> quobyte . validate_volume ( export_mnt_base ) <EOL> mock_execute . assert_called_once_with ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> export_mnt_base ) <EOL> @ mock . patch . object ( utils , "<STR_LIT>" , <EOL> side_effect = ( processutils . <EOL> ProcessExecutionError ) ) <EOL> def test_quobyte_is_valid_volume_vol_not_valid_volume ( self , mock_execute ) : <EOL> mnt_base = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <EOL> self . assertRaises ( exception . NovaException , <EOL> quobyte . validate_volume , <EOL> export_mnt_base ) <EOL> @ mock . patch . object ( os , "<STR_LIT>" , return_value = False ) <EOL> @ mock . patch . object ( utils , "<STR_LIT>" , <EOL> side_effect = ( processutils . <EOL> ProcessExecutionError ) ) <EOL> def test_quobyte_is_valid_volume_vol_no_valid_access ( self , <EOL> mock_execute , <EOL> mock_access ) : <EOL> mnt_base = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <EOL> self . assertRaises ( exception . NovaException , <EOL> quobyte . validate_volume , <EOL> export_mnt_base ) <EOL> class LibvirtQuobyteVolumeDriverTestCase ( <EOL> test_volume . LibvirtVolumeBaseTestCase ) : <EOL> """<STR_LIT>""" <EOL> @ mock . patch . object ( quobyte , '<STR_LIT>' ) <EOL> @ mock . patch . object ( quobyte , '<STR_LIT>' ) <EOL> @ mock . patch . object ( libvirt_utils , '<STR_LIT>' , return_value = False ) <EOL> def test_libvirt_quobyte_driver_mount ( self , <EOL> mock_is_mounted , <EOL> mock_mount_volume , <EOL> mock_validate_volume <EOL> ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( quobyte_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = quobyte . LibvirtQuobyteVolumeDriver ( self . fake_conn ) <EOL> export_string = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <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> libvirt_driver . connect_volume ( connection_info , self . disk_info ) <EOL> conf = libvirt_driver . get_config ( connection_info , self . disk_info ) <EOL> tree = conf . format_dom ( ) <EOL> self . _assertFileTypeEquals ( tree , file_path ) <EOL> mock_mount_volume . assert_called_once_with ( quobyte_volume , <EOL> export_mnt_base , <EOL> mock . ANY ) <EOL> mock_validate_volume . assert_called_with ( export_mnt_base ) <EOL> @ mock . patch . object ( quobyte , '<STR_LIT>' ) <EOL> @ mock . patch . object ( quobyte , '<STR_LIT>' ) <EOL> @ mock . patch . object ( libvirt_utils , '<STR_LIT>' , return_value = True ) <EOL> def test_libvirt_quobyte_driver_umount ( self , mock_is_mounted , <EOL> mock_umount_volume , <EOL> mock_validate_volume ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( quobyte_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = quobyte . LibvirtQuobyteVolumeDriver ( self . fake_conn ) <EOL> export_string = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <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> libvirt_driver . connect_volume ( connection_info , self . disk_info ) <EOL> conf = libvirt_driver . get_config ( connection_info , self . disk_info ) <EOL> tree = conf . format_dom ( ) <EOL> self . _assertFileTypeEquals ( tree , file_path ) <EOL> libvirt_driver . disconnect_volume ( connection_info , "<STR_LIT>" ) <EOL> mock_validate_volume . assert_called_once_with ( export_mnt_base ) <EOL> mock_umount_volume . assert_called_once_with ( export_mnt_base ) <EOL> @ mock . patch . object ( quobyte , '<STR_LIT>' ) <EOL> @ mock . patch . object ( quobyte , '<STR_LIT>' ) <EOL> def test_libvirt_quobyte_driver_already_mounted ( self , <EOL> mock_umount_volume , <EOL> mock_validate_volume <EOL> ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( quobyte_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = quobyte . LibvirtQuobyteVolumeDriver ( self . fake_conn ) <EOL> export_string = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <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> libvirt_driver . connect_volume ( connection_info , self . disk_info ) <EOL> conf = libvirt_driver . get_config ( connection_info , self . disk_info ) <EOL> tree = conf . format_dom ( ) <EOL> self . _assertFileTypeEquals ( tree , file_path ) <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>" + quobyte_volume ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , export_mnt_base , <EOL> '<STR_LIT>' , "<STR_LIT>" + quobyte_volume ) , <EOL> ] <EOL> self . assertEqual ( expected_commands , self . executes ) <EOL> mock_umount_volume . assert_called_once_with ( export_mnt_base ) <EOL> mock_validate_volume . assert_called_once_with ( export_mnt_base ) <EOL> @ mock . patch . object ( quobyte , '<STR_LIT>' ) <EOL> @ mock . patch . object ( quobyte , '<STR_LIT>' ) <EOL> @ mock . patch . object ( libvirt_utils , '<STR_LIT>' , return_value = False ) <EOL> def test_libvirt_quobyte_driver_qcow2 ( self , mock_is_mounted , <EOL> mock_mount_volume , <EOL> mock_validate_volume <EOL> ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( quobyte_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = quobyte . LibvirtQuobyteVolumeDriver ( self . fake_conn ) <EOL> export_string = '<STR_LIT>' <EOL> name = '<STR_LIT>' <EOL> image_format = '<STR_LIT>' <EOL> quobyte_volume = '<STR_LIT>' <EOL> connection_info = { '<STR_LIT:data>' : { '<STR_LIT>' : export_string , <EOL> '<STR_LIT:name>' : name , <EOL> '<STR_LIT>' : image_format } } <EOL> export_mnt_base = os . path . join ( mnt_base , <EOL> utils . get_hash_str ( quobyte_volume ) ) <EOL> libvirt_driver . connect_volume ( connection_info , self . disk_info ) <EOL> conf = libvirt_driver . get_config ( connection_info , self . disk_info ) <EOL> tree = conf . format_dom ( ) <EOL> self . assertEqual ( '<STR_LIT:file>' , tree . get ( '<STR_LIT:type>' ) ) <EOL> self . assertEqual ( '<STR_LIT>' , tree . find ( '<STR_LIT>' ) . get ( '<STR_LIT:type>' ) ) <EOL> ( mock_mount_volume . <EOL> assert_called_once_with ( '<STR_LIT>' , <EOL> export_mnt_base , <EOL> mock . ANY ) ) <EOL> mock_validate_volume . assert_called_with ( export_mnt_base ) <EOL> libvirt_driver . disconnect_volume ( connection_info , "<STR_LIT>" ) <EOL> def test_libvirt_quobyte_driver_mount_non_quobyte_volume ( self ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( quobyte_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = quobyte . LibvirtQuobyteVolumeDriver ( self . fake_conn ) <EOL> export_string = '<STR_LIT>' <EOL> connection_info = { '<STR_LIT:data>' : { '<STR_LIT>' : export_string , <EOL> '<STR_LIT:name>' : self . name } } <EOL> def exe_side_effect ( * cmd , ** kwargs ) : <EOL> if cmd == mock . ANY : <EOL> raise exception . NovaException ( ) <EOL> with mock . patch . object ( quobyte , <EOL> '<STR_LIT>' ) as mock_execute : <EOL> mock_execute . side_effect = exe_side_effect <EOL> self . assertRaises ( exception . NovaException , <EOL> libvirt_driver . connect_volume , <EOL> connection_info , <EOL> self . disk_info ) <EOL> def test_libvirt_quobyte_driver_normalize_export_with_protocol ( self ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( quobyte_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = quobyte . LibvirtQuobyteVolumeDriver ( self . fake_conn ) <EOL> export_string = '<STR_LIT>' <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> libvirt_driver . _normalize_export ( export_string ) ) <EOL> def test_libvirt_quobyte_driver_normalize_export_without_protocol ( self ) : <EOL> mnt_base = '<STR_LIT>' <EOL> self . flags ( quobyte_mount_point_base = mnt_base , group = '<STR_LIT>' ) <EOL> libvirt_driver = quobyte . LibvirtQuobyteVolumeDriver ( self . fake_conn ) <EOL> export_string = '<STR_LIT>' <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> libvirt_driver . _normalize_export ( export_string ) ) </s>
