<s> """<STR_LIT>""" <EOL> import heapq <EOL> import subunit <EOL> import sys <EOL> import unittest <EOL> import testtools <EOL> class _AnsiColorizer ( object ) : <EOL> """<STR_LIT>""" <EOL> _colors = dict ( black = <NUM_LIT:30> , red = <NUM_LIT> , green = <NUM_LIT:32> , yellow = <NUM_LIT> , <EOL> blue = <NUM_LIT> , magenta = <NUM_LIT> , cyan = <NUM_LIT> , white = <NUM_LIT> ) <EOL> def __init__ ( self , stream ) : <EOL> self . stream = stream <EOL> @ staticmethod <EOL> def supported ( stream = sys . stdout ) : <EOL> """<STR_LIT>""" <EOL> if not stream . isatty ( ) : <EOL> return False <EOL> try : <EOL> import curses <EOL> except ImportError : <EOL> return False <EOL> else : <EOL> try : <EOL> try : <EOL> return curses . tigetnum ( "<STR_LIT>" ) > <NUM_LIT:2> <EOL> except curses . error : <EOL> curses . setupterm ( ) <EOL> return curses . tigetnum ( "<STR_LIT>" ) > <NUM_LIT:2> <EOL> except Exception : <EOL> return False <EOL> def write ( self , text , color ) : <EOL> """<STR_LIT>""" <EOL> color = self . _colors [ color ] <EOL> self . stream . write ( '<STR_LIT>' % ( color , text ) ) <EOL> class _Win32Colorizer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , stream ) : <EOL> import win32console <EOL> red , green , blue , bold = ( win32console . FOREGROUND_RED , <EOL> win32console . FOREGROUND_GREEN , <EOL> win32console . FOREGROUND_BLUE , <EOL> win32console . FOREGROUND_INTENSITY ) <EOL> self . stream = stream <EOL> self . screenBuffer = win32console . GetStdHandle ( <EOL> win32console . STD_OUT_HANDLE ) <EOL> self . _colors = { <EOL> '<STR_LIT>' : red | green | blue , <EOL> '<STR_LIT>' : red | bold , <EOL> '<STR_LIT>' : green | bold , <EOL> '<STR_LIT>' : blue | bold , <EOL> '<STR_LIT>' : red | green | bold , <EOL> '<STR_LIT>' : red | blue | bold , <EOL> '<STR_LIT>' : green | blue | bold , <EOL> '<STR_LIT>' : red | green | blue | bold <EOL> } <EOL> @ staticmethod <EOL> def supported ( stream = sys . stdout ) : <EOL> try : <EOL> import win32console <EOL> screenBuffer = win32console . GetStdHandle ( <EOL> win32console . STD_OUT_HANDLE ) <EOL> except ImportError : <EOL> return False <EOL> import pywintypes <EOL> try : <EOL> screenBuffer . SetConsoleTextAttribute ( <EOL> win32console . FOREGROUND_RED | <EOL> win32console . FOREGROUND_GREEN | <EOL> win32console . FOREGROUND_BLUE ) <EOL> except pywintypes . error : <EOL> return False <EOL> else : <EOL> return True <EOL> def write ( self , text , color ) : <EOL> color = self . _colors [ color ] <EOL> self . screenBuffer . SetConsoleTextAttribute ( color ) <EOL> self . stream . write ( text ) <EOL> self . screenBuffer . SetConsoleTextAttribute ( self . _colors [ '<STR_LIT>' ] ) <EOL> class _NullColorizer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , stream ) : <EOL> self . stream = stream <EOL> @ staticmethod <EOL> def supported ( stream = sys . stdout ) : <EOL> return True <EOL> def write ( self , text , color ) : <EOL> self . stream . write ( text ) <EOL> def get_elapsed_time_color ( elapsed_time ) : <EOL> if elapsed_time > <NUM_LIT:1.0> : <EOL> return '<STR_LIT>' <EOL> elif elapsed_time > <NUM_LIT> : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return '<STR_LIT>' <EOL> class SubunitTestResult ( testtools . TestResult ) : <EOL> def __init__ ( self , stream , descriptions , verbosity ) : <EOL> super ( SubunitTestResult , self ) . __init__ ( ) <EOL> self . stream = stream <EOL> self . showAll = verbosity > <NUM_LIT:1> <EOL> self . num_slow_tests = <NUM_LIT:10> <EOL> self . slow_tests = [ ] <EOL> self . colorizer = None <EOL> stdout = sys . stdout <EOL> sys . stdout = sys . __stdout__ <EOL> for colorizer in [ _Win32Colorizer , _AnsiColorizer , _NullColorizer ] : <EOL> if colorizer . supported ( ) : <EOL> self . colorizer = colorizer ( self . stream ) <EOL> break <EOL> sys . stdout = stdout <EOL> self . start_time = None <EOL> self . last_time = { } <EOL> self . results = { } <EOL> self . last_written = None <EOL> def _writeElapsedTime ( self , elapsed ) : <EOL> color = get_elapsed_time_color ( elapsed ) <EOL> self . colorizer . write ( "<STR_LIT>" % elapsed , color ) <EOL> def _addResult ( self , test , * args ) : <EOL> try : <EOL> name = test . id ( ) <EOL> except AttributeError : <EOL> name = '<STR_LIT>' <EOL> test_class , test_name = name . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> elapsed = ( self . _now ( ) - self . start_time ) . total_seconds ( ) <EOL> item = ( elapsed , test_class , test_name ) <EOL> if len ( self . slow_tests ) >= self . num_slow_tests : <EOL> heapq . heappushpop ( self . slow_tests , item ) <EOL> else : <EOL> heapq . heappush ( self . slow_tests , item ) <EOL> self . results . setdefault ( test_class , [ ] ) <EOL> self . results [ test_class ] . append ( ( test_name , elapsed ) + args ) <EOL> self . last_time [ test_class ] = self . _now ( ) <EOL> self . writeTests ( ) <EOL> def _writeResult ( self , test_name , elapsed , long_result , color , <EOL> short_result , success ) : <EOL> if self . showAll : <EOL> self . stream . write ( '<STR_LIT>' % str ( test_name ) . ljust ( <NUM_LIT> ) ) <EOL> self . colorizer . write ( long_result , color ) <EOL> if success : <EOL> self . _writeElapsedTime ( elapsed ) <EOL> self . stream . writeln ( ) <EOL> else : <EOL> self . colorizer . write ( short_result , color ) <EOL> def addSuccess ( self , test ) : <EOL> super ( SubunitTestResult , self ) . addSuccess ( test ) <EOL> self . _addResult ( test , '<STR_LIT:OK>' , '<STR_LIT>' , '<STR_LIT:.>' , True ) <EOL> def addFailure ( self , test , err ) : <EOL> if test . id ( ) == '<STR_LIT>' : <EOL> return <EOL> super ( SubunitTestResult , self ) . addFailure ( test , err ) <EOL> self . _addResult ( test , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:F>' , False ) <EOL> def addError ( self , test , err ) : <EOL> super ( SubunitTestResult , self ) . addFailure ( test , err ) <EOL> self . _addResult ( test , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:E>' , False ) <EOL> def addSkip ( self , test , reason = None , details = None ) : <EOL> super ( SubunitTestResult , self ) . addSkip ( test , reason , details ) <EOL> self . _addResult ( test , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:S>' , True ) <EOL> def startTest ( self , test ) : <EOL> self . start_time = self . _now ( ) <EOL> super ( SubunitTestResult , self ) . startTest ( test ) <EOL> def writeTestCase ( self , cls ) : <EOL> if not self . results . get ( cls ) : <EOL> return <EOL> if cls != self . last_written : <EOL> self . colorizer . write ( cls , '<STR_LIT>' ) <EOL> self . stream . writeln ( ) <EOL> for result in self . results [ cls ] : <EOL> self . _writeResult ( * result ) <EOL> del self . results [ cls ] <EOL> self . stream . flush ( ) <EOL> self . last_written = cls <EOL> def writeTests ( self ) : <EOL> time = self . last_time . get ( self . last_written , self . _now ( ) ) <EOL> if not self . last_written or ( self . _now ( ) - time ) . total_seconds ( ) > <NUM_LIT> : <EOL> diff = <NUM_LIT> <EOL> while diff > <NUM_LIT> : <EOL> classes = self . results . keys ( ) <EOL> oldest = min ( classes , key = lambda x : self . last_time [ x ] ) <EOL> diff = ( self . _now ( ) - self . last_time [ oldest ] ) . total_seconds ( ) <EOL> self . writeTestCase ( oldest ) <EOL> else : <EOL> self . writeTestCase ( self . last_written ) <EOL> def done ( self ) : <EOL> self . stopTestRun ( ) <EOL> def stopTestRun ( self ) : <EOL> for cls in list ( self . results . iterkeys ( ) ) : <EOL> self . writeTestCase ( cls ) <EOL> self . stream . writeln ( ) <EOL> self . writeSlowTests ( ) <EOL> def writeSlowTests ( self ) : <EOL> slow_tests = [ item for item in self . slow_tests <EOL> if get_elapsed_time_color ( item [ <NUM_LIT:0> ] ) != '<STR_LIT>' ] <EOL> if slow_tests : <EOL> slow_total_time = sum ( item [ <NUM_LIT:0> ] for item in slow_tests ) <EOL> slow = ( "<STR_LIT>" <EOL> % ( len ( slow_tests ) , slow_total_time ) ) <EOL> self . colorizer . write ( slow , '<STR_LIT>' ) <EOL> self . stream . writeln ( ) <EOL> last_cls = None <EOL> for elapsed , cls , name in sorted ( slow_tests , <EOL> key = lambda x : x [ <NUM_LIT:1> ] + x [ <NUM_LIT:2> ] ) : <EOL> if cls != last_cls : <EOL> self . colorizer . write ( cls , '<STR_LIT>' ) <EOL> self . stream . writeln ( ) <EOL> last_cls = cls <EOL> self . stream . write ( '<STR_LIT>' % str ( name ) . ljust ( <NUM_LIT> ) ) <EOL> self . _writeElapsedTime ( elapsed ) <EOL> self . stream . writeln ( ) <EOL> def printErrors ( self ) : <EOL> if self . showAll : <EOL> self . stream . writeln ( ) <EOL> self . printErrorList ( '<STR_LIT>' , self . errors ) <EOL> self . printErrorList ( '<STR_LIT>' , self . failures ) <EOL> def printErrorList ( self , flavor , errors ) : <EOL> for test , err in errors : <EOL> self . colorizer . write ( "<STR_LIT:=>" * <NUM_LIT> , '<STR_LIT>' ) <EOL> self . stream . writeln ( ) <EOL> self . colorizer . write ( flavor , '<STR_LIT>' ) <EOL> self . stream . writeln ( "<STR_LIT>" % test . id ( ) ) <EOL> self . colorizer . write ( "<STR_LIT:->" * <NUM_LIT> , '<STR_LIT>' ) <EOL> self . stream . writeln ( ) <EOL> self . stream . writeln ( "<STR_LIT:%s>" % err ) <EOL> test = subunit . ProtocolTestCase ( sys . stdin , passthrough = None ) <EOL> if sys . version_info [ <NUM_LIT:0> : <NUM_LIT:2> ] <= ( <NUM_LIT:2> , <NUM_LIT:6> ) : <EOL> runner = unittest . TextTestRunner ( verbosity = <NUM_LIT:2> ) <EOL> else : <EOL> runner = unittest . TextTestRunner ( <EOL> verbosity = <NUM_LIT:2> , resultclass = SubunitTestResult ) <EOL> if runner . run ( test ) . wasSuccessful ( ) : <EOL> exit_code = <NUM_LIT:0> <EOL> else : <EOL> exit_code = <NUM_LIT:1> <EOL> sys . exit ( exit_code ) </s>
