#NEWLINE# @ mock . patch . object ( TaskProcessor , 'ensure_task_based_deploy_allowed' ) #NEWLINE# @ fake_tasks ( mock_rpc = True , fake_rpc = False ) #NEWLINE# def test_task_deploy_specified_tasks( self , rpc_cast , * _ ) : #NEWLINE# #INDENT# compute = next ( ( x for x in self . env . nodes if ( 'compute' in x . roles ) ) , None ) #NEWLINE# self . assertIsNotNone ( compute ) #NEWLINE# compute . status = consts . NODE_STATUSES . provisioned #NEWLINE# compute . pending_addition = False #NEWLINE# self . db . flush ( ) #NEWLINE# resp = self . app . put ( ( reverse ( 'DeploySelectedNodesWithTasks' , kwargs = { #NEWLINE# #INDENT# 'cluster_id' : x . cluster . id , #NEWLINE# #UNINDENT# } ) + '?nodes={0}' . format ( compute . uid ) ) , params = '["deploy_legacy"]' , headers = self . default_headers ) #NEWLINE# self . assertNotEqual ( consts . TASK_STATUSES . error , Task . get_by_uuid ( uuid = resp . json_body [ 'uuid' ] , fail_if_not_found = True ) . status ) #NEWLINE# links = rpc_cast . call_args [ 0 ] [ 1 ] [ 'args' ] [ 'tasks_graph' ] #NEWLINE# self . assertItemsEqual ( [ 'deploy_legacy' ] , ( task [ 'id' ] for task in links [ compute . uid ] if ( task [ 'type' ] != consts . ORCHESTRATOR_TASK_TYPES . skipped ) ) )