<s> import mock <EOL> from nova import exception <EOL> from nova import test <EOL> from nova . tests . unit . virt . xenapi import stubs <EOL> from nova . virt . xenapi import vm_utils <EOL> from nova . virt . xenapi import volume_utils <EOL> from nova . virt . xenapi import volumeops <EOL> class VolumeOpsTestBase ( stubs . XenAPITestBaseNoDB ) : <EOL> def setUp ( self ) : <EOL> super ( VolumeOpsTestBase , self ) . setUp ( ) <EOL> self . _setup_mock_volumeops ( ) <EOL> def _setup_mock_volumeops ( self ) : <EOL> self . session = stubs . FakeSessionForVolumeTests ( '<STR_LIT>' ) <EOL> self . ops = volumeops . VolumeOps ( self . session ) <EOL> class VolumeDetachTestCase ( VolumeOpsTestBase ) : <EOL> def test_detach_volume_call ( self ) : <EOL> registered_calls = [ ] <EOL> def regcall ( label ) : <EOL> def side_effect ( * args , ** kwargs ) : <EOL> registered_calls . append ( label ) <EOL> return side_effect <EOL> ops = volumeops . VolumeOps ( '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( volumeops . vm_utils , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( volumeops . volume_utils , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( volumeops . vm_utils , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( volumeops . vm_utils , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( volumeops . vm_utils , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( volumeops . volume_utils , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( volumeops . volume_utils , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( volumeops . volume_utils , '<STR_LIT>' ) <EOL> volumeops . vm_utils . lookup ( '<STR_LIT>' , '<STR_LIT>' ) . AndReturn ( <EOL> '<STR_LIT>' ) <EOL> volumeops . volume_utils . get_device_number ( '<STR_LIT>' ) . AndReturn ( <EOL> '<STR_LIT>' ) <EOL> volumeops . volume_utils . find_vbd_by_number ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) . AndReturn ( '<STR_LIT>' ) <EOL> volumeops . vm_utils . is_vm_shutdown ( '<STR_LIT>' , '<STR_LIT>' ) . AndReturn ( <EOL> False ) <EOL> volumeops . vm_utils . unplug_vbd ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> volumeops . vm_utils . destroy_vbd ( '<STR_LIT>' , '<STR_LIT>' ) . WithSideEffects ( <EOL> regcall ( '<STR_LIT>' ) ) <EOL> volumeops . volume_utils . find_sr_from_vbd ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) . WithSideEffects ( <EOL> regcall ( '<STR_LIT>' ) ) . AndReturn ( '<STR_LIT>' ) <EOL> volumeops . volume_utils . purge_sr ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> ops . detach_volume ( <EOL> dict ( driver_volume_type = '<STR_LIT>' , data = '<STR_LIT>' ) , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , registered_calls ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> def test_detach_volume ( self , mock_vm , mock_vbd , mock_detach ) : <EOL> mock_vm . return_value = "<STR_LIT>" <EOL> mock_vbd . return_value = "<STR_LIT>" <EOL> self . ops . detach_volume ( { } , "<STR_LIT:name>" , "<STR_LIT>" ) <EOL> mock_vm . assert_called_once_with ( self . session , "<STR_LIT:name>" ) <EOL> mock_vbd . assert_called_once_with ( self . session , "<STR_LIT>" , <NUM_LIT:3> ) <EOL> mock_detach . assert_called_once_with ( "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> def test_detach_volume_skips_error_skip_attach ( self , mock_vm , mock_vbd , <EOL> mock_detach ) : <EOL> mock_vm . return_value = "<STR_LIT>" <EOL> mock_vbd . return_value = None <EOL> self . ops . detach_volume ( { } , "<STR_LIT:name>" , "<STR_LIT>" ) <EOL> self . assertFalse ( mock_detach . called ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> def test_detach_volume_raises ( self , mock_vm , mock_vbd , <EOL> mock_detach ) : <EOL> mock_vm . return_value = "<STR_LIT>" <EOL> mock_vbd . side_effect = test . TestingException <EOL> self . assertRaises ( test . TestingException , <EOL> self . ops . detach_volume , { } , "<STR_LIT:name>" , "<STR_LIT>" ) <EOL> self . assertFalse ( mock_detach . called ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> def test_detach_vbds_and_srs_not_shutdown ( self , mock_shutdown , mock_unplug , <EOL> mock_find_sr , mock_destroy , mock_purge ) : <EOL> mock_shutdown . return_value = False <EOL> mock_find_sr . return_value = "<STR_LIT>" <EOL> self . ops . _detach_vbds_and_srs ( "<STR_LIT>" , [ "<STR_LIT>" ] ) <EOL> mock_shutdown . assert_called_once_with ( self . session , "<STR_LIT>" ) <EOL> mock_find_sr . assert_called_once_with ( self . session , "<STR_LIT>" ) <EOL> mock_unplug . assert_called_once_with ( self . session , "<STR_LIT>" , "<STR_LIT>" ) <EOL> mock_destroy . assert_called_once_with ( self . session , "<STR_LIT>" ) <EOL> mock_purge . assert_called_once_with ( self . session , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> def test_detach_vbds_and_srs_is_shutdown ( self , mock_shutdown , mock_unplug , <EOL> mock_find_sr , mock_destroy , mock_purge ) : <EOL> mock_shutdown . return_value = True <EOL> mock_find_sr . return_value = "<STR_LIT>" <EOL> self . ops . _detach_vbds_and_srs ( "<STR_LIT>" , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> expected = [ mock . call ( self . session , "<STR_LIT>" ) , <EOL> mock . call ( self . session , "<STR_LIT>" ) ] <EOL> self . assertEqual ( expected , mock_destroy . call_args_list ) <EOL> mock_purge . assert_called_with ( self . session , "<STR_LIT>" ) <EOL> self . assertFalse ( mock_unplug . called ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> def test_detach_all_no_volumes ( self , mock_get_all , mock_detach ) : <EOL> mock_get_all . return_value = [ ] <EOL> self . ops . detach_all ( "<STR_LIT>" ) <EOL> mock_get_all . assert_called_once_with ( "<STR_LIT>" ) <EOL> self . assertFalse ( mock_detach . called ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> def test_detach_all_volumes ( self , mock_get_all , mock_detach ) : <EOL> mock_get_all . return_value = [ "<STR_LIT:1>" ] <EOL> self . ops . detach_all ( "<STR_LIT>" ) <EOL> mock_get_all . assert_called_once_with ( "<STR_LIT>" ) <EOL> mock_detach . assert_called_once_with ( "<STR_LIT>" , [ "<STR_LIT:1>" ] ) <EOL> def test_get_all_volume_vbd_refs_no_vbds ( self ) : <EOL> with mock . patch . object ( self . session . VM , "<STR_LIT>" ) as mock_get : <EOL> with mock . patch . object ( self . session . VBD , <EOL> "<STR_LIT>" ) as mock_conf : <EOL> mock_get . return_value = [ ] <EOL> result = self . ops . _get_all_volume_vbd_refs ( "<STR_LIT>" ) <EOL> self . assertEqual ( [ ] , list ( result ) ) <EOL> mock_get . assert_called_once_with ( "<STR_LIT>" ) <EOL> self . assertFalse ( mock_conf . called ) <EOL> def test_get_all_volume_vbd_refs_no_volumes ( self ) : <EOL> with mock . patch . object ( self . session . VM , "<STR_LIT>" ) as mock_get : <EOL> with mock . patch . object ( self . session . VBD , <EOL> "<STR_LIT>" ) as mock_conf : <EOL> mock_get . return_value = [ "<STR_LIT:1>" ] <EOL> mock_conf . return_value = { } <EOL> result = self . ops . _get_all_volume_vbd_refs ( "<STR_LIT>" ) <EOL> self . assertEqual ( [ ] , list ( result ) ) <EOL> mock_get . assert_called_once_with ( "<STR_LIT>" ) <EOL> mock_conf . assert_called_once_with ( "<STR_LIT:1>" ) <EOL> def test_get_all_volume_vbd_refs_with_volumes ( self ) : <EOL> with mock . patch . object ( self . session . VM , "<STR_LIT>" ) as mock_get : <EOL> with mock . patch . object ( self . session . VBD , <EOL> "<STR_LIT>" ) as mock_conf : <EOL> mock_get . return_value = [ "<STR_LIT:1>" , "<STR_LIT:2>" ] <EOL> mock_conf . return_value = { "<STR_LIT>" : True } <EOL> result = self . ops . _get_all_volume_vbd_refs ( "<STR_LIT>" ) <EOL> self . assertEqual ( [ "<STR_LIT:1>" , "<STR_LIT:2>" ] , list ( result ) ) <EOL> mock_get . assert_called_once_with ( "<STR_LIT>" ) <EOL> class AttachVolumeTestCase ( VolumeOpsTestBase ) : <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> def test_attach_volume_default_hotplug ( self , mock_get_vm , mock_attach ) : <EOL> mock_get_vm . return_value = "<STR_LIT>" <EOL> self . ops . attach_volume ( { } , "<STR_LIT>" , "<STR_LIT>" ) <EOL> mock_attach . assert_called_once_with ( { } , "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:0> , <EOL> True ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> def test_attach_volume_hotplug ( self , mock_get_vm , mock_attach ) : <EOL> mock_get_vm . return_value = "<STR_LIT>" <EOL> self . ops . attach_volume ( { } , "<STR_LIT>" , "<STR_LIT>" , False ) <EOL> mock_attach . assert_called_once_with ( { } , "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:0> , <EOL> False ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> def test_attach_volume_default_hotplug_connect_volume ( self , mock_attach ) : <EOL> self . ops . connect_volume ( { } ) <EOL> mock_attach . assert_called_once_with ( { } ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> def test_attach_volume_with_defaults ( self , mock_attach , mock_hypervisor , <EOL> mock_provider , mock_driver ) : <EOL> connection_info = { "<STR_LIT:data>" : { } } <EOL> with mock . patch . object ( self . session . VDI , "<STR_LIT>" ) as mock_vdi : <EOL> mock_provider . return_value = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> mock_vdi . return_value = "<STR_LIT>" <EOL> result = self . ops . _attach_volume ( connection_info ) <EOL> self . assertEqual ( result , ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> mock_driver . assert_called_once_with ( connection_info ) <EOL> mock_provider . assert_called_once_with ( { } , None ) <EOL> mock_hypervisor . assert_called_once_with ( "<STR_LIT>" , { } ) <EOL> self . assertFalse ( mock_attach . called ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> def test_attach_volume_with_hot_attach ( self , mock_attach , mock_hypervisor , <EOL> mock_provider , mock_driver ) : <EOL> connection_info = { "<STR_LIT:data>" : { } } <EOL> with mock . patch . object ( self . session . VDI , "<STR_LIT>" ) as mock_vdi : <EOL> mock_provider . return_value = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> mock_hypervisor . return_value = "<STR_LIT>" <EOL> mock_vdi . return_value = "<STR_LIT>" <EOL> result = self . ops . _attach_volume ( connection_info , "<STR_LIT>" , <EOL> "<STR_LIT:name>" , <NUM_LIT:2> , True ) <EOL> self . assertEqual ( result , ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> mock_driver . assert_called_once_with ( connection_info ) <EOL> mock_provider . assert_called_once_with ( { } , "<STR_LIT:name>" ) <EOL> mock_hypervisor . assert_called_once_with ( "<STR_LIT>" , { } ) <EOL> mock_attach . assert_called_once_with ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:name>" , <NUM_LIT:2> , <EOL> True ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> def test_attach_volume_cleanup ( self , mock_attach , mock_hypervisor , <EOL> mock_provider , mock_driver , mock_forget ) : <EOL> connection_info = { "<STR_LIT:data>" : { } } <EOL> mock_provider . return_value = ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> mock_hypervisor . side_effect = test . TestingException <EOL> self . assertRaises ( test . TestingException , <EOL> self . ops . _attach_volume , connection_info ) <EOL> mock_driver . assert_called_once_with ( connection_info ) <EOL> mock_provider . assert_called_once_with ( { } , None ) <EOL> mock_hypervisor . assert_called_once_with ( "<STR_LIT>" , { } ) <EOL> mock_forget . assert_called_once_with ( self . session , "<STR_LIT>" ) <EOL> self . assertFalse ( mock_attach . called ) <EOL> def test_check_is_supported_driver_type_pass_iscsi ( self ) : <EOL> conn_info = { "<STR_LIT>" : "<STR_LIT>" } <EOL> self . ops . _check_is_supported_driver_type ( conn_info ) <EOL> def test_check_is_supported_driver_type_pass_xensm ( self ) : <EOL> conn_info = { "<STR_LIT>" : "<STR_LIT>" } <EOL> self . ops . _check_is_supported_driver_type ( conn_info ) <EOL> def test_check_is_supported_driver_type_pass_bad ( self ) : <EOL> conn_info = { "<STR_LIT>" : "<STR_LIT>" } <EOL> self . assertRaises ( exception . VolumeDriverNotFound , <EOL> self . ops . _check_is_supported_driver_type , conn_info ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> def test_connect_to_volume_provider_new_sr ( self , mock_parse , mock_find_sr , <EOL> mock_introduce_sr ) : <EOL> mock_parse . return_value = ( "<STR_LIT>" , "<STR_LIT:label>" , "<STR_LIT>" ) <EOL> mock_find_sr . return_value = None <EOL> mock_introduce_sr . return_value = "<STR_LIT>" <EOL> ref , uuid = self . ops . _connect_to_volume_provider ( { } , "<STR_LIT:name>" ) <EOL> self . assertEqual ( "<STR_LIT>" , ref ) <EOL> self . assertEqual ( "<STR_LIT>" , uuid ) <EOL> mock_parse . assert_called_once_with ( { } , "<STR_LIT>" ) <EOL> mock_find_sr . assert_called_once_with ( self . session , "<STR_LIT>" ) <EOL> mock_introduce_sr . assert_called_once_with ( self . session , "<STR_LIT>" , <EOL> "<STR_LIT:label>" , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> def test_connect_to_volume_provider_old_sr ( self , mock_parse , mock_find_sr , <EOL> mock_introduce_sr ) : <EOL> mock_parse . return_value = ( "<STR_LIT>" , "<STR_LIT:label>" , "<STR_LIT>" ) <EOL> mock_find_sr . return_value = "<STR_LIT>" <EOL> ref , uuid = self . ops . _connect_to_volume_provider ( { } , "<STR_LIT:name>" ) <EOL> self . assertEqual ( "<STR_LIT>" , ref ) <EOL> self . assertEqual ( "<STR_LIT>" , uuid ) <EOL> mock_parse . assert_called_once_with ( { } , "<STR_LIT>" ) <EOL> mock_find_sr . assert_called_once_with ( self . session , "<STR_LIT>" ) <EOL> self . assertFalse ( mock_introduce_sr . called ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> def test_connect_hypervisor_to_volume_regular ( self , mock_intro ) : <EOL> mock_intro . return_value = "<STR_LIT>" <EOL> result = self . ops . _connect_hypervisor_to_volume ( "<STR_LIT>" , { } ) <EOL> self . assertEqual ( "<STR_LIT>" , result ) <EOL> mock_intro . assert_called_once_with ( self . session , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> def test_connect_hypervisor_to_volume_vdi ( self , mock_intro ) : <EOL> mock_intro . return_value = "<STR_LIT>" <EOL> conn = { "<STR_LIT>" : "<STR_LIT:id>" } <EOL> result = self . ops . _connect_hypervisor_to_volume ( "<STR_LIT>" , conn ) <EOL> self . assertEqual ( "<STR_LIT>" , result ) <EOL> mock_intro . assert_called_once_with ( self . session , "<STR_LIT>" , <EOL> vdi_uuid = "<STR_LIT:id>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> def test_connect_hypervisor_to_volume_lun ( self , mock_intro ) : <EOL> mock_intro . return_value = "<STR_LIT>" <EOL> conn = { "<STR_LIT>" : "<STR_LIT>" } <EOL> result = self . ops . _connect_hypervisor_to_volume ( "<STR_LIT>" , conn ) <EOL> self . assertEqual ( "<STR_LIT>" , result ) <EOL> mock_intro . assert_called_once_with ( self . session , "<STR_LIT>" , <EOL> target_lun = "<STR_LIT>" ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . LOG , '<STR_LIT>' ) <EOL> def test_connect_hypervisor_to_volume_mask_password ( self , mock_debug , <EOL> mock_intro ) : <EOL> data = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . ops . _connect_hypervisor_to_volume ( "<STR_LIT>" , data ) <EOL> self . assertTrue ( mock_debug . called , '<STR_LIT>' ) <EOL> password_logged = False <EOL> for call in mock_debug . call_args_list : <EOL> if '<STR_LIT>' in call [ <NUM_LIT:0> ] : <EOL> password_logged = True <EOL> break <EOL> self . assertFalse ( password_logged , '<STR_LIT>' ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> def test_attach_volume_to_vm_plug ( self , mock_vbd , mock_shutdown ) : <EOL> mock_vbd . return_value = "<STR_LIT>" <EOL> mock_shutdown . return_value = False <EOL> with mock . patch . object ( self . session . VBD , "<STR_LIT>" ) as mock_plug : <EOL> self . ops . _attach_volume_to_vm ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:name>" , <NUM_LIT:2> , True ) <EOL> mock_plug . assert_called_once_with ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> mock_vbd . assert_called_once_with ( self . session , "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:2> , <EOL> bootable = False , osvol = True ) <EOL> mock_shutdown . assert_called_once_with ( self . session , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> def test_attach_volume_to_vm_no_plug ( self , mock_vbd , mock_shutdown ) : <EOL> mock_vbd . return_value = "<STR_LIT>" <EOL> mock_shutdown . return_value = True <EOL> with mock . patch . object ( self . session . VBD , "<STR_LIT>" ) as mock_plug : <EOL> self . ops . _attach_volume_to_vm ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:name>" , <NUM_LIT:2> , True ) <EOL> self . assertFalse ( mock_plug . called ) <EOL> mock_vbd . assert_called_once_with ( self . session , "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:2> , <EOL> bootable = False , osvol = True ) <EOL> mock_shutdown . assert_called_once_with ( self . session , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( vm_utils , "<STR_LIT>" ) <EOL> def test_attach_volume_to_vm_no_hotplug ( self , mock_vbd , mock_shutdown ) : <EOL> mock_vbd . return_value = "<STR_LIT>" <EOL> with mock . patch . object ( self . session . VBD , "<STR_LIT>" ) as mock_plug : <EOL> self . ops . _attach_volume_to_vm ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:name>" , <NUM_LIT:2> , False ) <EOL> self . assertFalse ( mock_plug . called ) <EOL> mock_vbd . assert_called_once_with ( self . session , "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:2> , <EOL> bootable = False , osvol = True ) <EOL> self . assertFalse ( mock_shutdown . called ) <EOL> class FindBadVolumeTestCase ( VolumeOpsTestBase ) : <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> def test_find_bad_volumes_no_vbds ( self , mock_get_all ) : <EOL> mock_get_all . return_value = [ ] <EOL> result = self . ops . find_bad_volumes ( "<STR_LIT>" ) <EOL> mock_get_all . assert_called_once_with ( "<STR_LIT>" ) <EOL> self . assertEqual ( [ ] , result ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> def test_find_bad_volumes_no_bad_vbds ( self , mock_get_all , mock_find_sr ) : <EOL> mock_get_all . return_value = [ "<STR_LIT:1>" , "<STR_LIT:2>" ] <EOL> mock_find_sr . return_value = "<STR_LIT>" <EOL> with mock . patch . object ( self . session . SR , "<STR_LIT>" ) as mock_scan : <EOL> result = self . ops . find_bad_volumes ( "<STR_LIT>" ) <EOL> mock_get_all . assert_called_once_with ( "<STR_LIT>" ) <EOL> expected_find = [ mock . call ( self . session , "<STR_LIT:1>" ) , <EOL> mock . call ( self . session , "<STR_LIT:2>" ) ] <EOL> self . assertEqual ( expected_find , mock_find_sr . call_args_list ) <EOL> expected_scan = [ mock . call ( "<STR_LIT>" ) , mock . call ( "<STR_LIT>" ) ] <EOL> self . assertEqual ( expected_scan , mock_scan . call_args_list ) <EOL> self . assertEqual ( [ ] , result ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> def test_find_bad_volumes_bad_vbds ( self , mock_get_all , mock_find_sr ) : <EOL> mock_get_all . return_value = [ "<STR_LIT>" ] <EOL> mock_find_sr . return_value = "<STR_LIT>" <EOL> class FakeException ( Exception ) : <EOL> details = [ '<STR_LIT>' , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> session = mock . Mock ( ) <EOL> session . XenAPI . Failure = FakeException <EOL> self . ops . _session = session <EOL> with mock . patch . object ( session . SR , "<STR_LIT>" ) as mock_scan : <EOL> with mock . patch . object ( session . VBD , <EOL> "<STR_LIT>" ) as mock_get : <EOL> mock_scan . side_effect = FakeException <EOL> mock_get . return_value = "<STR_LIT>" <EOL> result = self . ops . find_bad_volumes ( "<STR_LIT>" ) <EOL> mock_get_all . assert_called_once_with ( "<STR_LIT>" ) <EOL> mock_scan . assert_called_once_with ( "<STR_LIT>" ) <EOL> mock_get . assert_called_once_with ( "<STR_LIT>" ) <EOL> self . assertEqual ( [ "<STR_LIT>" ] , result ) <EOL> @ mock . patch . object ( volume_utils , "<STR_LIT>" ) <EOL> @ mock . patch . object ( volumeops . VolumeOps , "<STR_LIT>" ) <EOL> def test_find_bad_volumes_raises ( self , mock_get_all , mock_find_sr ) : <EOL> mock_get_all . return_value = [ "<STR_LIT>" ] <EOL> mock_find_sr . return_value = "<STR_LIT>" <EOL> class FakeException ( Exception ) : <EOL> details = [ '<STR_LIT:foo>' , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> session = mock . Mock ( ) <EOL> session . XenAPI . Failure = FakeException <EOL> self . ops . _session = session <EOL> with mock . patch . object ( session . SR , "<STR_LIT>" ) as mock_scan : <EOL> with mock . patch . object ( session . VBD , <EOL> "<STR_LIT>" ) as mock_get : <EOL> mock_scan . side_effect = FakeException <EOL> mock_get . return_value = "<STR_LIT>" <EOL> self . assertRaises ( FakeException , <EOL> self . ops . find_bad_volumes , "<STR_LIT>" ) <EOL> mock_scan . assert_called_once_with ( "<STR_LIT>" ) <EOL> class CleanupFromVDIsTestCase ( VolumeOpsTestBase ) : <EOL> def _check_find_purge_calls ( self , find_sr_from_vdi , purge_sr , vdi_refs , <EOL> sr_refs ) : <EOL> find_sr_calls = [ mock . call ( self . ops . _session , vdi_ref ) for vdi_ref <EOL> in vdi_refs ] <EOL> find_sr_from_vdi . assert_has_calls ( find_sr_calls ) <EOL> purge_sr_calls = [ mock . call ( self . ops . _session , sr_ref ) for sr_ref <EOL> in sr_refs ] <EOL> purge_sr . assert_has_calls ( purge_sr_calls ) <EOL> @ mock . patch . object ( volume_utils , '<STR_LIT>' ) <EOL> @ mock . patch . object ( volume_utils , '<STR_LIT>' ) <EOL> def test_safe_cleanup_from_vdis ( self , purge_sr , find_sr_from_vdi ) : <EOL> vdi_refs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> sr_refs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> find_sr_from_vdi . side_effect = sr_refs <EOL> self . ops . safe_cleanup_from_vdis ( vdi_refs ) <EOL> self . _check_find_purge_calls ( find_sr_from_vdi , purge_sr , vdi_refs , <EOL> sr_refs ) <EOL> @ mock . patch . object ( volume_utils , '<STR_LIT>' , <EOL> side_effect = [ exception . StorageError ( reason = '<STR_LIT>' ) , '<STR_LIT>' ] ) <EOL> @ mock . patch . object ( volume_utils , '<STR_LIT>' ) <EOL> def test_safe_cleanup_from_vdis_handles_find_sr_exception ( self , purge_sr , <EOL> find_sr_from_vdi ) : <EOL> vdi_refs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> sr_refs = [ '<STR_LIT>' ] <EOL> find_sr_from_vdi . side_effect = [ exception . StorageError ( reason = '<STR_LIT>' ) , <EOL> sr_refs [ <NUM_LIT:0> ] ] <EOL> self . ops . safe_cleanup_from_vdis ( vdi_refs ) <EOL> self . _check_find_purge_calls ( find_sr_from_vdi , purge_sr , vdi_refs , <EOL> sr_refs ) <EOL> @ mock . patch . object ( volume_utils , '<STR_LIT>' ) <EOL> @ mock . patch . object ( volume_utils , '<STR_LIT>' ) <EOL> def test_safe_cleanup_from_vdis_handles_purge_sr_exception ( self , purge_sr , <EOL> find_sr_from_vdi ) : <EOL> vdi_refs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> sr_refs = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> find_sr_from_vdi . side_effect = sr_refs <EOL> purge_sr . side_effects = [ test . TestingException , None ] <EOL> self . ops . safe_cleanup_from_vdis ( vdi_refs ) <EOL> self . _check_find_purge_calls ( find_sr_from_vdi , purge_sr , vdi_refs , <EOL> sr_refs ) </s>
