<s> from __future__ import ( absolute_import , division , generators , nested_scopes , print_function , <EOL> unicode_literals , with_statement ) <EOL> import itertools <EOL> from pants . engine . round_engine import RoundEngine <EOL> from pants . task . task import Task <EOL> from pants_test . base_test import BaseTest <EOL> from pants_test . engine . base_engine_test import EngineTestBase <EOL> class RoundEngineTest ( EngineTestBase , BaseTest ) : <EOL> def setUp ( self ) : <EOL> super ( RoundEngineTest , self ) . setUp ( ) <EOL> self . set_options_for_scope ( '<STR_LIT>' , explain = False ) <EOL> for outer in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> for inner in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> self . set_options_for_scope ( '<STR_LIT>' . format ( outer , inner ) , <EOL> level = '<STR_LIT:info>' , colors = False ) <EOL> self . engine = RoundEngine ( ) <EOL> self . actions = [ ] <EOL> def tearDown ( self ) : <EOL> self . assertTrue ( not self . _context or self . _context . is_unlocked ( ) ) <EOL> super ( RoundEngineTest , self ) . tearDown ( ) <EOL> def alternate_target_roots_action ( self , tag ) : <EOL> return '<STR_LIT>' , tag , self . _context <EOL> def prepare_action ( self , tag ) : <EOL> return '<STR_LIT>' , tag , self . _context <EOL> def execute_action ( self , tag ) : <EOL> return '<STR_LIT>' , tag , self . _context <EOL> def construct_action ( self , tag ) : <EOL> return '<STR_LIT>' , tag , self . _context <EOL> def record ( self , tag , product_types = None , required_data = None , alternate_target_roots = None ) : <EOL> class RecordingTask ( Task ) : <EOL> options_scope = tag <EOL> @ classmethod <EOL> def product_types ( cls ) : <EOL> return product_types or [ ] <EOL> @ classmethod <EOL> def alternate_target_roots ( cls , options , address_mapper , build_graph ) : <EOL> self . actions . append ( self . alternate_target_roots_action ( tag ) ) <EOL> return alternate_target_roots <EOL> @ classmethod <EOL> def prepare ( cls , options , round_manager ) : <EOL> for requirement in ( required_data or ( ) ) : <EOL> round_manager . require_data ( requirement ) <EOL> self . actions . append ( self . prepare_action ( tag ) ) <EOL> def __init__ ( me , * args , ** kwargs ) : <EOL> super ( RecordingTask , me ) . __init__ ( * args , ** kwargs ) <EOL> self . actions . append ( self . construct_action ( tag ) ) <EOL> def execute ( me ) : <EOL> self . actions . append ( self . execute_action ( tag ) ) <EOL> return RecordingTask <EOL> def install_task ( self , name , product_types = None , goal = None , required_data = None , <EOL> alternate_target_roots = None ) : <EOL> """<STR_LIT>""" <EOL> task_type = self . record ( name , product_types , required_data , alternate_target_roots ) <EOL> return super ( RoundEngineTest , <EOL> self ) . install_task ( name = name , action = task_type , goal = goal ) . task_types ( ) <EOL> def create_context ( self , for_task_types = None , target_roots = None ) : <EOL> self . _context = self . context ( for_task_types = for_task_types , target_roots = target_roots ) <EOL> self . assertTrue ( self . _context . is_unlocked ( ) ) <EOL> def assert_actions ( self , * expected_execute_ordering ) : <EOL> expected_pre_execute_actions = set ( ) <EOL> expected_execute_actions = [ ] <EOL> for action in expected_execute_ordering : <EOL> expected_pre_execute_actions . add ( self . alternate_target_roots_action ( action ) ) <EOL> expected_pre_execute_actions . add ( self . prepare_action ( action ) ) <EOL> expected_execute_actions . append ( self . construct_action ( action ) ) <EOL> expected_execute_actions . append ( self . execute_action ( action ) ) <EOL> expeceted_execute_actions_length = len ( expected_execute_ordering ) * <NUM_LIT:2> <EOL> self . assertEqual ( expected_pre_execute_actions , <EOL> set ( self . actions [ : - expeceted_execute_actions_length ] ) ) <EOL> self . assertEqual ( expected_execute_actions , self . actions [ - expeceted_execute_actions_length : ] ) <EOL> def test_lifecycle_ordering ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:1>' ] ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:2>' ] , required_data = [ '<STR_LIT:1>' ] ) <EOL> task3 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:3>' ] , required_data = [ '<STR_LIT:2>' ] ) <EOL> task4 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:1>' , '<STR_LIT:2>' , '<STR_LIT:3>' ] ) <EOL> self . create_context ( for_task_types = task1 + task2 + task3 + task4 ) <EOL> self . engine . attempt ( self . _context , self . as_goals ( '<STR_LIT>' ) ) <EOL> self . assert_actions ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_lifecycle_ordering_install_order_invariant ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:1>' ] ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:2>' ] , required_data = [ '<STR_LIT:1>' ] ) <EOL> task3 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:1>' , '<STR_LIT:2>' , '<STR_LIT:3>' ] ) <EOL> task4 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:3>' ] , required_data = [ '<STR_LIT:2>' ] ) <EOL> self . create_context ( for_task_types = task1 + task2 + task3 + task4 ) <EOL> self . engine . attempt ( self . _context , self . as_goals ( '<STR_LIT>' ) ) <EOL> self . assert_actions ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_inter_goal_dep ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:1>' ] ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:1>' ] ) <EOL> self . create_context ( for_task_types = task1 + task2 ) <EOL> self . engine . attempt ( self . _context , self . as_goals ( '<STR_LIT>' ) ) <EOL> self . assert_actions ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_inter_goal_dep_self_cycle_ok ( self ) : <EOL> task = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:1>' ] , <EOL> required_data = [ '<STR_LIT:1>' ] ) <EOL> self . create_context ( for_task_types = task ) <EOL> self . engine . attempt ( self . _context , self . as_goals ( '<STR_LIT>' ) ) <EOL> self . assert_actions ( '<STR_LIT>' ) <EOL> def test_inter_goal_dep_downstream ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:1>' ] ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:1>' ] ) <EOL> self . create_context ( for_task_types = task1 + task2 ) <EOL> with self . assertRaises ( self . engine . TaskOrderError ) : <EOL> self . engine . attempt ( self . _context , self . as_goals ( '<STR_LIT>' ) ) <EOL> def test_missing_product ( self ) : <EOL> task = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:1>' ] ) <EOL> self . create_context ( for_task_types = task ) <EOL> with self . assertRaises ( self . engine . MissingProductError ) : <EOL> self . engine . attempt ( self . _context , self . as_goals ( '<STR_LIT>' ) ) <EOL> def test_goal_cycle_direct ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:2>' ] , product_types = [ '<STR_LIT:1>' ] ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:1>' ] , product_types = [ '<STR_LIT:2>' ] ) <EOL> self . create_context ( for_task_types = task1 + task2 ) <EOL> for goal in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> with self . assertRaises ( self . engine . GoalCycleError ) : <EOL> self . engine . attempt ( self . _context , self . as_goals ( goal ) ) <EOL> def test_goal_cycle_indirect ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:2>' ] , product_types = [ '<STR_LIT:1>' ] ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:3>' ] , product_types = [ '<STR_LIT:2>' ] ) <EOL> task3 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:1>' ] , product_types = [ '<STR_LIT:3>' ] ) <EOL> self . create_context ( for_task_types = task1 + task2 + task3 ) <EOL> for goal in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> with self . assertRaises ( self . engine . GoalCycleError ) : <EOL> self . engine . attempt ( self . _context , self . as_goals ( goal ) ) <EOL> def test_goal_ordering_unconstrained_respects_cli_order ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> task3 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> self . create_context ( for_task_types = task1 + task2 + task3 ) <EOL> for permutation in itertools . permutations ( [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) : <EOL> self . actions = [ ] <EOL> self . engine . attempt ( self . _context , self . as_goals ( * [ goal for task , goal in permutation ] ) ) <EOL> expected_execute_actions = [ task for task , goal in permutation ] <EOL> self . assert_actions ( * expected_execute_actions ) <EOL> def test_goal_ordering_constrained_conflicts_cli_order ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:2>' ] ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:2>' ] ) <EOL> self . create_context ( for_task_types = task1 + task2 ) <EOL> self . engine . attempt ( self . _context , self . as_goals ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assert_actions ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_goal_ordering_mixed_constraints_and_cli_order ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> task3 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> task4 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , required_data = [ '<STR_LIT:5>' ] ) <EOL> task5 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , product_types = [ '<STR_LIT:5>' ] ) <EOL> self . create_context ( for_task_types = task1 + task2 + task3 + task4 + task5 ) <EOL> self . engine . attempt ( self . _context , <EOL> self . as_goals ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assert_actions ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_cli_goals_deduped ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> task3 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> self . create_context ( for_task_types = task1 + task2 + task3 ) <EOL> self . engine . attempt ( self . _context , <EOL> self . as_goals ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assert_actions ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_replace_target_roots ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , alternate_target_roots = [ <NUM_LIT> ] ) <EOL> self . create_context ( for_task_types = task1 + task2 ) <EOL> self . assertEquals ( [ ] , self . _context . target_roots ) <EOL> self . engine . attempt ( self . _context , self . as_goals ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertEquals ( [ <NUM_LIT> ] , self . _context . target_roots ) <EOL> def test_replace_target_roots_conflict ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , alternate_target_roots = [ <NUM_LIT> ] ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , alternate_target_roots = [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> self . create_context ( for_task_types = task1 + task2 ) <EOL> with self . assertRaises ( self . engine . TargetRootsReplacement . ConflictingProposalsError ) : <EOL> self . engine . attempt ( self . _context , self . as_goals ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_replace_target_roots_to_empty_list ( self ) : <EOL> task1 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' ) <EOL> task2 = self . install_task ( '<STR_LIT>' , goal = '<STR_LIT>' , alternate_target_roots = [ ] ) <EOL> target = self . make_target ( '<STR_LIT:t>' ) <EOL> self . create_context ( for_task_types = task1 + task2 , target_roots = [ target ] ) <EOL> self . engine . attempt ( self . _context , self . as_goals ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertEquals ( [ ] , self . _context . target_roots ) </s>
