<s> """<STR_LIT>""" <EOL> import unittest <EOL> from ganeti import constants <EOL> from ganeti . cmdlib import instance_storage <EOL> from ganeti import errors <EOL> from ganeti import objects <EOL> from ganeti import opcodes <EOL> import testutils <EOL> import mock <EOL> import time <EOL> from testsupport import CmdlibTestCase <EOL> class TestCheckNodesFreeDiskOnVG ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . node_uuid = "<STR_LIT>" <EOL> self . node_uuids = [ self . node_uuid ] <EOL> self . node_info = mock . Mock ( ) <EOL> self . es = True <EOL> self . ndparams = { constants . ND_EXCLUSIVE_STORAGE : self . es } <EOL> mock_rpc = mock . Mock ( ) <EOL> mock_rpc . call_node_info = mock . Mock ( ) <EOL> mock_cfg = mock . Mock ( ) <EOL> mock_cfg . GetNodeInfo = mock . Mock ( return_value = self . node_info ) <EOL> mock_cfg . GetNdParams = mock . Mock ( return_value = self . ndparams ) <EOL> self . hvname = "<STR_LIT>" <EOL> self . hvparams = mock . Mock ( ) <EOL> self . clusterinfo = mock . Mock ( ) <EOL> self . clusterinfo . hvparams = { self . hvname : self . hvparams } <EOL> mock_cfg . GetHypervisorType = mock . Mock ( return_value = self . hvname ) <EOL> mock_cfg . GetClusterInfo = mock . Mock ( return_value = self . clusterinfo ) <EOL> self . lu = mock . Mock ( ) <EOL> self . lu . rpc = mock_rpc <EOL> self . lu . cfg = mock_cfg <EOL> self . vg = "<STR_LIT>" <EOL> self . node_name = "<STR_LIT>" <EOL> self . space_info = [ { "<STR_LIT:type>" : constants . ST_LVM_VG , <EOL> "<STR_LIT:name>" : self . vg , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> } ] <EOL> def testPerformNodeInfoCall ( self ) : <EOL> expected_hv_arg = [ ( self . hvname , self . hvparams ) ] <EOL> expected_storage_arg = { self . node_uuid : <EOL> [ ( constants . ST_LVM_VG , self . vg , [ self . es ] ) , <EOL> ( constants . ST_LVM_PV , self . vg , [ self . es ] ) ] } <EOL> instance_storage . _PerformNodeInfoCall ( self . lu , self . node_uuids , self . vg ) <EOL> self . lu . rpc . call_node_info . assert_called_with ( <EOL> self . node_uuids , expected_storage_arg , expected_hv_arg ) <EOL> def testCheckVgCapacityForNode ( self ) : <EOL> requested = <NUM_LIT> <EOL> node_info = ( None , self . space_info , None ) <EOL> instance_storage . _CheckVgCapacityForNode ( self . node_name , node_info , <EOL> self . vg , requested ) <EOL> def testCheckVgCapacityForNodeNotEnough ( self ) : <EOL> requested = <NUM_LIT> <EOL> node_info = ( None , self . space_info , None ) <EOL> self . assertRaises ( <EOL> errors . OpPrereqError , <EOL> instance_storage . _CheckVgCapacityForNode , <EOL> self . node_name , node_info , self . vg , requested ) <EOL> def testCheckVgCapacityForNodeNoStorageData ( self ) : <EOL> node_info = ( None , [ ] , None ) <EOL> self . assertRaises ( <EOL> errors . OpPrereqError , <EOL> instance_storage . _CheckVgCapacityForNode , <EOL> self . node_name , node_info , self . vg , NotImplemented ) <EOL> def testCheckVgCapacityForNodeBogusSize ( self ) : <EOL> broken_space_info = [ { "<STR_LIT:type>" : constants . ST_LVM_VG , <EOL> "<STR_LIT:name>" : self . vg , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] <EOL> node_info = ( None , broken_space_info , None ) <EOL> self . assertRaises ( <EOL> errors . OpPrereqError , <EOL> instance_storage . _CheckVgCapacityForNode , <EOL> self . node_name , node_info , self . vg , NotImplemented ) <EOL> class TestCheckComputeDisksInfo ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . disks = [ <EOL> objects . Disk ( dev_type = constants . DT_PLAIN , size = <NUM_LIT> , <EOL> logical_id = ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> name = "<STR_LIT>" , mode = "<STR_LIT>" , params = { } , <EOL> children = [ ] , uuid = "<STR_LIT>" ) , <EOL> objects . Disk ( dev_type = constants . DT_PLAIN , size = <NUM_LIT> , <EOL> logical_id = ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> name = "<STR_LIT>" , mode = "<STR_LIT>" , params = { } , <EOL> children = [ ] , uuid = "<STR_LIT>" ) <EOL> ] <EOL> self . ext_params = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> self . default_vg = "<STR_LIT>" <EOL> def testComputeDisksInfo ( self ) : <EOL> """<STR_LIT>""" <EOL> disks_info = instance_storage . ComputeDisksInfo ( self . disks , <EOL> constants . DT_EXT , <EOL> self . default_vg , <EOL> self . ext_params ) <EOL> for disk , d in zip ( disks_info , self . disks ) : <EOL> self . assertEqual ( disk . get ( "<STR_LIT:size>" ) , d . size ) <EOL> self . assertEqual ( disk . get ( "<STR_LIT>" ) , d . mode ) <EOL> self . assertEqual ( disk . get ( "<STR_LIT:name>" ) , d . name ) <EOL> self . assertEqual ( disk . get ( "<STR_LIT>" ) , self . ext_params . get ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( disk . get ( "<STR_LIT>" ) , self . ext_params . get ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( disk . get ( "<STR_LIT>" ) , self . ext_params . get ( "<STR_LIT>" ) ) <EOL> def testComputeDisksInfoPlainToDrbd ( self ) : <EOL> disks = [ { constants . IDISK_TYPE : constants . DT_DRBD8 , <EOL> constants . IDISK_SIZE : d . size , <EOL> constants . IDISK_MODE : d . mode , <EOL> constants . IDISK_VG : d . logical_id [ <NUM_LIT:0> ] , <EOL> constants . IDISK_NAME : d . name } <EOL> for d in self . disks ] <EOL> disks_info = instance_storage . ComputeDisksInfo ( self . disks , <EOL> constants . DT_DRBD8 , <EOL> self . default_vg , { } ) <EOL> self . assertEqual ( disks , disks_info ) <EOL> def testComputeDisksInfoFails ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( <EOL> errors . OpPrereqError , instance_storage . ComputeDisksInfo , <EOL> self . disks , constants . DT_EXT , self . default_vg , { } ) <EOL> self . assertRaises ( <EOL> errors . OpPrereqError , instance_storage . ComputeDisksInfo , <EOL> self . disks , constants . DT_DRBD8 , self . default_vg , self . ext_params ) <EOL> self . ext_params . update ( { "<STR_LIT:size>" : <NUM_LIT> } ) <EOL> self . assertRaises ( <EOL> AssertionError , instance_storage . ComputeDisksInfo , <EOL> self . disks , constants . DT_EXT , self . default_vg , self . ext_params ) <EOL> class TestLUInstanceReplaceDisks ( CmdlibTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( TestLUInstanceReplaceDisks , self ) . setUp ( ) <EOL> self . MockOut ( time , '<STR_LIT>' ) <EOL> self . node1 = self . cfg . AddNewNode ( ) <EOL> self . node2 = self . cfg . AddNewNode ( ) <EOL> def MakeOpCode ( self , disks , early_release = False , ignore_ipolicy = False , <EOL> remote_node = False , mode = '<STR_LIT>' , iallocator = None ) : <EOL> return opcodes . OpInstanceReplaceDisks ( <EOL> instance_name = self . instance . name , <EOL> instance_uuid = self . instance . uuid , <EOL> early_release = early_release , <EOL> ignore_ipolicy = ignore_ipolicy , <EOL> mode = mode , <EOL> disks = disks , <EOL> remote_node = self . node2 . name if remote_node else None , <EOL> remote_node_uuid = self . node2 . uuid if remote_node else None , <EOL> iallocator = iallocator ) <EOL> def testInvalidTemplate ( self ) : <EOL> self . instance = self . cfg . AddNewInstance ( admin_state = constants . ADMINST_UP , <EOL> disk_template = '<STR_LIT>' , <EOL> primary_node = self . node1 ) <EOL> opcode = self . MakeOpCode ( [ ] ) <EOL> self . ExecOpCodeExpectOpPrereqError ( <EOL> opcode , '<STR_LIT>' ) <EOL> def SimulateDiskFailure ( self , node , disk ) : <EOL> def Faulty ( node_uuid ) : <EOL> disks = self . cfg . GetInstanceDisks ( node_uuid ) <EOL> return [ i for i , d in enumerate ( disks ) <EOL> if i == disk and node . uuid == node_uuid ] <EOL> self . MockOut ( instance_storage . TLReplaceDisks , '<STR_LIT>' , <EOL> side_effect = Faulty ) <EOL> self . MockOut ( instance_storage . TLReplaceDisks , '<STR_LIT>' ) <EOL> self . MockOut ( instance_storage . TLReplaceDisks , '<STR_LIT>' ) <EOL> self . MockOut ( instance_storage . TLReplaceDisks , '<STR_LIT>' ) <EOL> self . MockOut ( instance_storage . TLReplaceDisks , '<STR_LIT>' ) <EOL> self . MockOut ( instance_storage . LUInstanceReplaceDisks , '<STR_LIT>' ) <EOL> self . MockOut ( instance_storage , '<STR_LIT>' ) <EOL> self . rpc . call_blockdev_addchildren ( ) . fail_msg = None <EOL> def testReplacePrimary ( self ) : <EOL> self . instance = self . cfg . AddNewInstance ( admin_state = constants . ADMINST_UP , <EOL> disk_template = '<STR_LIT>' , <EOL> primary_node = self . node1 , <EOL> secondary_node = self . node2 ) <EOL> self . SimulateDiskFailure ( self . node1 , <NUM_LIT:0> ) <EOL> opcode = self . MakeOpCode ( [ <NUM_LIT:0> ] , mode = '<STR_LIT>' ) <EOL> self . ExecOpCode ( opcode ) <EOL> self . rpc . call_blockdev_rename . assert_any_call ( self . node1 . uuid , [ ] ) <EOL> def testReplaceSecondary ( self ) : <EOL> self . instance = self . cfg . AddNewInstance ( admin_state = constants . ADMINST_UP , <EOL> disk_template = '<STR_LIT>' , <EOL> primary_node = self . node1 , <EOL> secondary_node = self . node2 ) <EOL> self . SimulateDiskFailure ( self . node2 , <NUM_LIT:0> ) <EOL> opcode = self . MakeOpCode ( [ <NUM_LIT:0> ] , mode = '<STR_LIT>' ) <EOL> self . ExecOpCode ( opcode ) <EOL> self . rpc . call_blockdev_rename . assert_any_call ( self . node2 . uuid , [ ] ) <EOL> def testReplaceSecondaryNew ( self ) : <EOL> disk = self . cfg . CreateDisk ( dev_type = constants . DT_DRBD8 , <EOL> primary_node = self . node1 , <EOL> secondary_node = self . node2 ) <EOL> self . instance = self . cfg . AddNewInstance ( admin_state = constants . ADMINST_UP , <EOL> disk_template = '<STR_LIT>' , <EOL> disks = [ disk ] , <EOL> primary_node = self . node1 , <EOL> secondary_node = self . node2 ) <EOL> self . SimulateDiskFailure ( self . node2 , <NUM_LIT:0> ) <EOL> node3 = self . cfg . AddNewNode ( ) <EOL> self . MockOut ( instance_storage . TLReplaceDisks , '<STR_LIT>' , <EOL> return_value = node3 . uuid ) <EOL> self . rpc . call_drbd_disconnect_net ( ) . __getitem__ ( ) . fail_msg = None <EOL> self . rpc . call_blockdev_shutdown ( ) . fail_msg = None <EOL> self . rpc . call_drbd_attach_net ( ) . fail_msg = None <EOL> opcode = self . MakeOpCode ( [ ] , mode = '<STR_LIT>' , <EOL> iallocator = '<STR_LIT>' ) <EOL> self . ExecOpCode ( opcode ) <EOL> self . rpc . call_blockdev_shutdown . assert_any_call ( <EOL> self . node2 . uuid , ( disk , self . instance ) ) <EOL> self . rpc . call_drbd_attach_net . assert_any_call ( <EOL> [ self . node1 . uuid , node3 . uuid ] , ( [ disk ] , self . instance ) , <EOL> False ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> testutils . GanetiTestProgram ( ) </s>
