<s> from helpers import LuigiTestCase , with_config <EOL> import mock <EOL> import luigi <EOL> import luigi . scheduler <EOL> from luigi . cmdline import luigi_run <EOL> class RetcodesTest ( LuigiTestCase ) : <EOL> def run_and_expect ( self , joined_params , retcode , extra_args = [ '<STR_LIT>' , '<STR_LIT>' ] ) : <EOL> with self . assertRaises ( SystemExit ) as cm : <EOL> luigi_run ( ( joined_params . split ( '<STR_LIT:U+0020>' ) + extra_args ) ) <EOL> self . assertEqual ( cm . exception . code , retcode ) <EOL> def run_with_config ( self , retcode_config , * args , ** kwargs ) : <EOL> with_config ( dict ( retcode = retcode_config ) ) ( self . run_and_expect ) ( * args , ** kwargs ) <EOL> def test_task_failed ( self ) : <EOL> class FailingTask ( luigi . Task ) : <EOL> def run ( self ) : <EOL> raise ValueError ( ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . run_with_config ( dict ( task_failed = '<STR_LIT:3>' ) , '<STR_LIT>' , <NUM_LIT:3> ) <EOL> def test_missing_data ( self ) : <EOL> class MissingDataTask ( luigi . ExternalTask ) : <EOL> def complete ( self ) : <EOL> return False <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . run_with_config ( dict ( missing_data = '<STR_LIT:3>' ) , '<STR_LIT>' , <NUM_LIT:3> ) <EOL> def test_already_running ( self ) : <EOL> class AlreadyRunningTask ( luigi . Task ) : <EOL> def run ( self ) : <EOL> pass <EOL> old_func = luigi . scheduler . CentralPlannerScheduler . get_work <EOL> def new_func ( * args , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = None <EOL> old_func ( * args , ** kwargs ) <EOL> res = old_func ( * args , ** kwargs ) <EOL> res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> return res <EOL> with mock . patch ( '<STR_LIT>' , new_func ) : <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . run_with_config ( dict ( already_running = '<STR_LIT:3>' ) , '<STR_LIT>' , <NUM_LIT:3> ) <EOL> def test_when_locked ( self ) : <EOL> def new_func ( * args , ** kwargs ) : <EOL> return False <EOL> with mock . patch ( '<STR_LIT>' , new_func ) : <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:0> , extra_args = [ '<STR_LIT>' ] ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:5> , extra_args = [ '<STR_LIT>' ] ) <EOL> self . run_with_config ( dict ( already_running = '<STR_LIT:3>' ) , '<STR_LIT>' , <NUM_LIT:3> , extra_args = [ '<STR_LIT>' ] ) <EOL> def test_failure_in_complete ( self ) : <EOL> class FailingComplete ( luigi . Task ) : <EOL> def complete ( self ) : <EOL> raise Exception <EOL> class RequiringTask ( luigi . Task ) : <EOL> def requires ( self ) : <EOL> yield FailingComplete ( ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> def test_failure_in_requires ( self ) : <EOL> class FailingRequires ( luigi . Task ) : <EOL> def requires ( self ) : <EOL> raise Exception <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> def test_validate_dependency_error ( self ) : <EOL> class DependencyTask ( object ) : <EOL> pass <EOL> class RequiringTask ( luigi . Task ) : <EOL> def requires ( self ) : <EOL> yield DependencyTask ( ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:4> ) <EOL> def test_unhandled_exception ( self ) : <EOL> def new_func ( * args , ** kwargs ) : <EOL> raise Exception ( ) <EOL> with mock . patch ( '<STR_LIT>' , new_func ) : <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:4> ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:2> ) <EOL> class TaskWithRequiredParam ( luigi . Task ) : <EOL> param = luigi . Parameter ( ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:4> ) <EOL> def test_when_mixed_errors ( self ) : <EOL> class FailingTask ( luigi . Task ) : <EOL> def run ( self ) : <EOL> raise ValueError ( ) <EOL> class MissingDataTask ( luigi . ExternalTask ) : <EOL> def complete ( self ) : <EOL> return False <EOL> class RequiringTask ( luigi . Task ) : <EOL> def requires ( self ) : <EOL> yield FailingTask ( ) <EOL> yield MissingDataTask ( ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . run_and_expect ( '<STR_LIT>' , <NUM_LIT:7> ) </s>
