#NEWLINE# def poolRunner( target , queue , coverage_number = None , omit_patterns = [ ] ) : #NEWLINE# #INDENT# '\n    I am the function that pool worker processes run.  I run one unit test.\n    ' #NEWLINE# saved_tempdir = tempfile . tempdir #NEWLINE# tempfile . tempdir = tempfile . mkdtemp ( ) #NEWLINE# def cleanup( ) : #NEWLINE# #INDENT# shutil . rmtree ( tempfile . tempdir , ignore_errors = True ) #NEWLINE# tempfile . tempdir = saved_tempdir #NEWLINE# queue . put ( None ) #NEWLINE# if ( coverage_number and coverage ) : #NEWLINE# #INDENT# cov . stop ( ) #NEWLINE# cov . save ( ) #NEWLINE# #UNINDENT# #UNINDENT# if ( coverage_number and coverage ) : #NEWLINE# #INDENT# cov = coverage . coverage ( data_file = '.coverage.{}_{}' . format ( coverage_number , random . randint ( 0 , 10000 ) ) , omit = omit_patterns ) #NEWLINE# cov . _warn_no_data = False #NEWLINE# cov . start ( ) #NEWLINE# #UNINDENT# already_sent = set ( ) #NEWLINE# def start_callback( test ) : #NEWLINE# #INDENT# test = proto_test ( test ) #NEWLINE# if ( test not in already_sent ) : #NEWLINE# #INDENT# queue . put ( test ) #NEWLINE# already_sent . add ( test ) #NEWLINE# #UNINDENT# #UNINDENT# def finalize_callback( test_result ) : #NEWLINE# #INDENT# queue . put ( test_result ) #NEWLINE# #UNINDENT# result = ProtoTestResult ( start_callback , finalize_callback ) #NEWLINE# test = None #NEWLINE# try : #NEWLINE# #INDENT# test = loadTargets ( target ) #NEWLINE# #UNINDENT# except : #NEWLINE# #INDENT# err = sys . exc_info ( ) #NEWLINE# t = ProtoTest ( ) #NEWLINE# t . module = 'green.loader' #NEWLINE# t . class_name = 'N/A' #NEWLINE# t . description = 'Green encountered an error loading the unit test.' #NEWLINE# t . method_name = 'poolRunner' #NEWLINE# result . startTest ( t ) #NEWLINE# result . addError ( t , err ) #NEWLINE# result . stopTest ( t ) #NEWLINE# queue . put ( t ) #NEWLINE# queue . put ( result ) #NEWLINE# cleanup ( ) #NEWLINE# return #NEWLINE# #UNINDENT# if getattr ( test , 'run' , False ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# test . run ( result ) #NEWLINE# #UNINDENT# except : #NEWLINE# #INDENT# if result . errors : #NEWLINE# #INDENT# queue . put ( result ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# err = sys . exc_info ( ) #NEWLINE# result . startTest ( test ) #NEWLINE# result . addError ( test , err ) #NEWLINE# result . stopTest ( test ) #NEWLINE# queue . put ( result ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# description = 'Test loader returned an un-runnable object.  Is "{}" importable from your current location?  Maybe you forgot an __init__.py in your directory?  Unrunnable object looks like: {} of type {} with dir {}' . format ( target , str ( test ) , type ( test ) , dir ( test ) ) #NEWLINE# err = ( TypeError , TypeError ( description ) , None ) #NEWLINE# t = ProtoTest ( ) #NEWLINE# target_list = target . split ( '.' ) #NEWLINE# t . module = ( '.' . join ( target_list [ : ( - 2 ) ] ) if ( len ( target_list ) > 1 ) else target ) #NEWLINE# t . class_name = ( target . split ( '.' ) [ ( - 2 ) ] if ( len ( target_list ) > 1 ) else 'UnknownClass' ) #NEWLINE# t . description = description #NEWLINE# t . method_name = ( target . split ( '.' ) [ ( - 1 ) ] if ( len ( target_list ) > 1 ) else 'unknown_method' ) #NEWLINE# result . startTest ( t ) #NEWLINE# result . addError ( t , err ) #NEWLINE# result . stopTest ( t ) #NEWLINE# #UNINDENT# cleanup ( )