<s> """<STR_LIT>""" <EOL> import OpenSSL <EOL> import copy <EOL> import unittest <EOL> import operator <EOL> import re <EOL> import shutil <EOL> import os <EOL> from ganeti . cmdlib import cluster <EOL> from ganeti . cmdlib . cluster import verify <EOL> from ganeti import constants <EOL> from ganeti import errors <EOL> from ganeti import netutils <EOL> from ganeti import objects <EOL> from ganeti import opcodes <EOL> from ganeti import utils <EOL> from ganeti import pathutils <EOL> from ganeti import query <EOL> from ganeti . hypervisor import hv_xen <EOL> from testsupport import * <EOL> import testutils <EOL> class TestClusterVerifySsh ( unittest . TestCase ) : <EOL> def testMultipleGroups ( self ) : <EOL> fn = verify . LUClusterVerifyGroup . _SelectSshCheckNodes <EOL> mygroupnodes = [ <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = False ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = False ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = True , <EOL> master_candidate = True ) , <EOL> ] <EOL> nodes = [ <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = True , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = False ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = True , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = True , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = True , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT>" , offline = False , <EOL> master_candidate = True ) , <EOL> ] + mygroupnodes <EOL> assert not utils . FindDuplicates ( map ( operator . attrgetter ( "<STR_LIT:name>" ) , nodes ) ) <EOL> ( online , perhost , _ ) = fn ( mygroupnodes , "<STR_LIT>" , nodes ) <EOL> self . assertEqual ( online , [ "<STR_LIT>" % i for i in range ( <NUM_LIT:20> , <NUM_LIT> ) ] ) <EOL> self . assertEqual ( set ( perhost . keys ( ) ) , set ( online ) ) <EOL> self . assertEqual ( perhost , { <EOL> "<STR_LIT>" : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , <EOL> } ) <EOL> def testSingleGroup ( self ) : <EOL> fn = verify . LUClusterVerifyGroup . _SelectSshCheckNodes <EOL> nodes = [ <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT:default>" , offline = True , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT:default>" , offline = False , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT:default>" , offline = False , <EOL> master_candidate = True ) , <EOL> objects . Node ( name = "<STR_LIT>" , group = "<STR_LIT:default>" , offline = True , <EOL> master_candidate = True ) , <EOL> ] <EOL> assert not utils . FindDuplicates ( map ( operator . attrgetter ( "<STR_LIT:name>" ) , nodes ) ) <EOL> ( online , perhost , _ ) = fn ( nodes , "<STR_LIT:default>" , nodes ) <EOL> self . assertEqual ( online , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . assertEqual ( set ( perhost . keys ( ) ) , set ( online ) ) <EOL> self . assertEqual ( perhost , { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : [ ] , <EOL> } ) <EOL> class TestLUClusterActivateMasterIp ( CmdlibTestCase ) : <EOL> def testSuccess ( self ) : <EOL> op = opcodes . OpClusterActivateMasterIp ( ) <EOL> self . rpc . call_node_activate_master_ip . return_value = self . RpcResultsBuilder ( ) . CreateSuccessfulNodeResult ( self . master ) <EOL> self . ExecOpCode ( op ) <EOL> self . rpc . call_node_activate_master_ip . assert_called_once_with ( <EOL> self . master_uuid , self . cfg . GetMasterNetworkParameters ( ) , False ) <EOL> def testFailure ( self ) : <EOL> op = opcodes . OpClusterActivateMasterIp ( ) <EOL> self . rpc . call_node_activate_master_ip . return_value = self . RpcResultsBuilder ( ) . CreateFailedNodeResult ( self . master ) <EOL> self . ExecOpCodeExpectOpExecError ( op ) <EOL> class TestLUClusterDeactivateMasterIp ( CmdlibTestCase ) : <EOL> def testSuccess ( self ) : <EOL> op = opcodes . OpClusterDeactivateMasterIp ( ) <EOL> self . rpc . call_node_deactivate_master_ip . return_value = self . RpcResultsBuilder ( ) . CreateSuccessfulNodeResult ( self . master ) <EOL> self . ExecOpCode ( op ) <EOL> self . rpc . call_node_deactivate_master_ip . assert_called_once_with ( <EOL> self . master_uuid , self . cfg . GetMasterNetworkParameters ( ) , False ) <EOL> def testFailure ( self ) : <EOL> op = opcodes . OpClusterDeactivateMasterIp ( ) <EOL> self . rpc . call_node_deactivate_master_ip . return_value = self . RpcResultsBuilder ( ) . CreateFailedNodeResult ( self . master ) <EOL> self . ExecOpCodeExpectOpExecError ( op ) <EOL> class TestLUClusterConfigQuery ( CmdlibTestCase ) : <EOL> def testInvalidField ( self ) : <EOL> op = opcodes . OpClusterConfigQuery ( output_fields = [ "<STR_LIT>" ] ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testAllFields ( self ) : <EOL> op = opcodes . OpClusterConfigQuery ( output_fields = query . CLUSTER_FIELDS . keys ( ) ) <EOL> self . rpc . call_get_watcher_pause . return_value = self . RpcResultsBuilder ( ) . CreateSuccessfulNodeResult ( self . master , - <NUM_LIT:1> ) <EOL> ret = self . ExecOpCode ( op ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . rpc . call_get_watcher_pause . call_count ) <EOL> self . assertEqual ( len ( ret ) , len ( query . CLUSTER_FIELDS ) ) <EOL> def testEmpytFields ( self ) : <EOL> op = opcodes . OpClusterConfigQuery ( output_fields = [ ] ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertFalse ( self . rpc . call_get_watcher_pause . called ) <EOL> class TestLUClusterDestroy ( CmdlibTestCase ) : <EOL> def testExistingNodes ( self ) : <EOL> op = opcodes . OpClusterDestroy ( ) <EOL> self . cfg . AddNewNode ( ) <EOL> self . cfg . AddNewNode ( ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testExistingInstances ( self ) : <EOL> op = opcodes . OpClusterDestroy ( ) <EOL> self . cfg . AddNewInstance ( ) <EOL> self . cfg . AddNewInstance ( ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testEmptyCluster ( self ) : <EOL> op = opcodes . OpClusterDestroy ( ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertHooksCall ( [ self . master . uuid ] , constants . GLOBAL_HOOKS_DIR , <EOL> constants . HOOKS_PHASE_PRE , index = <NUM_LIT:0> ) <EOL> self . assertHooksCall ( [ self . master . uuid ] , "<STR_LIT>" , <EOL> constants . HOOKS_PHASE_POST , index = <NUM_LIT:1> ) <EOL> self . assertHooksCall ( [ self . master . uuid ] , constants . GLOBAL_HOOKS_DIR , <EOL> constants . HOOKS_PHASE_POST , index = <NUM_LIT:2> ) <EOL> class TestLUClusterPostInit ( CmdlibTestCase ) : <EOL> def testExecution ( self ) : <EOL> op = opcodes . OpClusterPostInit ( ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertHooksCall ( [ self . master . uuid ] , constants . GLOBAL_HOOKS_DIR , <EOL> constants . HOOKS_PHASE_PRE , index = <NUM_LIT:0> ) <EOL> self . assertHooksCall ( [ self . master . uuid ] , "<STR_LIT>" , <EOL> constants . HOOKS_PHASE_POST , index = <NUM_LIT:1> ) <EOL> self . assertHooksCall ( [ self . master . uuid ] , constants . GLOBAL_HOOKS_DIR , <EOL> constants . HOOKS_PHASE_POST , index = <NUM_LIT:2> ) <EOL> class TestLUClusterQuery ( CmdlibTestCase ) : <EOL> def testSimpleInvocation ( self ) : <EOL> op = opcodes . OpClusterQuery ( ) <EOL> self . ExecOpCode ( op ) <EOL> def testIPv6Cluster ( self ) : <EOL> op = opcodes . OpClusterQuery ( ) <EOL> self . cluster . primary_ip_family = netutils . IP6Address . family <EOL> self . ExecOpCode ( op ) <EOL> class TestLUClusterRedistConf ( CmdlibTestCase ) : <EOL> def testSimpleInvocation ( self ) : <EOL> op = opcodes . OpClusterRedistConf ( ) <EOL> self . ExecOpCode ( op ) <EOL> class TestLUClusterRename ( CmdlibTestCase ) : <EOL> NEW_NAME = "<STR_LIT>" <EOL> NEW_IP = "<STR_LIT>" <EOL> def testNoChanges ( self ) : <EOL> op = opcodes . OpClusterRename ( name = self . cfg . GetClusterName ( ) ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testReachableIp ( self ) : <EOL> op = opcodes . OpClusterRename ( name = self . NEW_NAME ) <EOL> self . netutils_mod . GetHostname . return_value = HostnameMock ( self . NEW_NAME , self . NEW_IP ) <EOL> self . netutils_mod . TcpPing . return_value = True <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testValidRename ( self ) : <EOL> op = opcodes . OpClusterRename ( name = self . NEW_NAME ) <EOL> self . netutils_mod . GetHostname . return_value = HostnameMock ( self . NEW_NAME , self . NEW_IP ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . ssh_mod . WriteKnownHostsFile . call_count ) <EOL> self . rpc . call_node_deactivate_master_ip . assert_called_once_with ( <EOL> self . master_uuid , self . cfg . GetMasterNetworkParameters ( ) , False ) <EOL> self . rpc . call_node_activate_master_ip . assert_called_once_with ( <EOL> self . master_uuid , self . cfg . GetMasterNetworkParameters ( ) , False ) <EOL> def testRenameOfflineMaster ( self ) : <EOL> op = opcodes . OpClusterRename ( name = self . NEW_NAME ) <EOL> self . master . offline = True <EOL> self . netutils_mod . GetHostname . return_value = HostnameMock ( self . NEW_NAME , self . NEW_IP ) <EOL> self . ExecOpCode ( op ) <EOL> class TestLUClusterRepairDiskSizes ( CmdlibTestCase ) : <EOL> def testNoInstances ( self ) : <EOL> op = opcodes . OpClusterRepairDiskSizes ( ) <EOL> self . ExecOpCode ( op ) <EOL> def _SetUpInstanceSingleDisk ( self , dev_type = constants . DT_PLAIN ) : <EOL> pnode = self . master <EOL> snode = self . cfg . AddNewNode ( ) <EOL> disk = self . cfg . CreateDisk ( dev_type = dev_type , <EOL> primary_node = pnode , <EOL> secondary_node = snode ) <EOL> inst = self . cfg . AddNewInstance ( disks = [ disk ] ) <EOL> return ( inst , disk ) <EOL> def testSingleInstanceOnFailingNode ( self ) : <EOL> ( inst , _ ) = self . _SetUpInstanceSingleDisk ( ) <EOL> op = opcodes . OpClusterRepairDiskSizes ( instances = [ inst . name ] ) <EOL> self . rpc . call_blockdev_getdimensions . return_value = self . RpcResultsBuilder ( ) . CreateFailedNodeResult ( self . master ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def _ExecOpClusterRepairDiskSizes ( self , node_data ) : <EOL> op = opcodes . OpClusterRepairDiskSizes ( ) <EOL> self . rpc . call_blockdev_getdimensions . return_value = self . RpcResultsBuilder ( ) . CreateSuccessfulNodeResult ( self . master , node_data ) <EOL> return self . ExecOpCode ( op ) <EOL> def testInvalidResultData ( self ) : <EOL> for data in [ [ ] , [ None ] , [ "<STR_LIT>" ] , [ ( "<STR_LIT>" , "<STR_LIT>" ) ] ] : <EOL> self . ResetMocks ( ) <EOL> self . _SetUpInstanceSingleDisk ( ) <EOL> self . _ExecOpClusterRepairDiskSizes ( data ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testCorrectSize ( self ) : <EOL> self . _SetUpInstanceSingleDisk ( ) <EOL> changed = self . _ExecOpClusterRepairDiskSizes ( [ ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , None ) ] ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( changed ) ) <EOL> def testWrongSize ( self ) : <EOL> self . _SetUpInstanceSingleDisk ( ) <EOL> changed = self . _ExecOpClusterRepairDiskSizes ( [ ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , None ) ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( changed ) ) <EOL> def testCorrectDRBD ( self ) : <EOL> self . _SetUpInstanceSingleDisk ( dev_type = constants . DT_DRBD8 ) <EOL> changed = self . _ExecOpClusterRepairDiskSizes ( [ ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , None ) ] ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( changed ) ) <EOL> def testWrongDRBDChild ( self ) : <EOL> ( _ , disk ) = self . _SetUpInstanceSingleDisk ( dev_type = constants . DT_DRBD8 ) <EOL> disk . children [ <NUM_LIT:0> ] . size = <NUM_LIT> <EOL> changed = self . _ExecOpClusterRepairDiskSizes ( [ ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , None ) ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( changed ) ) <EOL> def testExclusiveStorageInvalidResultData ( self ) : <EOL> self . _SetUpInstanceSingleDisk ( ) <EOL> self . master . ndparams [ constants . ND_EXCLUSIVE_STORAGE ] = True <EOL> self . _ExecOpClusterRepairDiskSizes ( [ ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , None ) ] ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> def testExclusiveStorageCorrectSpindles ( self ) : <EOL> ( _ , disk ) = self . _SetUpInstanceSingleDisk ( ) <EOL> disk . spindles = <NUM_LIT:1> <EOL> self . master . ndparams [ constants . ND_EXCLUSIVE_STORAGE ] = True <EOL> changed = self . _ExecOpClusterRepairDiskSizes ( [ ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , <NUM_LIT:1> ) ] ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( changed ) ) <EOL> def testExclusiveStorageWrongSpindles ( self ) : <EOL> self . _SetUpInstanceSingleDisk ( ) <EOL> self . master . ndparams [ constants . ND_EXCLUSIVE_STORAGE ] = True <EOL> changed = self . _ExecOpClusterRepairDiskSizes ( [ ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , <NUM_LIT:1> ) ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( changed ) ) <EOL> class TestLUClusterSetParams ( CmdlibTestCase ) : <EOL> UID_POOL = [ ( <NUM_LIT:10> , <NUM_LIT:1000> ) ] <EOL> def testUidPool ( self ) : <EOL> op = opcodes . OpClusterSetParams ( uid_pool = self . UID_POOL ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( self . UID_POOL , self . cluster . uid_pool ) <EOL> def testAddUids ( self ) : <EOL> old_pool = [ ( <NUM_LIT:1> , <NUM_LIT:9> ) ] <EOL> self . cluster . uid_pool = list ( old_pool ) <EOL> op = opcodes . OpClusterSetParams ( add_uids = self . UID_POOL ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( set ( self . UID_POOL + old_pool ) , <EOL> set ( self . cluster . uid_pool ) ) <EOL> def testRemoveUids ( self ) : <EOL> additional_pool = [ ( <NUM_LIT:1> , <NUM_LIT:9> ) ] <EOL> self . cluster . uid_pool = self . UID_POOL + additional_pool <EOL> op = opcodes . OpClusterSetParams ( remove_uids = self . UID_POOL ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( additional_pool , self . cluster . uid_pool ) <EOL> def testMacPrefix ( self ) : <EOL> mac_prefix = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( mac_prefix = mac_prefix ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( mac_prefix , self . cluster . mac_prefix ) <EOL> def testEmptyMacPrefix ( self ) : <EOL> mac_prefix = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( mac_prefix = mac_prefix ) <EOL> self . ExecOpCodeExpectOpPrereqError ( <EOL> op , "<STR_LIT>" ) <EOL> def testInvalidMacPrefix ( self ) : <EOL> mac_prefix = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( mac_prefix = mac_prefix ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testMasterNetmask ( self ) : <EOL> op = opcodes . OpClusterSetParams ( master_netmask = <NUM_LIT> ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( <NUM_LIT> , self . cluster . master_netmask ) <EOL> def testInvalidDiskparams ( self ) : <EOL> for diskparams in [ { constants . DT_DISKLESS : { constants . LV_STRIPES : <NUM_LIT:0> } } , <EOL> { constants . DT_DRBD8 : { constants . RBD_POOL : "<STR_LIT>" } } , <EOL> { constants . DT_DRBD8 : { constants . RBD_ACCESS : "<STR_LIT>" } } ] : <EOL> self . ResetMocks ( ) <EOL> op = opcodes . OpClusterSetParams ( diskparams = diskparams ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testValidDiskparams ( self ) : <EOL> diskparams = { constants . DT_RBD : { constants . RBD_POOL : "<STR_LIT>" , <EOL> constants . RBD_ACCESS : "<STR_LIT>" } } <EOL> op = opcodes . OpClusterSetParams ( diskparams = diskparams ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( diskparams [ constants . DT_RBD ] , <EOL> self . cluster . diskparams [ constants . DT_RBD ] ) <EOL> def testMinimalDiskparams ( self ) : <EOL> diskparams = { constants . DT_RBD : { constants . RBD_POOL : "<STR_LIT>" } } <EOL> self . cluster . diskparams = { } <EOL> op = opcodes . OpClusterSetParams ( diskparams = diskparams ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( diskparams , self . cluster . diskparams ) <EOL> def testValidDiskparamsAccess ( self ) : <EOL> for value in constants . DISK_VALID_ACCESS_MODES : <EOL> self . ResetMocks ( ) <EOL> op = opcodes . OpClusterSetParams ( diskparams = { <EOL> constants . DT_RBD : { constants . RBD_ACCESS : value } <EOL> } ) <EOL> self . ExecOpCode ( op ) <EOL> got = self . cluster . diskparams [ constants . DT_RBD ] [ constants . RBD_ACCESS ] <EOL> self . assertEqual ( value , got ) <EOL> def testInvalidDiskparamsAccess ( self ) : <EOL> for value in [ "<STR_LIT:default>" , "<STR_LIT>" ] : <EOL> self . ResetMocks ( ) <EOL> op = opcodes . OpClusterSetParams ( diskparams = { <EOL> constants . DT_RBD : { constants . RBD_ACCESS : value } <EOL> } ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testUnsetDrbdHelperWithDrbdDisks ( self ) : <EOL> self . cfg . AddNewInstance ( disks = [ <EOL> self . cfg . CreateDisk ( dev_type = constants . DT_DRBD8 , create_nodes = True ) ] ) <EOL> op = opcodes . OpClusterSetParams ( drbd_helper = "<STR_LIT>" ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testFileStorageDir ( self ) : <EOL> op = opcodes . OpClusterSetParams ( file_storage_dir = "<STR_LIT>" ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( "<STR_LIT>" , self . cluster . file_storage_dir ) <EOL> def testSetFileStorageDirToCurrentValue ( self ) : <EOL> op = opcodes . OpClusterSetParams ( <EOL> file_storage_dir = self . cluster . file_storage_dir ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testUnsetFileStorageDirFileStorageEnabled ( self ) : <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_FILE ] ) <EOL> op = opcodes . OpClusterSetParams ( file_storage_dir = '<STR_LIT>' ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testUnsetFileStorageDirFileStorageDisabled ( self ) : <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_PLAIN ] ) <EOL> op = opcodes . OpClusterSetParams ( file_storage_dir = '<STR_LIT>' ) <EOL> self . ExecOpCode ( op ) <EOL> def testSetFileStorageDirFileStorageDisabled ( self ) : <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_PLAIN ] ) <EOL> op = opcodes . OpClusterSetParams ( file_storage_dir = '<STR_LIT>' ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testSharedFileStorageDir ( self ) : <EOL> op = opcodes . OpClusterSetParams ( shared_file_storage_dir = "<STR_LIT>" ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( "<STR_LIT>" , self . cluster . shared_file_storage_dir ) <EOL> def testSetSharedFileStorageDirToCurrentValue ( self ) : <EOL> op = opcodes . OpClusterSetParams ( shared_file_storage_dir = "<STR_LIT>" ) <EOL> self . ExecOpCode ( op ) <EOL> op = opcodes . OpClusterSetParams ( shared_file_storage_dir = "<STR_LIT>" ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def testUnsetSharedFileStorageDirSharedFileStorageEnabled ( self ) : <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_SHARED_FILE ] ) <EOL> op = opcodes . OpClusterSetParams ( shared_file_storage_dir = '<STR_LIT>' ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testUnsetSharedFileStorageDirSharedFileStorageDisabled ( self ) : <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_PLAIN ] ) <EOL> op = opcodes . OpClusterSetParams ( shared_file_storage_dir = '<STR_LIT>' ) <EOL> self . ExecOpCode ( op ) <EOL> def testSetSharedFileStorageDirSharedFileStorageDisabled ( self ) : <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_PLAIN ] ) <EOL> op = opcodes . OpClusterSetParams ( shared_file_storage_dir = '<STR_LIT>' ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def testValidDrbdHelper ( self ) : <EOL> node1 = self . cfg . AddNewNode ( ) <EOL> node1 . offline = True <EOL> self . rpc . call_drbd_helper . return_value = self . RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , "<STR_LIT>" ) . AddOfflineNode ( node1 ) . Build ( ) <EOL> op = opcodes . OpClusterSetParams ( drbd_helper = "<STR_LIT>" ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testDrbdHelperFailingNode ( self ) : <EOL> self . rpc . call_drbd_helper . return_value = self . RpcResultsBuilder ( ) . AddFailedNode ( self . master ) . Build ( ) <EOL> op = opcodes . OpClusterSetParams ( drbd_helper = "<STR_LIT>" ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testInvalidDrbdHelper ( self ) : <EOL> self . rpc . call_drbd_helper . return_value = self . RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , "<STR_LIT>" ) . Build ( ) <EOL> op = opcodes . OpClusterSetParams ( drbd_helper = "<STR_LIT>" ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testDrbdHelperWithoutDrbdDiskTemplate ( self ) : <EOL> drbd_helper = "<STR_LIT>" <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_DISKLESS ] ) <EOL> self . rpc . call_drbd_helper . return_value = self . RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , drbd_helper ) . Build ( ) <EOL> op = opcodes . OpClusterSetParams ( drbd_helper = drbd_helper ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testResetDrbdHelperDrbdDisabled ( self ) : <EOL> drbd_helper = "<STR_LIT>" <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_DISKLESS ] ) <EOL> op = opcodes . OpClusterSetParams ( drbd_helper = drbd_helper ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( None , self . cluster . drbd_usermode_helper ) <EOL> def testResetDrbdHelperDrbdEnabled ( self ) : <EOL> drbd_helper = "<STR_LIT>" <EOL> self . cluster . enabled_disk_templates = [ constants . DT_DRBD8 ] <EOL> op = opcodes . OpClusterSetParams ( drbd_helper = drbd_helper ) <EOL> self . ExecOpCodeExpectOpPrereqError ( <EOL> op , "<STR_LIT>" ) <EOL> def testEnableDrbdNoHelper ( self ) : <EOL> self . cluster . enabled_disk_templates = [ constants . DT_DISKLESS ] <EOL> self . cluster . drbd_usermode_helper = None <EOL> enabled_disk_templates = [ constants . DT_DRBD8 ] <EOL> op = opcodes . OpClusterSetParams ( <EOL> enabled_disk_templates = enabled_disk_templates ) <EOL> self . ExecOpCodeExpectOpPrereqError ( <EOL> op , "<STR_LIT>" ) <EOL> def testEnableDrbdHelperSet ( self ) : <EOL> drbd_helper = "<STR_LIT>" <EOL> self . rpc . call_drbd_helper . return_value = self . RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , drbd_helper ) . Build ( ) <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_DISKLESS ] ) <EOL> self . cluster . drbd_usermode_helper = drbd_helper <EOL> enabled_disk_templates = [ constants . DT_DRBD8 ] <EOL> op = opcodes . OpClusterSetParams ( <EOL> enabled_disk_templates = enabled_disk_templates , <EOL> ipolicy = { constants . IPOLICY_DTS : enabled_disk_templates } ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( drbd_helper , self . cluster . drbd_usermode_helper ) <EOL> def testDrbdHelperAlreadySet ( self ) : <EOL> drbd_helper = "<STR_LIT>" <EOL> self . rpc . call_drbd_helper . return_value = self . RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , "<STR_LIT>" ) . Build ( ) <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_DISKLESS ] ) <EOL> op = opcodes . OpClusterSetParams ( drbd_helper = drbd_helper ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( drbd_helper , self . cluster . drbd_usermode_helper ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testSetDrbdHelper ( self ) : <EOL> drbd_helper = "<STR_LIT>" <EOL> self . rpc . call_drbd_helper . return_value = self . RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , "<STR_LIT>" ) . Build ( ) <EOL> self . cluster . drbd_usermode_helper = "<STR_LIT>" <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_DRBD8 ] ) <EOL> op = opcodes . OpClusterSetParams ( drbd_helper = drbd_helper ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( drbd_helper , self . cluster . drbd_usermode_helper ) <EOL> def testBeparams ( self ) : <EOL> beparams = { constants . BE_VCPUS : <NUM_LIT:32> } <EOL> op = opcodes . OpClusterSetParams ( beparams = beparams ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( <NUM_LIT:32> , self . cluster <EOL> . beparams [ constants . PP_DEFAULT ] [ constants . BE_VCPUS ] ) <EOL> def testNdparams ( self ) : <EOL> ndparams = { constants . ND_EXCLUSIVE_STORAGE : True } <EOL> op = opcodes . OpClusterSetParams ( ndparams = ndparams ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( True , self . cluster <EOL> . ndparams [ constants . ND_EXCLUSIVE_STORAGE ] ) <EOL> def testNdparamsResetOobProgram ( self ) : <EOL> ndparams = { constants . ND_OOB_PROGRAM : "<STR_LIT>" } <EOL> op = opcodes . OpClusterSetParams ( ndparams = ndparams ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( constants . NDC_DEFAULTS [ constants . ND_OOB_PROGRAM ] , <EOL> self . cluster . ndparams [ constants . ND_OOB_PROGRAM ] ) <EOL> def testHvState ( self ) : <EOL> hv_state = { constants . HT_FAKE : { constants . HVST_CPU_TOTAL : <NUM_LIT:8> } } <EOL> op = opcodes . OpClusterSetParams ( hv_state = hv_state ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( <NUM_LIT:8> , self . cluster . hv_state_static <EOL> [ constants . HT_FAKE ] [ constants . HVST_CPU_TOTAL ] ) <EOL> def testDiskState ( self ) : <EOL> disk_state = { <EOL> constants . DT_PLAIN : { <EOL> "<STR_LIT>" : { constants . DS_DISK_TOTAL : <NUM_LIT:10> } <EOL> } <EOL> } <EOL> op = opcodes . OpClusterSetParams ( disk_state = disk_state ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( <NUM_LIT:10> , self . cluster <EOL> . disk_state_static [ constants . DT_PLAIN ] [ "<STR_LIT>" ] <EOL> [ constants . DS_DISK_TOTAL ] ) <EOL> def testDefaultIPolicy ( self ) : <EOL> ipolicy = constants . IPOLICY_DEFAULTS <EOL> op = opcodes . OpClusterSetParams ( ipolicy = ipolicy ) <EOL> self . ExecOpCode ( op ) <EOL> def testIPolicyNewViolation ( self ) : <EOL> import ganeti . constants as C <EOL> ipolicy = C . IPOLICY_DEFAULTS <EOL> ipolicy [ C . ISPECS_MINMAX ] [ <NUM_LIT:0> ] [ C . ISPECS_MIN ] [ C . ISPEC_MEM_SIZE ] = <NUM_LIT> <EOL> ipolicy [ C . ISPECS_MINMAX ] [ <NUM_LIT:0> ] [ C . ISPECS_MAX ] [ C . ISPEC_MEM_SIZE ] = <NUM_LIT> <EOL> self . cfg . AddNewInstance ( beparams = { C . BE_MINMEM : <NUM_LIT> , C . BE_MAXMEM : <NUM_LIT> } ) <EOL> op = opcodes . OpClusterSetParams ( ipolicy = ipolicy ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testNicparamsNoInstance ( self ) : <EOL> nicparams = { <EOL> constants . NIC_LINK : "<STR_LIT>" <EOL> } <EOL> op = opcodes . OpClusterSetParams ( nicparams = nicparams ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> self . cluster . nicparams <EOL> [ constants . PP_DEFAULT ] [ constants . NIC_LINK ] ) <EOL> def testNicparamsInvalidConf ( self ) : <EOL> nicparams = { <EOL> constants . NIC_MODE : constants . NIC_MODE_BRIDGED , <EOL> constants . NIC_LINK : "<STR_LIT>" <EOL> } <EOL> op = opcodes . OpClusterSetParams ( nicparams = nicparams ) <EOL> self . ExecOpCodeExpectException ( op , errors . ConfigurationError , "<STR_LIT>" ) <EOL> def testNicparamsInvalidInstanceConf ( self ) : <EOL> nicparams = { <EOL> constants . NIC_MODE : constants . NIC_MODE_BRIDGED , <EOL> constants . NIC_LINK : "<STR_LIT>" <EOL> } <EOL> self . cfg . AddNewInstance ( nics = [ <EOL> self . cfg . CreateNic ( nicparams = { constants . NIC_LINK : None } ) ] ) <EOL> op = opcodes . OpClusterSetParams ( nicparams = nicparams ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testNicparamsMissingIp ( self ) : <EOL> nicparams = { <EOL> constants . NIC_MODE : constants . NIC_MODE_ROUTED <EOL> } <EOL> self . cfg . AddNewInstance ( ) <EOL> op = opcodes . OpClusterSetParams ( nicparams = nicparams ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testNicparamsWithInstance ( self ) : <EOL> nicparams = { <EOL> constants . NIC_LINK : "<STR_LIT>" <EOL> } <EOL> self . cfg . AddNewInstance ( ) <EOL> op = opcodes . OpClusterSetParams ( nicparams = nicparams ) <EOL> self . ExecOpCode ( op ) <EOL> def testDefaultHvparams ( self ) : <EOL> hvparams = constants . HVC_DEFAULTS <EOL> op = opcodes . OpClusterSetParams ( hvparams = hvparams ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( hvparams , self . cluster . hvparams ) <EOL> def testMinimalHvparams ( self ) : <EOL> hvparams = { <EOL> constants . HT_FAKE : { <EOL> constants . HV_MIGRATION_MODE : constants . HT_MIGRATION_NONLIVE <EOL> } <EOL> } <EOL> self . cluster . hvparams = { } <EOL> op = opcodes . OpClusterSetParams ( hvparams = hvparams ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( hvparams , self . cluster . hvparams ) <EOL> def testOsHvp ( self ) : <EOL> os_hvp = { <EOL> "<STR_LIT>" : { <EOL> constants . HT_FAKE : { <EOL> constants . HV_MIGRATION_MODE : constants . HT_MIGRATION_NONLIVE <EOL> } <EOL> } , <EOL> "<STR_LIT>" : constants . HVC_DEFAULTS <EOL> } <EOL> op = opcodes . OpClusterSetParams ( os_hvp = os_hvp ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( constants . HT_MIGRATION_NONLIVE , <EOL> self . cluster . os_hvp [ "<STR_LIT>" ] [ constants . HT_FAKE ] <EOL> [ constants . HV_MIGRATION_MODE ] ) <EOL> self . assertEqual ( constants . HVC_DEFAULTS , self . cluster . os_hvp [ "<STR_LIT>" ] ) <EOL> def testRemoveOsHvp ( self ) : <EOL> os_hvp = { "<STR_LIT>" : { constants . HT_FAKE : None } } <EOL> op = opcodes . OpClusterSetParams ( os_hvp = os_hvp ) <EOL> self . ExecOpCode ( op ) <EOL> assert constants . HT_FAKE not in self . cluster . os_hvp [ "<STR_LIT>" ] <EOL> def testDefaultOsHvp ( self ) : <EOL> os_hvp = { "<STR_LIT>" : constants . HVC_DEFAULTS . copy ( ) } <EOL> self . cluster . os_hvp = { "<STR_LIT>" : { } } <EOL> op = opcodes . OpClusterSetParams ( os_hvp = os_hvp ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( os_hvp , self . cluster . os_hvp ) <EOL> def testOsparams ( self ) : <EOL> osparams = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : None <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : None <EOL> } <EOL> } <EOL> self . cluster . osparams = { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } <EOL> self . cluster . osparams_private_cluster = { } <EOL> op = opcodes . OpClusterSetParams ( osparams = osparams ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } , self . cluster . osparams ) <EOL> def testEnabledHypervisors ( self ) : <EOL> enabled_hypervisors = [ constants . HT_XEN_HVM , constants . HT_XEN_PVM ] <EOL> op = opcodes . OpClusterSetParams ( enabled_hypervisors = enabled_hypervisors ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( enabled_hypervisors , self . cluster . enabled_hypervisors ) <EOL> def testEnabledHypervisorsWithoutHypervisorParams ( self ) : <EOL> enabled_hypervisors = [ constants . HT_FAKE ] <EOL> self . cluster . hvparams = { } <EOL> op = opcodes . OpClusterSetParams ( enabled_hypervisors = enabled_hypervisors ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( enabled_hypervisors , self . cluster . enabled_hypervisors ) <EOL> self . assertEqual ( constants . HVC_DEFAULTS [ constants . HT_FAKE ] , <EOL> self . cluster . hvparams [ constants . HT_FAKE ] ) <EOL> @ testutils . patch_object ( utils , "<STR_LIT>" ) <EOL> def testValidDefaultIallocator ( self , find_file_mock ) : <EOL> find_file_mock . return_value = "<STR_LIT>" <EOL> default_iallocator = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( default_iallocator = default_iallocator ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( default_iallocator , self . cluster . default_iallocator ) <EOL> @ testutils . patch_object ( utils , "<STR_LIT>" ) <EOL> def testInvalidDefaultIallocator ( self , find_file_mock ) : <EOL> find_file_mock . return_value = None <EOL> default_iallocator = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( default_iallocator = default_iallocator ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testEnabledDiskTemplates ( self ) : <EOL> enabled_disk_templates = [ constants . DT_DISKLESS , constants . DT_PLAIN ] <EOL> op = opcodes . OpClusterSetParams ( <EOL> enabled_disk_templates = enabled_disk_templates , <EOL> ipolicy = { constants . IPOLICY_DTS : enabled_disk_templates } ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( enabled_disk_templates , <EOL> self . cluster . enabled_disk_templates ) <EOL> def testEnabledDiskTemplatesVsIpolicy ( self ) : <EOL> enabled_disk_templates = [ constants . DT_DISKLESS , constants . DT_PLAIN ] <EOL> op = opcodes . OpClusterSetParams ( <EOL> enabled_disk_templates = enabled_disk_templates , <EOL> ipolicy = { constants . IPOLICY_DTS : [ constants . DT_FILE ] } ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testDisablingDiskTemplatesOfInstances ( self ) : <EOL> old_disk_templates = [ constants . DT_DISKLESS , constants . DT_PLAIN ] <EOL> self . cfg . SetEnabledDiskTemplates ( old_disk_templates ) <EOL> self . cfg . AddNewInstance ( <EOL> disks = [ self . cfg . CreateDisk ( dev_type = constants . DT_PLAIN ) ] ) <EOL> new_disk_templates = [ constants . DT_DISKLESS , constants . DT_DRBD8 ] <EOL> op = opcodes . OpClusterSetParams ( <EOL> enabled_disk_templates = new_disk_templates , <EOL> ipolicy = { constants . IPOLICY_DTS : new_disk_templates } ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testEnabledDiskTemplatesWithoutVgName ( self ) : <EOL> enabled_disk_templates = [ constants . DT_PLAIN ] <EOL> self . cluster . volume_group_name = None <EOL> op = opcodes . OpClusterSetParams ( <EOL> enabled_disk_templates = enabled_disk_templates ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testDisableDiskTemplateWithExistingInstance ( self ) : <EOL> enabled_disk_templates = [ constants . DT_DISKLESS ] <EOL> self . cfg . AddNewInstance ( <EOL> disks = [ self . cfg . CreateDisk ( dev_type = constants . DT_PLAIN ) ] ) <EOL> op = opcodes . OpClusterSetParams ( <EOL> enabled_disk_templates = enabled_disk_templates , <EOL> ipolicy = { constants . IPOLICY_DTS : enabled_disk_templates } ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testDisableDiskTemplateWithExistingInstanceDiskless ( self ) : <EOL> self . cfg . AddNewInstance ( disks = [ ] ) <EOL> enabled_disk_templates = [ constants . DT_PLAIN ] <EOL> op = opcodes . OpClusterSetParams ( <EOL> enabled_disk_templates = enabled_disk_templates , <EOL> ipolicy = { constants . IPOLICY_DTS : enabled_disk_templates } ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testVgNameNoLvmDiskTemplateEnabled ( self ) : <EOL> vg_name = "<STR_LIT>" <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_DISKLESS ] ) <EOL> op = opcodes . OpClusterSetParams ( vg_name = vg_name ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( vg_name , self . cluster . volume_group_name ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> def testUnsetVgNameWithLvmDiskTemplateEnabled ( self ) : <EOL> vg_name = "<STR_LIT>" <EOL> self . cluster . enabled_disk_templates = [ constants . DT_PLAIN ] <EOL> op = opcodes . OpClusterSetParams ( vg_name = vg_name ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testUnsetVgNameWithLvmInstance ( self ) : <EOL> vg_name = "<STR_LIT>" <EOL> self . cfg . AddNewInstance ( <EOL> disks = [ self . cfg . CreateDisk ( dev_type = constants . DT_PLAIN ) ] ) <EOL> op = opcodes . OpClusterSetParams ( vg_name = vg_name ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testUnsetVgNameWithNoLvmDiskTemplateEnabled ( self ) : <EOL> vg_name = "<STR_LIT>" <EOL> self . cfg . SetEnabledDiskTemplates ( [ constants . DT_DISKLESS ] ) <EOL> op = opcodes . OpClusterSetParams ( vg_name = vg_name ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( None , self . cluster . volume_group_name ) <EOL> def testVgNameToOldName ( self ) : <EOL> vg_name = self . cluster . volume_group_name <EOL> op = opcodes . OpClusterSetParams ( vg_name = vg_name ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testVgNameWithFailingNode ( self ) : <EOL> vg_name = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( vg_name = vg_name ) <EOL> self . rpc . call_vg_list . return_value = self . RpcResultsBuilder ( ) . AddFailedNode ( self . master ) . Build ( ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testVgNameWithValidNode ( self ) : <EOL> vg_name = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( vg_name = vg_name ) <EOL> self . rpc . call_vg_list . return_value = self . RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , { vg_name : <NUM_LIT> * <NUM_LIT> } ) . Build ( ) <EOL> self . ExecOpCode ( op ) <EOL> def testVgNameWithTooSmallNode ( self ) : <EOL> vg_name = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( vg_name = vg_name ) <EOL> self . rpc . call_vg_list . return_value = self . RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , { vg_name : <NUM_LIT:1> } ) . Build ( ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testMiscParameters ( self ) : <EOL> op = opcodes . OpClusterSetParams ( candidate_pool_size = <NUM_LIT> , <EOL> maintain_node_health = True , <EOL> modify_etc_hosts = True , <EOL> prealloc_wipe_disks = True , <EOL> reserved_lvs = [ "<STR_LIT>" ] , <EOL> use_external_mip_script = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> self . assertEqual ( <NUM_LIT> , self . cluster . candidate_pool_size ) <EOL> self . assertEqual ( True , self . cluster . maintain_node_health ) <EOL> self . assertEqual ( True , self . cluster . modify_etc_hosts ) <EOL> self . assertEqual ( True , self . cluster . prealloc_wipe_disks ) <EOL> self . assertEqual ( [ "<STR_LIT>" ] , self . cluster . reserved_lvs ) <EOL> self . assertEqual ( True , self . cluster . use_external_mip_script ) <EOL> def testAddHiddenOs ( self ) : <EOL> self . cluster . hidden_os = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> op = opcodes . OpClusterSetParams ( hidden_os = [ ( constants . DDM_ADD , "<STR_LIT>" ) , <EOL> ( constants . DDM_ADD , "<STR_LIT>" ) ] ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] , self . cluster . hidden_os ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testRemoveBlacklistedOs ( self ) : <EOL> self . cluster . blacklisted_os = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> op = opcodes . OpClusterSetParams ( blacklisted_os = [ <EOL> ( constants . DDM_REMOVE , "<STR_LIT>" ) , <EOL> ( constants . DDM_REMOVE , "<STR_LIT>" ) ] ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( [ "<STR_LIT>" ] , self . cluster . blacklisted_os ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testMasterNetdev ( self ) : <EOL> master_netdev = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( master_netdev = master_netdev ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( master_netdev , self . cluster . master_netdev ) <EOL> def testMasterNetdevFailNoForce ( self ) : <EOL> master_netdev = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( master_netdev = master_netdev ) <EOL> self . rpc . call_node_deactivate_master_ip . return_value = self . RpcResultsBuilder ( ) . CreateFailedNodeResult ( self . master ) <EOL> self . ExecOpCodeExpectOpExecError ( op , "<STR_LIT>" ) <EOL> def testMasterNetdevFailForce ( self ) : <EOL> master_netdev = "<STR_LIT>" <EOL> op = opcodes . OpClusterSetParams ( master_netdev = master_netdev , <EOL> force = True ) <EOL> self . rpc . call_node_deactivate_master_ip . return_value = self . RpcResultsBuilder ( ) . CreateFailedNodeResult ( self . master ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testCompressionToolSuccess ( self ) : <EOL> compression_tools = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> op = opcodes . OpClusterSetParams ( compression_tools = compression_tools ) <EOL> self . ExecOpCode ( op ) <EOL> self . assertEqual ( compression_tools , self . cluster . compression_tools ) <EOL> def testCompressionToolCompatibility ( self ) : <EOL> compression_tools = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> op = opcodes . OpClusterSetParams ( compression_tools = compression_tools ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , "<STR_LIT>" ) <EOL> def testCompressionToolForbiddenValues ( self ) : <EOL> for value in [ "<STR_LIT:none>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> compression_tools = [ value , "<STR_LIT>" ] <EOL> op = opcodes . OpClusterSetParams ( compression_tools = compression_tools ) <EOL> self . ExecOpCodeExpectOpPrereqError ( op , re . escape ( value ) ) <EOL> class TestLUClusterVerify ( CmdlibTestCase ) : <EOL> def testVerifyAllGroups ( self ) : <EOL> op = opcodes . OpClusterVerify ( ) <EOL> result = self . ExecOpCode ( op ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( result [ "<STR_LIT>" ] ) ) <EOL> def testVerifyDefaultGroups ( self ) : <EOL> op = opcodes . OpClusterVerify ( group_name = "<STR_LIT:default>" ) <EOL> result = self . ExecOpCode ( op ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( result [ "<STR_LIT>" ] ) ) <EOL> class TestLUClusterVerifyConfig ( CmdlibTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterVerifyConfig , self ) . setUp ( ) <EOL> self . _load_cert_patcher = testutils . patch_object ( OpenSSL . crypto , "<STR_LIT>" ) <EOL> self . _load_cert_mock = self . _load_cert_patcher . start ( ) <EOL> self . _verify_cert_patcher = testutils . patch_object ( utils , "<STR_LIT>" ) <EOL> self . _verify_cert_mock = self . _verify_cert_patcher . start ( ) <EOL> self . _read_file_patcher = testutils . patch_object ( utils , "<STR_LIT>" ) <EOL> self . _read_file_mock = self . _read_file_patcher . start ( ) <EOL> self . _can_read_patcher = testutils . patch_object ( utils , "<STR_LIT>" ) <EOL> self . _can_read_mock = self . _can_read_patcher . start ( ) <EOL> self . _can_read_mock . return_value = True <EOL> self . _read_file_mock . return_value = True <EOL> self . _verify_cert_mock . return_value = ( None , "<STR_LIT>" ) <EOL> self . _load_cert_mock . return_value = True <EOL> def tearDown ( self ) : <EOL> super ( TestLUClusterVerifyConfig , self ) . tearDown ( ) <EOL> self . _can_read_patcher . stop ( ) <EOL> self . _read_file_patcher . stop ( ) <EOL> self . _verify_cert_patcher . stop ( ) <EOL> self . _load_cert_patcher . stop ( ) <EOL> def testSuccessfulRun ( self ) : <EOL> self . cfg . AddNewInstance ( ) <EOL> op = opcodes . OpClusterVerifyConfig ( ) <EOL> result = self . ExecOpCode ( op ) <EOL> self . assertTrue ( result ) <EOL> def testDanglingNode ( self ) : <EOL> node = self . cfg . AddNewNode ( ) <EOL> self . cfg . AddNewInstance ( primary_node = node ) <EOL> node . group = "<STR_LIT>" <EOL> op = opcodes . OpClusterVerifyConfig ( ) <EOL> result = self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> self . assertFalse ( result ) <EOL> def testDanglingInstance ( self ) : <EOL> inst = self . cfg . AddNewInstance ( ) <EOL> inst . primary_node = "<STR_LIT>" <EOL> op = opcodes . OpClusterVerifyConfig ( ) <EOL> result = self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> self . assertFalse ( result ) <EOL> def testDanglingDisk ( self ) : <EOL> self . cfg . AddOrphanDisk ( ) <EOL> op = opcodes . OpClusterVerifyConfig ( ) <EOL> result = self . ExecOpCode ( op ) <EOL> self . assertTrue ( result ) <EOL> class TestLUClusterVerifyGroup ( CmdlibTestCase ) : <EOL> def testEmptyNodeGroup ( self ) : <EOL> group = self . cfg . AddNewNodeGroup ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = group . name , verbose = True ) <EOL> result = self . ExecOpCode ( op ) <EOL> self . assertTrue ( result ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testSimpleInvocation ( self ) : <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> def testSimpleInvocationWithInstance ( self ) : <EOL> self . cfg . AddNewInstance ( disks = [ ] ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> def testGhostNode ( self ) : <EOL> group = self . cfg . AddNewNodeGroup ( ) <EOL> node = self . cfg . AddNewNode ( group = group . uuid , offline = True ) <EOL> self . master . offline = True <EOL> self . cfg . AddNewInstance ( disk_template = constants . DT_DRBD8 , <EOL> primary_node = self . master , <EOL> secondary_node = node ) <EOL> self . rpc . call_blockdev_getmirrorstatus_multi . return_value = RpcResultsBuilder ( ) . AddOfflineNode ( self . master ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> def testValidRpcResult ( self ) : <EOL> self . cfg . AddNewInstance ( disks = [ ] ) <EOL> self . rpc . call_node_verify . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , { } ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> def testVerifyNodeDrbdSuccess ( self ) : <EOL> ninfo = self . cfg . AddNewNode ( ) <EOL> disk = self . cfg . CreateDisk ( dev_type = constants . DT_DRBD8 , <EOL> primary_node = self . master , <EOL> secondary_node = ninfo ) <EOL> instance = self . cfg . AddNewInstance ( disks = [ disk ] ) <EOL> instanceinfo = self . cfg . GetAllInstancesInfo ( ) <EOL> disks_info = self . cfg . GetAllDisksInfo ( ) <EOL> drbd_map = { ninfo . uuid : { <NUM_LIT:0> : disk . uuid } } <EOL> minors = verify . LUClusterVerifyGroup . _ComputeDrbdMinors ( <EOL> ninfo , instanceinfo , disks_info , drbd_map , lambda * args : None ) <EOL> self . assertEquals ( minors , { <NUM_LIT:0> : ( disk . uuid , instance . uuid , False ) } ) <EOL> class TestLUClusterVerifyClientCerts ( CmdlibTestCase ) : <EOL> def _AddNormalNode ( self ) : <EOL> self . normalnode = copy . deepcopy ( self . master ) <EOL> self . normalnode . master_candidate = False <EOL> self . normalnode . uuid = "<STR_LIT>" <EOL> self . cfg . AddNode ( self . normalnode , None ) <EOL> def testVerifyMasterCandidate ( self ) : <EOL> client_cert = "<STR_LIT>" <EOL> self . cluster . candidate_certs = { self . master . uuid : client_cert } <EOL> self . rpc . call_node_verify . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , <EOL> { constants . NV_CLIENT_CERT : ( None , client_cert ) } ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> def testVerifyMasterCandidateInvalid ( self ) : <EOL> client_cert = "<STR_LIT>" <EOL> self . cluster . candidate_certs = { self . master . uuid : client_cert } <EOL> self . rpc . call_node_verify . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , <EOL> { constants . NV_CLIENT_CERT : ( <NUM_LIT> , "<STR_LIT>" ) } ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testVerifyNoMasterCandidateMap ( self ) : <EOL> client_cert = "<STR_LIT>" <EOL> self . cluster . candidate_certs = { } <EOL> self . rpc . call_node_verify . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , <EOL> { constants . NV_CLIENT_CERT : ( None , client_cert ) } ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> def testVerifyNoSharingMasterCandidates ( self ) : <EOL> client_cert = "<STR_LIT>" <EOL> self . cluster . candidate_certs = { <EOL> self . master . uuid : client_cert , <EOL> "<STR_LIT>" : client_cert } <EOL> self . rpc . call_node_verify . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , <EOL> { constants . NV_CLIENT_CERT : ( None , client_cert ) } ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> def testVerifyMasterCandidateCertMismatch ( self ) : <EOL> client_cert = "<STR_LIT>" <EOL> self . cluster . candidate_certs = { self . master . uuid : "<STR_LIT>" } <EOL> self . rpc . call_node_verify . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , <EOL> { constants . NV_CLIENT_CERT : ( None , client_cert ) } ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testVerifyMasterCandidateUnregistered ( self ) : <EOL> client_cert = "<STR_LIT>" <EOL> self . cluster . candidate_certs = { "<STR_LIT>" : "<STR_LIT>" } <EOL> self . rpc . call_node_verify . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , <EOL> { constants . NV_CLIENT_CERT : ( None , client_cert ) } ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testVerifyMasterCandidateOtherNodesCert ( self ) : <EOL> client_cert = "<STR_LIT>" <EOL> self . cluster . candidate_certs = { "<STR_LIT>" : client_cert } <EOL> self . rpc . call_node_verify . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , <EOL> { constants . NV_CLIENT_CERT : ( None , client_cert ) } ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testNormalNodeStillInList ( self ) : <EOL> self . _AddNormalNode ( ) <EOL> client_cert_master = "<STR_LIT>" <EOL> client_cert_normal = "<STR_LIT>" <EOL> self . cluster . candidate_certs = { <EOL> self . normalnode . uuid : client_cert_normal , <EOL> self . master . uuid : client_cert_master } <EOL> self . rpc . call_node_verify . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . normalnode , <EOL> { constants . NV_CLIENT_CERT : ( None , client_cert_normal ) } ) . AddSuccessfulNode ( self . master , <EOL> { constants . NV_CLIENT_CERT : ( None , client_cert_master ) } ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def testNormalNodeStealingMasterCandidateCert ( self ) : <EOL> self . _AddNormalNode ( ) <EOL> client_cert_master = "<STR_LIT>" <EOL> self . cluster . candidate_certs = { <EOL> self . master . uuid : client_cert_master } <EOL> self . rpc . call_node_verify . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . normalnode , <EOL> { constants . NV_CLIENT_CERT : ( None , client_cert_master ) } ) . AddSuccessfulNode ( self . master , <EOL> { constants . NV_CLIENT_CERT : ( None , client_cert_master ) } ) . Build ( ) <EOL> op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" , verbose = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> class TestLUClusterVerifyGroupMethods ( CmdlibTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterVerifyGroupMethods , self ) . setUp ( ) <EOL> self . op = opcodes . OpClusterVerifyGroup ( group_name = "<STR_LIT:default>" ) <EOL> def PrepareLU ( self , lu ) : <EOL> lu . _exclusive_storage = False <EOL> lu . master_node = self . master_uuid <EOL> lu . group_info = self . group <EOL> verify . LUClusterVerifyGroup . all_node_info = property ( fget = lambda _ : self . cfg . GetAllNodesInfo ( ) ) <EOL> class TestLUClusterVerifyGroupVerifyNode ( TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testInvalidNodeResult ( self , lu ) : <EOL> self . assertFalse ( lu . _VerifyNode ( self . master , None ) ) <EOL> self . assertFalse ( lu . _VerifyNode ( self . master , "<STR_LIT>" ) ) <EOL> @ withLockedLU <EOL> def testInvalidVersion ( self , lu ) : <EOL> self . assertFalse ( lu . _VerifyNode ( self . master , { "<STR_LIT:version>" : None } ) ) <EOL> self . assertFalse ( lu . _VerifyNode ( self . master , { "<STR_LIT:version>" : "<STR_LIT>" } ) ) <EOL> self . assertFalse ( lu . _VerifyNode ( self . master , { <EOL> "<STR_LIT:version>" : ( constants . PROTOCOL_VERSION - <NUM_LIT:1> , constants . RELEASE_VERSION ) <EOL> } ) ) <EOL> self . mcpu . ClearLogMessages ( ) <EOL> self . assertTrue ( lu . _VerifyNode ( self . master , { <EOL> "<STR_LIT:version>" : ( constants . PROTOCOL_VERSION , constants . RELEASE_VERSION + "<STR_LIT:x>" ) <EOL> } ) ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> def _GetValidNodeResult ( self , additional_fields ) : <EOL> ret = { <EOL> "<STR_LIT:version>" : ( constants . PROTOCOL_VERSION , constants . RELEASE_VERSION ) , <EOL> constants . NV_NODESETUP : [ ] <EOL> } <EOL> ret . update ( additional_fields ) <EOL> return ret <EOL> @ withLockedLU <EOL> def testHypervisor ( self , lu ) : <EOL> lu . _VerifyNode ( self . master , self . _GetValidNodeResult ( { <EOL> constants . NV_HYPERVISOR : { <EOL> constants . HT_XEN_PVM : None , <EOL> constants . HT_XEN_HVM : "<STR_LIT>" <EOL> } <EOL> } ) ) <EOL> self . mcpu . assertLogContainsRegex ( constants . HT_XEN_HVM ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testHvParams ( self , lu ) : <EOL> lu . _VerifyNode ( self . master , self . _GetValidNodeResult ( { <EOL> constants . NV_HVPARAMS : [ ( "<STR_LIT>" , constants . HT_XEN_HVM , "<STR_LIT>" ) ] <EOL> } ) ) <EOL> self . mcpu . assertLogContainsRegex ( constants . HT_XEN_HVM ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testSuccessfulResult ( self , lu ) : <EOL> self . assertTrue ( lu . _VerifyNode ( self . master , self . _GetValidNodeResult ( { } ) ) ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> class TestLUClusterVerifyGroupVerifyNodeTime ( TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testInvalidNodeResult ( self , lu ) : <EOL> for ndata in [ { } , { constants . NV_TIME : "<STR_LIT>" } ] : <EOL> self . mcpu . ClearLogMessages ( ) <EOL> lu . _VerifyNodeTime ( self . master , ndata , None , None ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testNodeDiverges ( self , lu ) : <EOL> for ntime in [ ( <NUM_LIT:0> , <NUM_LIT:0> ) , ( <NUM_LIT> , <NUM_LIT:0> ) ] : <EOL> self . mcpu . ClearLogMessages ( ) <EOL> lu . _VerifyNodeTime ( self . master , { constants . NV_TIME : ntime } , <NUM_LIT:1000> , <NUM_LIT> ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testSuccessfulResult ( self , lu ) : <EOL> lu . _VerifyNodeTime ( self . master , { constants . NV_TIME : ( <NUM_LIT:0> , <NUM_LIT:0> ) } , <NUM_LIT:0> , <NUM_LIT:5> ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> class TestLUClusterVerifyGroupUpdateVerifyNodeLVM ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterVerifyGroupUpdateVerifyNodeLVM , self ) . setUp ( ) <EOL> self . VALID_NRESULT = { <EOL> constants . NV_VGLIST : { "<STR_LIT>" : <NUM_LIT> } , <EOL> constants . NV_PVLIST : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:size>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : [ ] <EOL> } <EOL> ] <EOL> } <EOL> @ withLockedLU <EOL> def testNoVgName ( self , lu ) : <EOL> lu . _UpdateVerifyNodeLVM ( self . master , { } , None , None ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testEmptyNodeResult ( self , lu ) : <EOL> lu . _UpdateVerifyNodeLVM ( self . master , { } , "<STR_LIT>" , None ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testValidNodeResult ( self , lu ) : <EOL> lu . _UpdateVerifyNodeLVM ( self . master , self . VALID_NRESULT , "<STR_LIT>" , None ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testValidNodeResultExclusiveStorage ( self , lu ) : <EOL> lu . _exclusive_storage = True <EOL> lu . _UpdateVerifyNodeLVM ( self . master , self . VALID_NRESULT , "<STR_LIT>" , <EOL> verify . LUClusterVerifyGroup . NodeImage ( ) ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> class TestLUClusterVerifyGroupVerifyGroupDRBDVersion ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testEmptyNodeResult ( self , lu ) : <EOL> lu . _VerifyGroupDRBDVersion ( { } ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testValidNodeResult ( self , lu ) : <EOL> lu . _VerifyGroupDRBDVersion ( <EOL> RpcResultsBuilder ( ) <EOL> . AddSuccessfulNode ( self . master , { <EOL> constants . NV_DRBDVERSION : "<STR_LIT>" <EOL> } ) <EOL> . Build ( ) ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testDifferentVersions ( self , lu ) : <EOL> node1 = self . cfg . AddNewNode ( ) <EOL> lu . _VerifyGroupDRBDVersion ( <EOL> RpcResultsBuilder ( ) <EOL> . AddSuccessfulNode ( self . master , { <EOL> constants . NV_DRBDVERSION : "<STR_LIT>" <EOL> } ) <EOL> . AddSuccessfulNode ( node1 , { <EOL> constants . NV_DRBDVERSION : "<STR_LIT>" <EOL> } ) <EOL> . Build ( ) ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> class TestLUClusterVerifyGroupVerifyGroupLVM ( TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testNoVgName ( self , lu ) : <EOL> lu . _VerifyGroupLVM ( None , None ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testNoExclusiveStorage ( self , lu ) : <EOL> lu . _VerifyGroupLVM ( None , "<STR_LIT>" ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testNoPvInfo ( self , lu ) : <EOL> lu . _exclusive_storage = True <EOL> nimg = verify . LUClusterVerifyGroup . NodeImage ( ) <EOL> lu . _VerifyGroupLVM ( { self . master . uuid : nimg } , "<STR_LIT>" ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testValidPvInfos ( self , lu ) : <EOL> lu . _exclusive_storage = True <EOL> node2 = self . cfg . AddNewNode ( ) <EOL> nimg1 = verify . LUClusterVerifyGroup . NodeImage ( uuid = self . master . uuid ) <EOL> nimg1 . pv_min = <NUM_LIT> <EOL> nimg1 . pv_max = <NUM_LIT> <EOL> nimg2 = verify . LUClusterVerifyGroup . NodeImage ( uuid = node2 . uuid ) <EOL> nimg2 . pv_min = <NUM_LIT> <EOL> nimg2 . pv_max = <NUM_LIT> <EOL> lu . _VerifyGroupLVM ( { self . master . uuid : nimg1 , node2 . uuid : nimg2 } , "<STR_LIT>" ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> class TestLUClusterVerifyGroupVerifyNodeBridges ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testNoBridges ( self , lu ) : <EOL> lu . _VerifyNodeBridges ( None , None , None ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testInvalidBridges ( self , lu ) : <EOL> for ndata in [ { } , { constants . NV_BRIDGES : "<STR_LIT>" } ] : <EOL> self . mcpu . ClearLogMessages ( ) <EOL> lu . _VerifyNodeBridges ( self . master , ndata , [ "<STR_LIT>" ] ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . mcpu . ClearLogMessages ( ) <EOL> lu . _VerifyNodeBridges ( self . master , { constants . NV_BRIDGES : [ "<STR_LIT>" ] } , <EOL> [ "<STR_LIT>" ] ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> class TestLUClusterVerifyGroupVerifyNodeUserScripts ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testNoUserScripts ( self , lu ) : <EOL> lu . _VerifyNodeUserScripts ( self . master , { } ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testBrokenUserScripts ( self , lu ) : <EOL> lu . _VerifyNodeUserScripts ( self . master , <EOL> { constants . NV_USERSCRIPTS : [ "<STR_LIT>" ] } ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> class TestLUClusterVerifyGroupVerifyNodeNetwork ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterVerifyGroupVerifyNodeNetwork , self ) . setUp ( ) <EOL> self . VALID_NRESULT = { <EOL> constants . NV_NODELIST : { } , <EOL> constants . NV_NODENETTEST : { } , <EOL> constants . NV_MASTERIP : True <EOL> } <EOL> @ withLockedLU <EOL> def testEmptyNodeResult ( self , lu ) : <EOL> lu . _VerifyNodeNetwork ( self . master , { } ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testValidResult ( self , lu ) : <EOL> lu . _VerifyNodeNetwork ( self . master , self . VALID_NRESULT ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testSshProblem ( self , lu ) : <EOL> self . VALID_NRESULT . update ( { <EOL> constants . NV_NODELIST : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } ) <EOL> lu . _VerifyNodeNetwork ( self . master , self . VALID_NRESULT ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testTcpProblem ( self , lu ) : <EOL> self . VALID_NRESULT . update ( { <EOL> constants . NV_NODENETTEST : { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> } ) <EOL> lu . _VerifyNodeNetwork ( self . master , self . VALID_NRESULT ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testMasterIpNotReachable ( self , lu ) : <EOL> self . VALID_NRESULT . update ( { <EOL> constants . NV_MASTERIP : False <EOL> } ) <EOL> node1 = self . cfg . AddNewNode ( ) <EOL> lu . _VerifyNodeNetwork ( self . master , self . VALID_NRESULT ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> self . mcpu . ClearLogMessages ( ) <EOL> lu . _VerifyNodeNetwork ( node1 , self . VALID_NRESULT ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> class TestLUClusterVerifyGroupVerifyInstance ( TestLUClusterVerifyGroupMethods ) : <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterVerifyGroupVerifyInstance , self ) . setUp ( ) <EOL> self . node1 = self . cfg . AddNewNode ( ) <EOL> self . drbd_inst = self . cfg . AddNewInstance ( <EOL> disks = [ self . cfg . CreateDisk ( dev_type = constants . DT_DRBD8 , <EOL> primary_node = self . master , <EOL> secondary_node = self . node1 ) ] ) <EOL> self . running_inst = self . cfg . AddNewInstance ( <EOL> admin_state = constants . ADMINST_UP , disks_active = True ) <EOL> self . diskless_inst = self . cfg . AddNewInstance ( disks = [ ] ) <EOL> self . master_img = verify . LUClusterVerifyGroup . NodeImage ( uuid = self . master_uuid ) <EOL> self . master_img . volumes = [ "<STR_LIT:/>" . join ( disk . logical_id ) <EOL> for inst in [ self . running_inst , <EOL> self . diskless_inst ] <EOL> for disk in <EOL> self . cfg . GetInstanceDisks ( inst . uuid ) ] <EOL> drbd_inst_disks = self . cfg . GetInstanceDisks ( self . drbd_inst . uuid ) <EOL> self . master_img . volumes . extend ( <EOL> [ "<STR_LIT:/>" . join ( disk . logical_id ) for disk in drbd_inst_disks [ <NUM_LIT:0> ] . children ] ) <EOL> self . master_img . instances = [ self . running_inst . uuid ] <EOL> self . node1_img = verify . LUClusterVerifyGroup . NodeImage ( uuid = self . node1 . uuid ) <EOL> self . node1_img . volumes = [ "<STR_LIT:/>" . join ( disk . logical_id ) for disk in drbd_inst_disks [ <NUM_LIT:0> ] . children ] <EOL> self . node_imgs = { <EOL> self . master_uuid : self . master_img , <EOL> self . node1 . uuid : self . node1_img <EOL> } <EOL> running_inst_disks = self . cfg . GetInstanceDisks ( self . running_inst . uuid ) <EOL> self . diskstatus = { <EOL> self . master_uuid : [ <EOL> ( True , objects . BlockDevStatus ( ldisk_status = constants . LDS_OKAY ) ) <EOL> for _ in running_inst_disks <EOL> ] <EOL> } <EOL> @ withLockedLU <EOL> def testDisklessInst ( self , lu ) : <EOL> lu . _VerifyInstance ( self . diskless_inst , self . node_imgs , { } ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testOfflineNode ( self , lu ) : <EOL> self . master_img . offline = True <EOL> lu . _VerifyInstance ( self . drbd_inst , self . node_imgs , { } ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testRunningOnOfflineNode ( self , lu ) : <EOL> self . master_img . offline = True <EOL> lu . _VerifyInstance ( self . running_inst , self . node_imgs , { } ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testMissingVolume ( self , lu ) : <EOL> self . master_img . volumes = [ ] <EOL> lu . _VerifyInstance ( self . running_inst , self . node_imgs , { } ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testRunningInstanceOnWrongNode ( self , lu ) : <EOL> self . master_img . instances = [ ] <EOL> self . diskless_inst . admin_state = constants . ADMINST_UP <EOL> lu . _VerifyInstance ( self . running_inst , self . node_imgs , { } ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testRunningInstanceOnRightNode ( self , lu ) : <EOL> self . master_img . instances = [ self . running_inst . uuid ] <EOL> lu . _VerifyInstance ( self . running_inst , self . node_imgs , { } ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testValidDiskStatus ( self , lu ) : <EOL> lu . _VerifyInstance ( self . running_inst , self . node_imgs , self . diskstatus ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testDegradedDiskStatus ( self , lu ) : <EOL> self . diskstatus [ self . master_uuid ] [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . is_degraded = True <EOL> lu . _VerifyInstance ( self . running_inst , self . node_imgs , self . diskstatus ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testNotOkayDiskStatus ( self , lu ) : <EOL> self . diskstatus [ self . master_uuid ] [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . is_degraded = True <EOL> self . diskstatus [ self . master_uuid ] [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . ldisk_status = constants . LDS_FAULTY <EOL> lu . _VerifyInstance ( self . running_inst , self . node_imgs , self . diskstatus ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testExclusiveStorageWithInvalidInstance ( self , lu ) : <EOL> self . master . ndparams [ constants . ND_EXCLUSIVE_STORAGE ] = True <EOL> lu . _VerifyInstance ( self . drbd_inst , self . node_imgs , self . diskstatus ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testExclusiveStorageWithValidInstance ( self , lu ) : <EOL> self . master . ndparams [ constants . ND_EXCLUSIVE_STORAGE ] = True <EOL> running_inst_disks = self . cfg . GetInstanceDisks ( self . running_inst . uuid ) <EOL> running_inst_disks [ <NUM_LIT:0> ] . spindles = <NUM_LIT:1> <EOL> feedback_fn = lambda _ : None <EOL> self . cfg . Update ( running_inst_disks [ <NUM_LIT:0> ] , feedback_fn ) <EOL> lu . _VerifyInstance ( self . running_inst , self . node_imgs , self . diskstatus ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testDrbdInTwoGroups ( self , lu ) : <EOL> group = self . cfg . AddNewNodeGroup ( ) <EOL> self . node1 . group = group . uuid <EOL> lu . _VerifyInstance ( self . drbd_inst , self . node_imgs , self . diskstatus ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testOfflineSecondary ( self , lu ) : <EOL> self . node1_img . offline = True <EOL> lu . _VerifyInstance ( self . drbd_inst , self . node_imgs , self . diskstatus ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> class TestLUClusterVerifyGroupVerifyOrphanVolumes ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testOrphanedVolume ( self , lu ) : <EOL> master_img = verify . LUClusterVerifyGroup . NodeImage ( uuid = self . master_uuid ) <EOL> master_img . volumes = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] <EOL> node_imgs = { <EOL> self . master_uuid : master_img <EOL> } <EOL> node_vol_should = { <EOL> self . master_uuid : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> } <EOL> lu . _VerifyOrphanVolumes ( "<STR_LIT>" , node_vol_should , node_imgs , <EOL> utils . FieldSet ( "<STR_LIT>" ) ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogDoesNotContainRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogDoesNotContainRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogDoesNotContainRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogDoesNotContainRegex ( "<STR_LIT>" ) <EOL> class TestLUClusterVerifyGroupVerifyNPlusOneMemory ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testN1Failure ( self , lu ) : <EOL> group1 = self . cfg . AddNewNodeGroup ( ) <EOL> node1 = self . cfg . AddNewNode ( ) <EOL> node2 = self . cfg . AddNewNode ( group = group1 ) <EOL> node3 = self . cfg . AddNewNode ( ) <EOL> inst1 = self . cfg . AddNewInstance ( ) <EOL> inst2 = self . cfg . AddNewInstance ( ) <EOL> inst3 = self . cfg . AddNewInstance ( ) <EOL> node1_img = verify . LUClusterVerifyGroup . NodeImage ( uuid = node1 . uuid ) <EOL> node1_img . sbp = { <EOL> self . master_uuid : [ inst1 . uuid , inst2 . uuid , inst3 . uuid ] <EOL> } <EOL> node2_img = verify . LUClusterVerifyGroup . NodeImage ( uuid = node2 . uuid ) <EOL> node3_img = verify . LUClusterVerifyGroup . NodeImage ( uuid = node3 . uuid ) <EOL> node3_img . offline = True <EOL> node_imgs = { <EOL> node1 . uuid : node1_img , <EOL> node2 . uuid : node2_img , <EOL> node3 . uuid : node3_img <EOL> } <EOL> lu . _VerifyNPlusOneMemory ( node_imgs , self . cfg . GetAllInstancesInfo ( ) ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> self . mcpu . ClearLogMessages ( ) <EOL> node1_img . mfree = <NUM_LIT:1000> <EOL> lu . _VerifyNPlusOneMemory ( node_imgs , self . cfg . GetAllInstancesInfo ( ) ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> class TestLUClusterVerifyGroupVerifyFiles ( TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def test ( self , lu ) : <EOL> node1 = self . cfg . AddNewNode ( master_candidate = False , offline = False , <EOL> vm_capable = True ) <EOL> node2 = self . cfg . AddNewNode ( master_candidate = True , vm_capable = False ) <EOL> node3 = self . cfg . AddNewNode ( master_candidate = False , offline = False , <EOL> vm_capable = True ) <EOL> node4 = self . cfg . AddNewNode ( master_candidate = False , offline = False , <EOL> vm_capable = True ) <EOL> node5 = self . cfg . AddNewNode ( master_candidate = False , offline = True ) <EOL> nodeinfo = [ self . master , node1 , node2 , node3 , node4 , node5 ] <EOL> files_all = set ( [ <EOL> pathutils . CLUSTER_DOMAIN_SECRET_FILE , <EOL> pathutils . RAPI_CERT_FILE , <EOL> pathutils . RAPI_USERS_FILE , <EOL> ] ) <EOL> files_opt = set ( [ <EOL> pathutils . RAPI_USERS_FILE , <EOL> hv_xen . XL_CONFIG_FILE , <EOL> pathutils . VNC_PASSWORD_FILE , <EOL> ] ) <EOL> files_mc = set ( [ <EOL> pathutils . CLUSTER_CONF_FILE , <EOL> ] ) <EOL> files_vm = set ( [ <EOL> hv_xen . XEND_CONFIG_FILE , <EOL> hv_xen . XL_CONFIG_FILE , <EOL> pathutils . VNC_PASSWORD_FILE , <EOL> ] ) <EOL> nvinfo = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . master , { <EOL> constants . NV_FILELIST : { <EOL> pathutils . CLUSTER_CONF_FILE : "<STR_LIT>" , <EOL> pathutils . RAPI_CERT_FILE : "<STR_LIT>" , <EOL> pathutils . CLUSTER_DOMAIN_SECRET_FILE : "<STR_LIT>" , <EOL> hv_xen . XEND_CONFIG_FILE : "<STR_LIT>" , <EOL> hv_xen . XL_CONFIG_FILE : "<STR_LIT>" <EOL> } } ) . AddSuccessfulNode ( node1 , { <EOL> constants . NV_FILELIST : { <EOL> pathutils . RAPI_CERT_FILE : "<STR_LIT>" , <EOL> hv_xen . XEND_CONFIG_FILE : "<STR_LIT>" , <EOL> } <EOL> } ) . AddSuccessfulNode ( node2 , { <EOL> constants . NV_FILELIST : { <EOL> pathutils . RAPI_CERT_FILE : "<STR_LIT>" , <EOL> pathutils . CLUSTER_DOMAIN_SECRET_FILE : "<STR_LIT>" , <EOL> } <EOL> } ) . AddSuccessfulNode ( node3 , { <EOL> constants . NV_FILELIST : { <EOL> pathutils . RAPI_CERT_FILE : "<STR_LIT>" , <EOL> pathutils . CLUSTER_CONF_FILE : "<STR_LIT>" , <EOL> pathutils . CLUSTER_DOMAIN_SECRET_FILE : "<STR_LIT>" , <EOL> pathutils . RAPI_USERS_FILE : "<STR_LIT>" , <EOL> hv_xen . XL_CONFIG_FILE : "<STR_LIT>" <EOL> } <EOL> } ) . AddSuccessfulNode ( node4 , { } ) . AddOfflineNode ( node5 ) . Build ( ) <EOL> assert set ( nvinfo . keys ( ) ) == set ( map ( operator . attrgetter ( "<STR_LIT>" ) , nodeinfo ) ) <EOL> lu . _VerifyFiles ( nodeinfo , self . master_uuid , nvinfo , <EOL> ( files_all , files_opt , files_mc , files_vm ) ) <EOL> expected_msgs = [ <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( pathutils . RAPI_CERT_FILE , node1 . name , node2 . name , node3 . name , <EOL> self . master . name ) , <EOL> "<STR_LIT>" % <EOL> ( pathutils . CLUSTER_DOMAIN_SECRET_FILE , node1 . name ) , <EOL> "<STR_LIT>" % <EOL> ( pathutils . CLUSTER_CONF_FILE , node3 . name ) , <EOL> "<STR_LIT>" % <EOL> ( hv_xen . XEND_CONFIG_FILE , node3 . name ) , <EOL> "<STR_LIT>" % <EOL> ( pathutils . CLUSTER_CONF_FILE , node2 . name ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( pathutils . CLUSTER_CONF_FILE , self . master . name , node3 . name ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( pathutils . RAPI_USERS_FILE , self . master . name , node1 . name , node2 . name ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( hv_xen . XL_CONFIG_FILE , node1 . name ) , <EOL> "<STR_LIT>" , <EOL> ] <EOL> self . assertEqual ( len ( self . mcpu . GetLogMessages ( ) ) , len ( expected_msgs ) ) <EOL> for expected_msg in expected_msgs : <EOL> self . mcpu . assertLogContainsInLine ( expected_msg ) <EOL> class TestLUClusterVerifyGroupVerifyNodeOs ( TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testUpdateNodeOsInvalidNodeResult ( self , lu ) : <EOL> for ndata in [ { } , { constants . NV_OSLIST : "<STR_LIT>" } , { constants . NV_OSLIST : [ "<STR_LIT>" ] } , <EOL> { constants . NV_OSLIST : [ [ "<STR_LIT:1>" , "<STR_LIT:2>" ] ] } ] : <EOL> self . mcpu . ClearLogMessages ( ) <EOL> nimage = verify . LUClusterVerifyGroup . NodeImage ( uuid = self . master_uuid ) <EOL> lu . _UpdateNodeOS ( self . master , ndata , nimage ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testUpdateNodeOsValidNodeResult ( self , lu ) : <EOL> ndata = { <EOL> constants . NV_OSLIST : [ <EOL> [ "<STR_LIT>" , "<STR_LIT>" , True , "<STR_LIT>" , [ "<STR_LIT:default>" ] , [ ] , <EOL> [ constants . OS_API_V20 ] , True ] , <EOL> [ "<STR_LIT>" , "<STR_LIT>" , True , "<STR_LIT>" , [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> [ { "<STR_LIT>" : "<STR_LIT>" } , { "<STR_LIT>" : "<STR_LIT>" } ] , constants . OS_API_VERSIONS , <EOL> True ] <EOL> ] <EOL> } <EOL> nimage = verify . LUClusterVerifyGroup . NodeImage ( uuid = self . master_uuid ) <EOL> lu . _UpdateNodeOS ( self . master , ndata , nimage ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testVerifyNodeOs ( self , lu ) : <EOL> node = self . cfg . AddNewNode ( ) <EOL> nimg_root = verify . LUClusterVerifyGroup . NodeImage ( uuid = self . master_uuid ) <EOL> nimg = verify . LUClusterVerifyGroup . NodeImage ( uuid = node . uuid ) <EOL> nimg_root . os_fail = False <EOL> nimg_root . oslist = { <EOL> "<STR_LIT>" : [ ( "<STR_LIT>" , True , "<STR_LIT>" , set ( [ "<STR_LIT:default>" ] ) , set ( ) , <EOL> set ( [ constants . OS_API_V20 ] ) , True ) ] , <EOL> "<STR_LIT>" : [ ( "<STR_LIT>" , False , "<STR_LIT>" , set ( ) , set ( ) , <EOL> set ( [ constants . OS_API_V20 ] ) , True ) ] , <EOL> "<STR_LIT>" : [ ( "<STR_LIT>" , True , "<STR_LIT>" , set ( ) , set ( ) , set ( ) , True ) ] , <EOL> "<STR_LIT>" : [ ( "<STR_LIT>" , True , "<STR_LIT>" , set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> set ( [ ( "<STR_LIT>" , "<STR_LIT>" ) , ( "<STR_LIT>" , "<STR_LIT>" ) ] ) , <EOL> set ( [ constants . OS_API_V20 ] ) , True ) ] , <EOL> "<STR_LIT>" : [ ( "<STR_LIT>" , True , "<STR_LIT>" , set ( ) , set ( ) , set ( ) , True ) ] , <EOL> } <EOL> nimg . os_fail = False <EOL> nimg . oslist = { <EOL> "<STR_LIT>" : [ ( "<STR_LIT>" , True , "<STR_LIT>" , set ( [ "<STR_LIT:default>" ] ) , set ( ) , <EOL> set ( [ constants . OS_API_V20 ] ) , True ) ] , <EOL> "<STR_LIT>" : [ ( "<STR_LIT>" , True , "<STR_LIT>" , set ( ) , set ( ) , set ( ) , True ) ] , <EOL> "<STR_LIT>" : [ ( "<STR_LIT>" , True , "<STR_LIT>" , set ( [ "<STR_LIT>" , "<STR_LIT>" ] ) , <EOL> set ( [ ( "<STR_LIT>" , "<STR_LIT>" ) , ( "<STR_LIT>" , "<STR_LIT>" ) ] ) , <EOL> set ( [ constants . OS_API_V15 ] ) , True ) ] , <EOL> "<STR_LIT>" : [ ( "<STR_LIT>" , False , "<STR_LIT>" , set ( ) , set ( ) , <EOL> set ( [ constants . OS_API_V20 ] ) , True ) ] , <EOL> "<STR_LIT>" : [ <EOL> ( "<STR_LIT>" , True , "<STR_LIT>" , set ( ) , set ( ) , set ( [ constants . OS_API_V20 ] ) , True ) , <EOL> ( "<STR_LIT>" , True , "<STR_LIT>" , set ( ) , set ( ) , set ( [ constants . OS_API_V20 ] ) , True ) ] , <EOL> "<STR_LIT>" : [ ( "<STR_LIT>" , True , "<STR_LIT>" , set ( ) , set ( ) , set ( ) , False ) ] , <EOL> } <EOL> lu . _VerifyNodeOS ( node , nimg , nimg_root ) <EOL> expected_msgs = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> ] <EOL> self . assertEqual ( len ( expected_msgs ) , len ( self . mcpu . GetLogMessages ( ) ) ) <EOL> for expected_msg in expected_msgs : <EOL> self . mcpu . assertLogContainsRegex ( expected_msg ) <EOL> class TestLUClusterVerifyGroupVerifyAcceptedFileStoragePaths ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testNotMaster ( self , lu ) : <EOL> lu . _VerifyAcceptedFileStoragePaths ( self . master , { } , False ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testNotMasterButRetunedValue ( self , lu ) : <EOL> lu . _VerifyAcceptedFileStoragePaths ( <EOL> self . master , { constants . NV_ACCEPTED_STORAGE_PATHS : [ ] } , False ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testMasterInvalidNodeResult ( self , lu ) : <EOL> lu . _VerifyAcceptedFileStoragePaths ( self . master , { } , True ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testMasterForbiddenPaths ( self , lu ) : <EOL> lu . _VerifyAcceptedFileStoragePaths ( <EOL> self . master , { constants . NV_ACCEPTED_STORAGE_PATHS : [ "<STR_LIT>" ] } , True ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testMasterSuccess ( self , lu ) : <EOL> lu . _VerifyAcceptedFileStoragePaths ( <EOL> self . master , { constants . NV_ACCEPTED_STORAGE_PATHS : [ ] } , True ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> class TestLUClusterVerifyGroupVerifyStoragePaths ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testVerifyFileStoragePathsSuccess ( self , lu ) : <EOL> lu . _VerifyFileStoragePaths ( self . master , { } ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testVerifyFileStoragePathsFailure ( self , lu ) : <EOL> lu . _VerifyFileStoragePaths ( self . master , <EOL> { constants . NV_FILE_STORAGE_PATH : "<STR_LIT>" } ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testVerifySharedFileStoragePathsSuccess ( self , lu ) : <EOL> lu . _VerifySharedFileStoragePaths ( self . master , { } ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testVerifySharedFileStoragePathsFailure ( self , lu ) : <EOL> lu . _VerifySharedFileStoragePaths ( <EOL> self . master , { constants . NV_SHARED_FILE_STORAGE_PATH : "<STR_LIT>" } ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> class TestLUClusterVerifyGroupVerifyOob ( TestLUClusterVerifyGroupMethods ) : <EOL> @ withLockedLU <EOL> def testEmptyResult ( self , lu ) : <EOL> lu . _VerifyOob ( self . master , { } ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testErrorResults ( self , lu ) : <EOL> lu . _VerifyOob ( self . master , { constants . NV_OOB_PATHS : [ "<STR_LIT>" , "<STR_LIT>" ] } ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> class TestLUClusterVerifyGroupUpdateNodeVolumes ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterVerifyGroupUpdateNodeVolumes , self ) . setUp ( ) <EOL> self . nimg = verify . LUClusterVerifyGroup . NodeImage ( uuid = self . master_uuid ) <EOL> @ withLockedLU <EOL> def testNoVgName ( self , lu ) : <EOL> lu . _UpdateNodeVolumes ( self . master , { } , self . nimg , None ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> self . assertTrue ( self . nimg . lvm_fail ) <EOL> @ withLockedLU <EOL> def testErrorMessage ( self , lu ) : <EOL> lu . _UpdateNodeVolumes ( self . master , { constants . NV_LVLIST : "<STR_LIT>" } , <EOL> self . nimg , "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . assertTrue ( self . nimg . lvm_fail ) <EOL> @ withLockedLU <EOL> def testInvalidNodeResult ( self , lu ) : <EOL> lu . _UpdateNodeVolumes ( self . master , { constants . NV_LVLIST : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] } , <EOL> self . nimg , "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> self . assertTrue ( self . nimg . lvm_fail ) <EOL> @ withLockedLU <EOL> def testValidNodeResult ( self , lu ) : <EOL> lu . _UpdateNodeVolumes ( self . master , { constants . NV_LVLIST : { } } , <EOL> self . nimg , "<STR_LIT>" ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> self . assertFalse ( self . nimg . lvm_fail ) <EOL> class TestLUClusterVerifyGroupUpdateNodeInstances ( <EOL> TestLUClusterVerifyGroupMethods ) : <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterVerifyGroupUpdateNodeInstances , self ) . setUp ( ) <EOL> self . nimg = verify . LUClusterVerifyGroup . NodeImage ( uuid = self . master_uuid ) <EOL> @ withLockedLU <EOL> def testInvalidNodeResult ( self , lu ) : <EOL> lu . _UpdateNodeInstances ( self . master , { } , self . nimg ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testValidNodeResult ( self , lu ) : <EOL> inst = self . cfg . AddNewInstance ( ) <EOL> lu . _UpdateNodeInstances ( self . master , <EOL> { constants . NV_INSTANCELIST : [ inst . name ] } , <EOL> self . nimg ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> class TestLUClusterVerifyGroupUpdateNodeInfo ( TestLUClusterVerifyGroupMethods ) : <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterVerifyGroupUpdateNodeInfo , self ) . setUp ( ) <EOL> self . nimg = verify . LUClusterVerifyGroup . NodeImage ( uuid = self . master_uuid ) <EOL> self . valid_hvresult = { constants . NV_HVINFO : { "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> } } <EOL> @ withLockedLU <EOL> def testInvalidHvNodeResult ( self , lu ) : <EOL> for ndata in [ { } , { constants . NV_HVINFO : "<STR_LIT>" } ] : <EOL> self . mcpu . ClearLogMessages ( ) <EOL> lu . _UpdateNodeInfo ( self . master , ndata , self . nimg , None ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testInvalidMemoryFreeHvNodeResult ( self , lu ) : <EOL> lu . _UpdateNodeInfo ( self . master , <EOL> { constants . NV_HVINFO : { "<STR_LIT>" : '<STR_LIT:abc>' , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> } } , <EOL> self . nimg , None ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testValidHvNodeResult ( self , lu ) : <EOL> lu . _UpdateNodeInfo ( self . master , self . valid_hvresult , self . nimg , None ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testInvalidVgNodeResult ( self , lu ) : <EOL> for vgdata in [ [ ] , "<STR_LIT>" ] : <EOL> self . mcpu . ClearLogMessages ( ) <EOL> ndata = { constants . NV_VGLIST : vgdata } <EOL> ndata . update ( self . valid_hvresult ) <EOL> lu . _UpdateNodeInfo ( self . master , ndata , self . nimg , "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testInvalidDiskFreeVgNodeResult ( self , lu ) : <EOL> self . valid_hvresult . update ( { <EOL> constants . NV_VGLIST : { "<STR_LIT>" : "<STR_LIT:abc>" } <EOL> } ) <EOL> lu . _UpdateNodeInfo ( self . master , self . valid_hvresult , self . nimg , "<STR_LIT>" ) <EOL> self . mcpu . assertLogContainsRegex ( <EOL> "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testValidVgNodeResult ( self , lu ) : <EOL> self . valid_hvresult . update ( { <EOL> constants . NV_VGLIST : { "<STR_LIT>" : <NUM_LIT> } <EOL> } ) <EOL> lu . _UpdateNodeInfo ( self . master , self . valid_hvresult , self . nimg , "<STR_LIT>" ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> class TestLUClusterVerifyGroupCollectDiskInfo ( TestLUClusterVerifyGroupMethods ) : <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterVerifyGroupCollectDiskInfo , self ) . setUp ( ) <EOL> self . node1 = self . cfg . AddNewNode ( ) <EOL> self . node2 = self . cfg . AddNewNode ( ) <EOL> self . node3 = self . cfg . AddNewNode ( ) <EOL> self . diskless_inst = self . cfg . AddNewInstance ( primary_node = self . node1 , <EOL> disk_template = constants . DT_DISKLESS ) <EOL> self . plain_inst = self . cfg . AddNewInstance ( primary_node = self . node2 , <EOL> disk_template = constants . DT_PLAIN ) <EOL> self . drbd_inst = self . cfg . AddNewInstance ( primary_node = self . node3 , <EOL> secondary_node = self . node2 , <EOL> disk_template = constants . DT_DRBD8 ) <EOL> self . node1_img = verify . LUClusterVerifyGroup . NodeImage ( <EOL> uuid = self . node1 . uuid ) <EOL> self . node1_img . pinst = [ self . diskless_inst . uuid ] <EOL> self . node1_img . sinst = [ ] <EOL> self . node2_img = verify . LUClusterVerifyGroup . NodeImage ( <EOL> uuid = self . node2 . uuid ) <EOL> self . node2_img . pinst = [ self . plain_inst . uuid ] <EOL> self . node2_img . sinst = [ self . drbd_inst . uuid ] <EOL> self . node3_img = verify . LUClusterVerifyGroup . NodeImage ( <EOL> uuid = self . node3 . uuid ) <EOL> self . node3_img . pinst = [ self . drbd_inst . uuid ] <EOL> self . node3_img . sinst = [ ] <EOL> self . node_images = { <EOL> self . node1 . uuid : self . node1_img , <EOL> self . node2 . uuid : self . node2_img , <EOL> self . node3 . uuid : self . node3_img <EOL> } <EOL> self . node_uuids = [ self . node1 . uuid , self . node2 . uuid , self . node3 . uuid ] <EOL> @ withLockedLU <EOL> def testSuccessfulRun ( self , lu ) : <EOL> self . rpc . call_blockdev_getmirrorstatus_multi . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . node2 , [ ( True , "<STR_LIT>" ) , ( True , "<STR_LIT>" ) ] ) . AddSuccessfulNode ( self . node3 , [ ( True , "<STR_LIT>" ) ] ) . Build ( ) <EOL> lu . _CollectDiskInfo ( self . node_uuids , self . node_images , <EOL> self . cfg . GetAllInstancesInfo ( ) ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> @ withLockedLU <EOL> def testOfflineAndFailingNodes ( self , lu ) : <EOL> self . rpc . call_blockdev_getmirrorstatus_multi . return_value = RpcResultsBuilder ( ) . AddOfflineNode ( self . node2 ) . AddFailedNode ( self . node3 ) . Build ( ) <EOL> lu . _CollectDiskInfo ( self . node_uuids , self . node_images , <EOL> self . cfg . GetAllInstancesInfo ( ) ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testInvalidNodeResult ( self , lu ) : <EOL> self . rpc . call_blockdev_getmirrorstatus_multi . return_value = RpcResultsBuilder ( ) . AddSuccessfulNode ( self . node2 , [ ( True , ) , ( False , ) ] ) . AddSuccessfulNode ( self . node3 , [ "<STR_LIT>" ] ) . Build ( ) <EOL> lu . _CollectDiskInfo ( self . node_uuids , self . node_images , <EOL> self . cfg . GetAllInstancesInfo ( ) ) <EOL> self . mcpu . assertLogIsEmpty ( ) <EOL> class TestLUClusterVerifyGroupHooksCallBack ( TestLUClusterVerifyGroupMethods ) : <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterVerifyGroupHooksCallBack , self ) . setUp ( ) <EOL> self . feedback_fn = lambda _ : None <EOL> def PrepareLU ( self , lu ) : <EOL> super ( TestLUClusterVerifyGroupHooksCallBack , self ) . PrepareLU ( lu ) <EOL> lu . my_node_uuids = list ( self . cfg . GetAllNodesInfo ( ) . keys ( ) ) <EOL> @ withLockedLU <EOL> def testEmptyGroup ( self , lu ) : <EOL> lu . my_node_uuids = [ ] <EOL> lu . HooksCallBack ( constants . HOOKS_PHASE_POST , None , self . feedback_fn , None ) <EOL> @ withLockedLU <EOL> def testFailedResult ( self , lu ) : <EOL> lu . HooksCallBack ( constants . HOOKS_PHASE_POST , <EOL> RpcResultsBuilder ( use_node_names = True ) <EOL> . AddFailedNode ( self . master ) . Build ( ) , <EOL> self . feedback_fn , <EOL> None ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> @ withLockedLU <EOL> def testOfflineNode ( self , lu ) : <EOL> lu . HooksCallBack ( constants . HOOKS_PHASE_POST , <EOL> RpcResultsBuilder ( use_node_names = True ) <EOL> . AddOfflineNode ( self . master ) . Build ( ) , <EOL> self . feedback_fn , <EOL> None ) <EOL> @ withLockedLU <EOL> def testValidResult ( self , lu ) : <EOL> lu . HooksCallBack ( constants . HOOKS_PHASE_POST , <EOL> RpcResultsBuilder ( use_node_names = True ) <EOL> . AddSuccessfulNode ( self . master , <EOL> [ ( "<STR_LIT>" , <EOL> constants . HKR_SUCCESS , <EOL> "<STR_LIT>" ) ] ) <EOL> . Build ( ) , <EOL> self . feedback_fn , <EOL> None ) <EOL> @ withLockedLU <EOL> def testFailedScriptResult ( self , lu ) : <EOL> lu . HooksCallBack ( constants . HOOKS_PHASE_POST , <EOL> RpcResultsBuilder ( use_node_names = True ) <EOL> . AddSuccessfulNode ( self . master , <EOL> [ ( "<STR_LIT>" , <EOL> constants . HKR_FAIL , <EOL> "<STR_LIT>" ) ] ) <EOL> . Build ( ) , <EOL> self . feedback_fn , <EOL> None ) <EOL> self . mcpu . assertLogContainsRegex ( "<STR_LIT>" ) <EOL> class TestLUClusterVerifyDisks ( CmdlibTestCase ) : <EOL> def testVerifyDisks ( self ) : <EOL> op = opcodes . OpClusterVerifyDisks ( ) <EOL> result = self . ExecOpCode ( op ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( result [ "<STR_LIT>" ] ) ) <EOL> class TestLUClusterRenewCrypto ( CmdlibTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestLUClusterRenewCrypto , self ) . setUp ( ) <EOL> self . _node_cert = self . _CreateTempFile ( ) <EOL> shutil . copy ( testutils . TestDataFilename ( "<STR_LIT>" ) , self . _node_cert ) <EOL> self . _client_node_cert = self . _CreateTempFile ( ) <EOL> shutil . copy ( testutils . TestDataFilename ( "<STR_LIT>" ) , self . _client_node_cert ) <EOL> self . _client_node_cert_digest = "<STR_LIT>" <EOL> def tearDown ( self ) : <EOL> super ( TestLUClusterRenewCrypto , self ) . tearDown ( ) <EOL> def _GetFakeDigest ( self , uuid ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( uuid [ <NUM_LIT:0> : <NUM_LIT:2> ] , uuid [ <NUM_LIT:2> : <NUM_LIT:4> ] , uuid [ <NUM_LIT:4> : <NUM_LIT:6> ] , uuid [ <NUM_LIT:6> : <NUM_LIT:8> ] ) <EOL> def _InitPathutils ( self , pathutils ) : <EOL> """<STR_LIT>""" <EOL> pathutils . NODED_CERT_FILE = self . _node_cert <EOL> pathutils . NODED_CLIENT_CERT_FILE = self . _client_node_cert <EOL> def _AssertCertFiles ( self , pathutils ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( os . path . exists ( pathutils . NODED_CERT_FILE ) ) <EOL> self . assertTrue ( os . path . exists ( pathutils . NODED_CLIENT_CERT_FILE ) ) <EOL> def _CompletelySuccessfulRpc ( self , node_uuid , _ ) : <EOL> """<STR_LIT>""" <EOL> return self . RpcResultsBuilder ( ) . CreateSuccessfulNodeResult ( node_uuid , <EOL> [ ( constants . CRYPTO_TYPE_SSL_DIGEST , <EOL> self . _GetFakeDigest ( node_uuid ) ) ] ) <EOL> @ patchPathutils ( "<STR_LIT>" ) <EOL> def testSuccessfulCase ( self , pathutils ) : <EOL> self . _InitPathutils ( pathutils ) <EOL> num_nodes = <NUM_LIT:3> <EOL> for _ in range ( num_nodes ) : <EOL> self . cfg . AddNewNode ( ) <EOL> self . rpc . call_node_crypto_tokens = self . _CompletelySuccessfulRpc <EOL> op = opcodes . OpClusterRenewCrypto ( node_certificates = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . _AssertCertFiles ( pathutils ) <EOL> cluster = self . cfg . GetClusterInfo ( ) <EOL> self . assertEqual ( num_nodes + <NUM_LIT:1> , len ( cluster . candidate_certs ) ) <EOL> nodes = self . cfg . GetAllNodesInfo ( ) <EOL> master_uuid = self . cfg . GetMasterNode ( ) <EOL> for ( node_uuid , _ ) in nodes . items ( ) : <EOL> if node_uuid == master_uuid : <EOL> self . assertEqual ( self . _client_node_cert_digest , <EOL> cluster . candidate_certs [ node_uuid ] ) <EOL> else : <EOL> expected_digest = self . _GetFakeDigest ( node_uuid ) <EOL> self . assertEqual ( expected_digest , cluster . candidate_certs [ node_uuid ] ) <EOL> def _partiallyFailingRpc ( self , node_uuid , _ ) : <EOL> if node_uuid == self . _failed_node : <EOL> return self . RpcResultsBuilder ( ) . CreateFailedNodeResult ( node_uuid ) <EOL> else : <EOL> return self . RpcResultsBuilder ( ) . CreateSuccessfulNodeResult ( node_uuid , <EOL> [ ( constants . CRYPTO_TYPE_SSL_DIGEST , self . _GetFakeDigest ( node_uuid ) ) ] ) <EOL> @ patchPathutils ( "<STR_LIT>" ) <EOL> def testNonMasterFails ( self , pathutils ) : <EOL> self . _InitPathutils ( pathutils ) <EOL> num_nodes = <NUM_LIT:3> <EOL> for _ in range ( num_nodes ) : <EOL> self . cfg . AddNewNode ( ) <EOL> nodes = self . cfg . GetAllNodesInfo ( ) <EOL> master_uuid = self . cfg . GetMasterNode ( ) <EOL> self . _failed_node = [ node_uuid for node_uuid in nodes <EOL> if node_uuid != master_uuid ] [ <NUM_LIT:1> ] <EOL> self . rpc . call_node_crypto_tokens = self . _partiallyFailingRpc <EOL> op = opcodes . OpClusterRenewCrypto ( node_certificates = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . _AssertCertFiles ( pathutils ) <EOL> cluster = self . cfg . GetClusterInfo ( ) <EOL> self . assertEqual ( num_nodes , len ( cluster . candidate_certs ) ) <EOL> nodes = self . cfg . GetAllNodesInfo ( ) <EOL> for ( node_uuid , _ ) in nodes . items ( ) : <EOL> if node_uuid == self . _failed_node : <EOL> self . assertTrue ( node_uuid not in cluster . candidate_certs ) <EOL> else : <EOL> self . assertTrue ( node_uuid in cluster . candidate_certs ) <EOL> @ patchPathutils ( "<STR_LIT>" ) <EOL> def testOfflineNodes ( self , pathutils ) : <EOL> self . _InitPathutils ( pathutils ) <EOL> num_nodes = <NUM_LIT:3> <EOL> offline_index = <NUM_LIT:1> <EOL> for i in range ( num_nodes ) : <EOL> self . cfg . AddNewNode ( offline = ( i == offline_index ) ) <EOL> self . rpc . call_node_crypto_tokens = self . _CompletelySuccessfulRpc <EOL> op = opcodes . OpClusterRenewCrypto ( node_certificates = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . _AssertCertFiles ( pathutils ) <EOL> cluster = self . cfg . GetClusterInfo ( ) <EOL> self . assertEqual ( num_nodes , len ( cluster . candidate_certs ) ) <EOL> nodes = self . cfg . GetAllNodesInfo ( ) <EOL> for ( node_uuid , node_info ) in nodes . items ( ) : <EOL> if node_info . offline == True : <EOL> self . assertTrue ( node_uuid not in cluster . candidate_certs ) <EOL> else : <EOL> self . assertTrue ( node_uuid in cluster . candidate_certs ) <EOL> def _RpcSuccessfulAfterRetries ( self , node_uuid , _ ) : <EOL> if self . _retries < self . _max_retries : <EOL> self . _retries += <NUM_LIT:1> <EOL> return self . RpcResultsBuilder ( ) . CreateFailedNodeResult ( node_uuid ) <EOL> else : <EOL> return self . RpcResultsBuilder ( ) . CreateSuccessfulNodeResult ( node_uuid , <EOL> [ ( constants . CRYPTO_TYPE_SSL_DIGEST , self . _GetFakeDigest ( node_uuid ) ) ] ) <EOL> def _RpcSuccessfulAfterRetriesNonMaster ( self , node_uuid , _ ) : <EOL> if self . _retries < self . _max_retries and node_uuid != self . _master_uuid : <EOL> self . _retries += <NUM_LIT:1> <EOL> return self . RpcResultsBuilder ( ) . CreateFailedNodeResult ( node_uuid ) <EOL> else : <EOL> return self . RpcResultsBuilder ( ) . CreateSuccessfulNodeResult ( node_uuid , <EOL> [ ( constants . CRYPTO_TYPE_SSL_DIGEST , self . _GetFakeDigest ( node_uuid ) ) ] ) <EOL> def _NonMasterRetries ( self , pathutils , max_retries ) : <EOL> self . _InitPathutils ( pathutils ) <EOL> self . _master_uuid = self . cfg . GetMasterNode ( ) <EOL> self . _max_retries = max_retries <EOL> self . _retries = <NUM_LIT:0> <EOL> self . rpc . call_node_crypto_tokens = self . _RpcSuccessfulAfterRetriesNonMaster <EOL> self . cfg . AddNewNode ( ) <EOL> op = opcodes . OpClusterRenewCrypto ( node_certificates = True ) <EOL> self . ExecOpCode ( op ) <EOL> self . _AssertCertFiles ( pathutils ) <EOL> return self . cfg . GetClusterInfo ( ) <EOL> @ patchPathutils ( "<STR_LIT>" ) <EOL> def testNonMasterRetriesSuccess ( self , pathutils ) : <EOL> cluster = self . _NonMasterRetries ( pathutils , <NUM_LIT:2> ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( cluster . candidate_certs . values ( ) ) ) <EOL> @ patchPathutils ( "<STR_LIT>" ) <EOL> def testNonMasterRetriesFail ( self , pathutils ) : <EOL> cluster = self . _NonMasterRetries ( pathutils , <NUM_LIT:5> ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( cluster . candidate_certs . values ( ) ) ) <EOL> self . assertTrue ( self . _master_uuid in cluster . candidate_certs ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> testutils . GanetiTestProgram ( ) </s>
