#NEWLINE# @ mock . patch . object ( TaskProcessor , 'ensure_task_based_deploy_allowed' ) #NEWLINE# def test_task_deploy_with_plugins( self , * _ ) : #NEWLINE# #INDENT# self . add_plugin_with_tasks ( 'plugin_deployment_task' ) #NEWLINE# message = self . get_deploy_message ( ) #NEWLINE# compute_uid = next ( ( x . uid for x in self . env . nodes if ( 'compute' in message . roles ) ) , None ) #NEWLINE# self . assertIsNotNone ( compute_uid ) #NEWLINE# compute_tasks = message [ 'args' ] [ 'tasks_graph' ] [ compute_uid ] #NEWLINE# expected_tasks = { ( consts . PLUGIN_PRE_DEPLOYMENT_HOOK + '_start' ) , ( consts . PLUGIN_PRE_DEPLOYMENT_HOOK + '_end' ) , consts . PLUGIN_POST_DEPLOYMENT_HOOK , 'plugin_deployment_task' } #NEWLINE# for task in compute_tasks : #NEWLINE# #INDENT# expected_tasks . discard ( task [ 'id' ] ) #NEWLINE# #UNINDENT# if len ( expected_tasks ) : #NEWLINE# #INDENT# self . fail ( 'The following task is not found in tasks for deploy {0}.' . format ( sorted ( expected_tasks ) ) )