<s> from __future__ import ( absolute_import , division , generators , nested_scopes , print_function , <EOL> unicode_literals , with_statement ) <EOL> import functools <EOL> from contextlib import contextmanager <EOL> from pants . build_graph . target import Target <EOL> from pants . task . mutex_task_mixin import MutexTaskMixin <EOL> from pants . util . contextutil import temporary_dir <EOL> from pants_test . base_test import BaseTest <EOL> class LogViewerTaskMixin ( MutexTaskMixin ) : <EOL> @ classmethod <EOL> def mutex_base ( cls ) : <EOL> return LogViewerTaskMixin <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( LogViewerTaskMixin , self ) . __init__ ( * args , ** kwargs ) <EOL> self . _executed = None <EOL> @ property <EOL> def executed ( self ) : <EOL> return self . _executed <EOL> def execute_for ( self , targets ) : <EOL> self . _executed = targets <EOL> class RedTarget ( Target ) : <EOL> pass <EOL> class RedLogViewer ( LogViewerTaskMixin ) : <EOL> options_scope = '<STR_LIT>' <EOL> @ classmethod <EOL> def select_targets ( cls , target ) : <EOL> return isinstance ( target , RedTarget ) <EOL> class BlueTarget ( Target ) : <EOL> pass <EOL> class BlueLogViewer ( LogViewerTaskMixin ) : <EOL> options_scope = '<STR_LIT>' <EOL> @ classmethod <EOL> def select_targets ( cls , target ) : <EOL> return isinstance ( target , BlueTarget ) <EOL> class GreenTarget ( Target ) : <EOL> pass <EOL> class GreenLogViewer ( LogViewerTaskMixin ) : <EOL> options_scope = '<STR_LIT>' <EOL> @ classmethod <EOL> def select_targets ( cls , target ) : <EOL> return isinstance ( target , GreenTarget ) <EOL> class MutexTaskMixinTest ( BaseTest ) : <EOL> def tearDown ( self ) : <EOL> super ( MutexTaskMixinTest , self ) . tearDown ( ) <EOL> LogViewerTaskMixin . reset_implementations ( ) <EOL> @ contextmanager <EOL> def mutex_group ( self , targets = None ) : <EOL> context = self . context ( target_roots = targets , <EOL> for_task_types = [ RedLogViewer , BlueLogViewer , GreenLogViewer ] ) <EOL> def prepare_task ( task_type ) : <EOL> task_type . prepare ( self . options , round_manager = None ) <EOL> prepare_task ( RedLogViewer ) <EOL> prepare_task ( BlueLogViewer ) <EOL> prepare_task ( GreenLogViewer ) <EOL> def create_task ( workdir , task_type ) : <EOL> return task_type ( context , workdir ) <EOL> with temporary_dir ( ) as red , temporary_dir ( ) as blue , temporary_dir ( ) as green : <EOL> red_viewer = create_task ( red , RedLogViewer ) <EOL> blue_viewer = create_task ( blue , BlueLogViewer ) <EOL> green_viewer = create_task ( green , GreenLogViewer ) <EOL> yield red_viewer , blue_viewer , green_viewer <EOL> def test_one ( self ) : <EOL> red = self . make_target ( '<STR_LIT>' , RedTarget ) <EOL> with self . mutex_group ( targets = [ red ] ) as ( red_viewer , blue_viewer , green_viewer ) : <EOL> red_viewer . execute ( ) <EOL> blue_viewer . execute ( ) <EOL> green_viewer . execute ( ) <EOL> self . assertEqual ( [ red ] , red_viewer . executed ) <EOL> self . assertIsNone ( blue_viewer . executed ) <EOL> self . assertIsNone ( green_viewer . executed ) <EOL> def assert_activation_error ( self , error_type , viewer ) : <EOL> with self . assertRaises ( error_type ) : <EOL> viewer . execute ( ) <EOL> self . assertIsNone ( viewer . executed ) <EOL> def test_none ( self ) : <EOL> assert_no_activations = functools . partial ( self . assert_activation_error , <EOL> MutexTaskMixin . NoActivationsError ) <EOL> with self . mutex_group ( ) as ( red_viewer , blue_viewer , green_viewer ) : <EOL> assert_no_activations ( red_viewer ) <EOL> assert_no_activations ( blue_viewer ) <EOL> assert_no_activations ( green_viewer ) <EOL> def assert_incompatible_activations ( self , viewer ) : <EOL> self . assert_activation_error ( MutexTaskMixin . IncompatibleActivationsError , viewer ) <EOL> def test_some_incompatible ( self ) : <EOL> red = self . make_target ( '<STR_LIT>' , RedTarget ) <EOL> blue = self . make_target ( '<STR_LIT>' , BlueTarget ) <EOL> with self . mutex_group ( targets = [ red , blue ] ) as ( red_viewer , blue_viewer , green_viewer ) : <EOL> self . assert_incompatible_activations ( red_viewer ) <EOL> self . assert_incompatible_activations ( blue_viewer ) <EOL> green_viewer . execute ( ) <EOL> self . assertIsNone ( green_viewer . executed ) <EOL> def test_all_incompatible ( self ) : <EOL> red = self . make_target ( '<STR_LIT>' , RedTarget ) <EOL> blue = self . make_target ( '<STR_LIT>' , BlueTarget ) <EOL> green = self . make_target ( '<STR_LIT>' , GreenTarget ) <EOL> with self . mutex_group ( targets = [ red , blue , green ] ) as ( red_viewer , blue_viewer , green_viewer ) : <EOL> self . assert_incompatible_activations ( red_viewer ) <EOL> self . assert_incompatible_activations ( blue_viewer ) <EOL> self . assert_incompatible_activations ( green_viewer ) </s>
