<s> import random <EOL> from collections import defaultdict <EOL> from unittest import TestCase <EOL> from spockbot . plugins . core . taskmanager import TaskManager <EOL> from spockbot . plugins . tools . task import Task , TaskFailed <EOL> class EventMock ( object ) : <EOL> def __init__ ( self ) : <EOL> self . handlers = defaultdict ( list ) <EOL> def reg_event_handler ( self , evt , handler ) : <EOL> self . handlers [ evt ] . append ( handler ) <EOL> def emit ( self , event , data ) : <EOL> to_remove = [ ] <EOL> for handler in reversed ( self . handlers [ event ] ) : <EOL> if handler ( event , data ) : <EOL> to_remove . append ( handler ) <EOL> else : <EOL> raise AssertionError ( '<STR_LIT>' % ( handler . __name__ , handler ) ) <EOL> for handler in to_remove : <EOL> self . handlers [ event ] . remove ( handler ) <EOL> class PluginLoaderMock ( object ) : <EOL> def provides ( self , ident , obj ) : <EOL> self . provides_ident = ident <EOL> self . provides_obj = obj <EOL> def requires ( self , plug_name ) : <EOL> assert plug_name == '<STR_LIT>' , '<STR_LIT>' % plug_name <EOL> return EventMock ( ) <EOL> class TaskManagerTest ( TestCase ) : <EOL> def setUp ( self ) : <EOL> ploader = PluginLoaderMock ( ) <EOL> self . task_manager = TaskManager ( ploader , { } ) <EOL> assert ploader . provides_ident == '<STR_LIT>' <EOL> assert isinstance ( ploader . provides_obj , TaskManager ) <EOL> def test_run_task ( self ) : <EOL> def some_task ( level ) : <EOL> if level <= <NUM_LIT:0> : <EOL> raise StopIteration ( '<STR_LIT>' , level ) <EOL> event , data = yield '<STR_LIT>' <EOL> self . assertEqual ( '<STR_LIT>' , event ) <EOL> self . assertDictEqual ( { '<STR_LIT>' : '<STR_LIT:data>' } , data ) <EOL> event , data = yield '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> self . assertEqual ( '<STR_LIT>' , event ) <EOL> self . assertDictEqual ( { '<STR_LIT>' : '<STR_LIT:data>' } , data ) <EOL> event , data = yield '<STR_LIT>' , lambda e , d : True <EOL> self . assertEqual ( '<STR_LIT>' , event ) <EOL> self . assertDictEqual ( { '<STR_LIT>' : '<STR_LIT:data>' } , data ) <EOL> event , data = yield some_task ( level - <NUM_LIT:1> ) <EOL> self . assertEqual ( None , event ) <EOL> self . assertTupleEqual ( ( '<STR_LIT>' , level - <NUM_LIT:1> ) , data ) <EOL> raise StopIteration ( '<STR_LIT>' , level ) <EOL> def emit_them ( level ) : <EOL> """<STR_LIT>""" <EOL> if level <= <NUM_LIT:0> : <EOL> return <EOL> self . assertIn ( '<STR_LIT>' , self . task_manager . event . handlers , <EOL> '<STR_LIT>' ) <EOL> self . task_manager . event . emit ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT:data>' } ) <EOL> self . assertFalse ( self . task_manager . event . handlers [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , self . task_manager . event . handlers , <EOL> '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , self . task_manager . event . handlers , <EOL> '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , self . task_manager . event . handlers , <EOL> '<STR_LIT>' ) <EOL> self . task_manager . event . emit ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT:data>' } ) <EOL> self . assertFalse ( self . task_manager . event . handlers [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , self . task_manager . event . handlers , <EOL> '<STR_LIT>' ) <EOL> self . task_manager . event . emit ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT:data>' } ) <EOL> self . assertFalse ( self . task_manager . event . handlers [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> emit_them ( level - <NUM_LIT:1> ) <EOL> done = [ False ] <EOL> test_case = self <EOL> class SomeParent ( object ) : <EOL> def on_success ( self , data ) : <EOL> done [ <NUM_LIT:0> ] = True <EOL> def on_error ( self , e ) : <EOL> test_case . fail ( '<STR_LIT>' <EOL> % ( e . __class__ . __name__ , e . args ) ) <EOL> levels = <NUM_LIT:3> <EOL> task = some_task ( levels ) <EOL> parent = SomeParent ( ) <EOL> ret = self . task_manager . run_task ( task , parent ) <EOL> self . assertIsInstance ( ret , Task ) <EOL> emit_them ( levels ) <EOL> assert done [ <NUM_LIT:0> ] <EOL> def test_failure ( self ) : <EOL> last_data = [ '<STR_LIT>' ] <EOL> def task_with_failure ( ) : <EOL> last_data [ <NUM_LIT:0> ] = yield '<STR_LIT>' <EOL> raise TaskFailed ( '<STR_LIT>' ) <EOL> def task_deep_failure ( ) : <EOL> def task_deepest_failure ( ) : <EOL> last_data [ <NUM_LIT:0> ] = yield '<STR_LIT>' <EOL> raise TaskFailed ( '<STR_LIT>' ) <EOL> def task_deeper_failure ( ) : <EOL> yield task_deepest_failure ( ) <EOL> try : <EOL> yield task_deeper_failure ( ) <EOL> except TaskFailed as error : <EOL> raise TaskFailed ( '<STR_LIT>' ) . with_error ( error ) <EOL> def task_with_exception ( ) : <EOL> last_data [ <NUM_LIT:0> ] = yield '<STR_LIT>' <EOL> <NUM_LIT:1> / <NUM_LIT:0> <EOL> def top_task ( ) : <EOL> try : <EOL> yield task_with_failure ( ) <EOL> except TaskFailed as e : <EOL> self . assertEqual ( '<STR_LIT>' , e . message ) <EOL> self . assertEqual ( None , e . prev_error ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( e . tasktrace ) ) <EOL> self . assertEqual ( '<STR_LIT>' , e . tasktrace [ <NUM_LIT:0> ] . name ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( e . full_tasktrace ) ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> e . full_tasktrace [ <NUM_LIT:0> ] . name ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> try : <EOL> yield task_deep_failure ( ) <EOL> except TaskFailed as e : <EOL> self . assertEqual ( '<STR_LIT>' , e . message ) <EOL> self . assertEqual ( '<STR_LIT>' , e . prev_error . message ) <EOL> self . assertEqual ( None , e . prev_error . prev_error ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( e . tasktrace ) ) <EOL> self . assertEqual ( '<STR_LIT>' , e . tasktrace [ <NUM_LIT:0> ] . name ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( e . full_tasktrace ) ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> e . full_tasktrace [ <NUM_LIT:0> ] . name ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> e . full_tasktrace [ <NUM_LIT:1> ] . name ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> e . full_tasktrace [ <NUM_LIT:2> ] . name ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> try : <EOL> yield task_with_exception ( ) <EOL> except ZeroDivisionError as e : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> task = top_task ( ) <EOL> ret = self . task_manager . run_task ( task ) <EOL> self . assertIsInstance ( ret , Task ) <EOL> self . assertEqual ( '<STR_LIT>' , last_data [ <NUM_LIT:0> ] ) <EOL> self . emit_and_check ( '<STR_LIT>' , last_data ) <EOL> self . emit_and_check ( '<STR_LIT>' , last_data ) <EOL> self . emit_and_check ( '<STR_LIT>' , last_data ) <EOL> def emit_and_check ( self , event , last_data ) : <EOL> data = random . random ( ) <EOL> self . task_manager . event . emit ( event , data ) <EOL> self . assertEqual ( event , last_data [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( data , last_data [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) </s>
