<s> """<STR_LIT>""" <EOL> import inspect <EOL> import mock <EOL> import re <EOL> import traceback <EOL> import functools <EOL> import sys <EOL> from testutils . config_mock import ConfigMock <EOL> from cmdlib . testsupport . iallocator_mock import patchIAllocator <EOL> from cmdlib . testsupport . livelock_mock import LiveLockMock <EOL> from cmdlib . testsupport . netutils_mock import patchNetutils , SetupDefaultNetutilsMock <EOL> from cmdlib . testsupport . processor_mock import ProcessorMock <EOL> from cmdlib . testsupport . rpc_runner_mock import CreateRpcRunnerMock , RpcResultsBuilder , patchRpc , SetupDefaultRpcModuleMock <EOL> from cmdlib . testsupport . ssh_mock import patchSsh <EOL> from cmdlib . testsupport . wconfd_mock import WConfdMock <EOL> from ganeti . cmdlib . base import LogicalUnit <EOL> from ganeti import errors <EOL> from ganeti import objects <EOL> from ganeti import opcodes <EOL> from ganeti import runtime <EOL> import testutils <EOL> class GanetiContextMock ( object ) : <EOL> cfg = property ( fget = lambda self : self . _test_case . cfg ) <EOL> rpc = property ( fget = lambda self : self . _test_case . rpc ) <EOL> def __init__ ( self , test_case ) : <EOL> self . _test_case = test_case <EOL> self . livelock = LiveLockMock ( ) <EOL> def GetWConfdContext ( self , _ec_id ) : <EOL> return ( None , None , None ) <EOL> def GetConfig ( self , _ec_id ) : <EOL> return self . _test_case . cfg <EOL> def GetRpc ( self , _cfg ) : <EOL> return self . _test_case . rpc <EOL> def AddNode ( self , cfg , node , ec_id ) : <EOL> cfg . AddNode ( node , ec_id ) <EOL> def RemoveNode ( self , cfg , node ) : <EOL> cfg . RemoveNode ( node . uuid ) <EOL> class MockLU ( LogicalUnit ) : <EOL> def BuildHooksNodes ( self ) : <EOL> pass <EOL> def BuildHooksEnv ( self ) : <EOL> pass <EOL> class CmdlibTestCase ( testutils . GanetiTestCase ) : <EOL> """<STR_LIT>""" <EOL> REMOVE = object ( ) <EOL> cluster = property ( fget = lambda self : self . cfg . GetClusterInfo ( ) , <EOL> doc = "<STR_LIT>" ) <EOL> master = property ( fget = lambda self : self . cfg . GetMasterNodeInfo ( ) , <EOL> doc = "<STR_LIT>" ) <EOL> master_uuid = property ( fget = lambda self : self . cfg . GetMasterNode ( ) , <EOL> doc = "<STR_LIT>" ) <EOL> group = property ( fget = lambda self : self . _GetDefaultGroup ( ) , <EOL> doc = "<STR_LIT>" ) <EOL> os = property ( fget = lambda self : self . cfg . GetDefaultOs ( ) , <EOL> doc = "<STR_LIT>" ) <EOL> os_name_variant = property ( <EOL> fget = lambda self : self . os . name + objects . OS . VARIANT_DELIM + <EOL> self . os . supported_variants [ <NUM_LIT:0> ] , <EOL> doc = "<STR_LIT>" ) <EOL> def setUp ( self ) : <EOL> super ( CmdlibTestCase , self ) . setUp ( ) <EOL> self . _iallocator_patcher = None <EOL> self . _netutils_patcher = None <EOL> self . _ssh_patcher = None <EOL> self . _rpc_patcher = None <EOL> try : <EOL> runtime . InitArchInfo ( ) <EOL> except errors . ProgrammerError : <EOL> pass <EOL> self . ResetMocks ( ) <EOL> self . _cleanups = [ ] <EOL> def _StopPatchers ( self ) : <EOL> if self . _iallocator_patcher is not None : <EOL> self . _iallocator_patcher . stop ( ) <EOL> self . _iallocator_patcher = None <EOL> if self . _netutils_patcher is not None : <EOL> self . _netutils_patcher . stop ( ) <EOL> self . _netutils_patcher = None <EOL> if self . _ssh_patcher is not None : <EOL> self . _ssh_patcher . stop ( ) <EOL> self . _ssh_patcher = None <EOL> if self . _rpc_patcher is not None : <EOL> self . _rpc_patcher . stop ( ) <EOL> self . _rpc_patcher = None <EOL> def tearDown ( self ) : <EOL> super ( CmdlibTestCase , self ) . tearDown ( ) <EOL> self . _StopPatchers ( ) <EOL> while self . _cleanups : <EOL> f , args , kwargs = self . _cleanups . pop ( - <NUM_LIT:1> ) <EOL> try : <EOL> f ( * args , ** kwargs ) <EOL> except BaseException , e : <EOL> sys . stderr . write ( '<STR_LIT>' % e ) <EOL> def _GetTestModule ( self ) : <EOL> module = inspect . getsourcefile ( self . __class__ ) . split ( "<STR_LIT:/>" ) [ - <NUM_LIT:1> ] <EOL> suffix = "<STR_LIT>" <EOL> assert module . endswith ( suffix ) , "<STR_LIT>" "<STR_LIT>" % ( suffix , module ) <EOL> return module [ : - len ( suffix ) ] <EOL> def ResetMocks ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cfg = ConfigMock ( ) <EOL> self . rpc = CreateRpcRunnerMock ( ) <EOL> self . ctx = GanetiContextMock ( self ) <EOL> self . wconfd = WConfdMock ( ) <EOL> self . mcpu = ProcessorMock ( self . ctx , self . wconfd ) <EOL> self . _StopPatchers ( ) <EOL> try : <EOL> self . _iallocator_patcher = patchIAllocator ( self . _GetTestModule ( ) ) <EOL> self . iallocator_cls = self . _iallocator_patcher . start ( ) <EOL> except ( ImportError , AttributeError ) : <EOL> self . _iallocator_patcher = None <EOL> try : <EOL> self . _netutils_patcher = patchNetutils ( self . _GetTestModule ( ) ) <EOL> self . netutils_mod = self . _netutils_patcher . start ( ) <EOL> SetupDefaultNetutilsMock ( self . netutils_mod , self . cfg ) <EOL> except ( ImportError , AttributeError ) : <EOL> self . _netutils_patcher = None <EOL> try : <EOL> self . _ssh_patcher = patchSsh ( self . _GetTestModule ( ) ) <EOL> self . ssh_mod = self . _ssh_patcher . start ( ) <EOL> except ( ImportError , AttributeError ) : <EOL> self . _ssh_patcher = None <EOL> try : <EOL> self . _rpc_patcher = patchRpc ( self . _GetTestModule ( ) ) <EOL> self . rpc_mod = self . _rpc_patcher . start ( ) <EOL> SetupDefaultRpcModuleMock ( self . rpc_mod ) <EOL> except ( ImportError , AttributeError ) : <EOL> self . _rpc_patcher = None <EOL> def GetMockLU ( self ) : <EOL> """<STR_LIT>""" <EOL> return MockLU ( self . mcpu , mock . MagicMock ( ) , self . cfg , self . rpc , <EOL> ( <NUM_LIT> , "<STR_LIT>" ) , self . wconfd ) <EOL> def RpcResultsBuilder ( self , use_node_names = False ) : <EOL> """<STR_LIT>""" <EOL> return RpcResultsBuilder ( cfg = self . cfg , use_node_names = use_node_names ) <EOL> def ExecOpCode ( self , opcode ) : <EOL> """<STR_LIT>""" <EOL> return self . mcpu . ExecOpCodeAndRecordOutput ( opcode ) <EOL> def ExecOpCodeExpectException ( self , opcode , <EOL> expected_exception , <EOL> expected_regex = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . ExecOpCode ( opcode ) <EOL> except expected_exception , e : <EOL> if expected_regex is not None : <EOL> assert re . search ( expected_regex , str ( e ) ) is not None , "<STR_LIT>" % ( str ( e ) , expected_regex ) <EOL> except Exception , e : <EOL> tb = traceback . format_exc ( ) <EOL> raise AssertionError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( tb , expected_exception , e , e . __class__ ) ) <EOL> else : <EOL> raise AssertionError ( "<STR_LIT>" % <EOL> expected_exception ) <EOL> def ExecOpCodeExpectOpPrereqError ( self , opcode , expected_regex = None ) : <EOL> """<STR_LIT>""" <EOL> self . ExecOpCodeExpectException ( opcode , errors . OpPrereqError , expected_regex ) <EOL> def ExecOpCodeExpectOpExecError ( self , opcode , expected_regex = None ) : <EOL> """<STR_LIT>""" <EOL> self . ExecOpCodeExpectException ( opcode , errors . OpExecError , expected_regex ) <EOL> def RunWithLockedLU ( self , opcode , test_func ) : <EOL> """<STR_LIT>""" <EOL> return self . mcpu . RunWithLockedLU ( opcode , test_func ) <EOL> def assertLogContainsMessage ( self , expected_msg ) : <EOL> """<STR_LIT>""" <EOL> self . mcpu . assertLogContainsMessage ( expected_msg ) <EOL> def assertLogContainsRegex ( self , expected_regex ) : <EOL> """<STR_LIT>""" <EOL> self . mcpu . assertLogContainsRegex ( expected_regex ) <EOL> def assertHooksCall ( self , nodes , hook_path , phase , <EOL> environment = None , index = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> args = self . rpc . call_hooks_runner . mock_calls [ index ] [ <NUM_LIT:1> ] <EOL> self . assertEqual ( set ( nodes ) , set ( args [ <NUM_LIT:0> ] ) ) <EOL> self . assertEqual ( hook_path , args [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( phase , args [ <NUM_LIT:2> ] ) <EOL> if environment is not None : <EOL> self . assertEqual ( environment , args [ <NUM_LIT:3> ] ) <EOL> def CopyOpCode ( self , opcode , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> state = opcode . __getstate__ ( ) <EOL> for key , value in kwargs . items ( ) : <EOL> if value == self . REMOVE and key in state : <EOL> del state [ key ] <EOL> else : <EOL> state [ key ] = value <EOL> return opcodes . OpCode . LoadOpCode ( state ) <EOL> def _GetDefaultGroup ( self ) : <EOL> for group in self . cfg . GetAllNodeGroupsInfo ( ) . values ( ) : <EOL> if group . name == "<STR_LIT:default>" : <EOL> return group <EOL> assert False <EOL> def MockOut ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> patcher = mock . patch . object ( * args , ** kwargs ) <EOL> mocked = patcher . start ( ) <EOL> self . AddCleanup ( patcher . stop ) <EOL> return mocked <EOL> def AddCleanup ( self , func , * args , ** kwargs ) : <EOL> self . _cleanups . append ( ( func , args , kwargs ) ) <EOL> def assertIn ( self , first , second , msg = None ) : <EOL> if first not in second : <EOL> if msg is None : <EOL> msg = "<STR_LIT>" % ( first , second ) <EOL> self . fail ( msg ) <EOL> def withLockedLU ( func ) : <EOL> """<STR_LIT>""" <EOL> @ functools . wraps ( func ) <EOL> def wrapper ( * args , ** kwargs ) : <EOL> test = args [ <NUM_LIT:0> ] <EOL> assert isinstance ( test , CmdlibTestCase ) <EOL> op = None <EOL> for attr_name in [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] : <EOL> if hasattr ( test , attr_name ) : <EOL> op = getattr ( test , attr_name ) <EOL> break <EOL> assert op is not None <EOL> prepare_fn = None <EOL> if hasattr ( test , "<STR_LIT>" ) : <EOL> prepare_fn = getattr ( test , "<STR_LIT>" ) <EOL> assert callable ( prepare_fn ) <EOL> def callWithLU ( lu ) : <EOL> if prepare_fn : <EOL> prepare_fn ( lu ) <EOL> new_args = list ( args ) <EOL> new_args . append ( lu ) <EOL> func ( * new_args , ** kwargs ) <EOL> return test . RunWithLockedLU ( op , callWithLU ) <EOL> return wrapper </s>
