<s> """<STR_LIT>""" <EOL> import sys <EOL> from testtools import clone_test_with_new_id , iterate_tests <EOL> from twisted . python . usage import Options , UsageError <EOL> from twisted . trial . reporter import TreeReporter <EOL> from twisted . trial . runner import TestLoader <EOL> from flocker . testtools . _flaky import _get_flaky_annotation <EOL> def _load_tests ( name ) : <EOL> """<STR_LIT>""" <EOL> loader = TestLoader ( ) <EOL> return loader . loadByName ( name , recurse = True ) <EOL> def _iter_tests ( names ) : <EOL> """<STR_LIT>""" <EOL> for name in names : <EOL> suite = _load_tests ( name ) <EOL> for test in iterate_tests ( suite ) : <EOL> yield test <EOL> def get_flaky_annotation ( case ) : <EOL> """<STR_LIT>""" <EOL> error = getattr ( case , '<STR_LIT:error>' , None ) <EOL> if error : <EOL> raise error [ <NUM_LIT:0> ] , error [ <NUM_LIT:1> ] , error [ <NUM_LIT:2> ] <EOL> return _get_flaky_annotation ( case ) <EOL> def find_flaky_tests ( suites ) : <EOL> """<STR_LIT>""" <EOL> for test in _iter_tests ( suites ) : <EOL> annotation = get_flaky_annotation ( test ) <EOL> if annotation : <EOL> yield test , annotation <EOL> def report_flaky_tests ( output , flaky_tests ) : <EOL> """<STR_LIT>""" <EOL> for test in flaky_tests : <EOL> output . write ( '<STR_LIT>' . format ( test ) ) <EOL> def report_issues ( output , flaky_tests ) : <EOL> """<STR_LIT>""" <EOL> jira_keys = frozenset ( ) . union ( <EOL> * ( flaky . jira_keys for ( _ , flaky ) in flaky_tests ) ) <EOL> for jira_key in sorted ( jira_keys ) : <EOL> output . write ( '<STR_LIT>' . format ( jira_key ) ) <EOL> def report_tests ( output , flaky_tests ) : <EOL> """<STR_LIT>""" <EOL> tests = list ( test . id ( ) for ( test , _ ) in flaky_tests ) <EOL> for test in sorted ( tests ) : <EOL> output . write ( '<STR_LIT>' . format ( test ) ) <EOL> def report_test_tree ( output , flaky_tests ) : <EOL> """<STR_LIT>""" <EOL> reporter = TreeReporter ( output ) <EOL> for ( test , flaky ) in flaky_tests : <EOL> new_test = clone_test_with_new_id ( <EOL> test , '<STR_LIT>' . format ( test . id ( ) , '<STR_LIT:U+002CU+0020>' . join ( flaky . jira_keys ) ) ) <EOL> reporter . startTest ( new_test ) <EOL> reporter . addSuccess ( new_test ) <EOL> reporter . stopTest ( new_test ) <EOL> reporter . done ( ) <EOL> _REPORTS = { <EOL> '<STR_LIT>' : report_issues , <EOL> '<STR_LIT>' : report_tests , <EOL> '<STR_LIT>' : report_test_tree , <EOL> } <EOL> """<STR_LIT>""" <EOL> def _to_report ( name ) : <EOL> """<STR_LIT>""" <EOL> if name in _REPORTS : <EOL> return name <EOL> raise UsageError ( <EOL> "<STR_LIT>" . format ( name , _to_report . coerceDoc ) ) <EOL> _to_report . coerceDoc = "<STR_LIT>" . format ( '<STR_LIT:U+002CU+0020>' . join ( _REPORTS . keys ( ) ) ) <EOL> class FindFlakyTestsOptions ( Options ) : <EOL> """<STR_LIT>""" <EOL> optParameters = [ <EOL> [ '<STR_LIT>' , '<STR_LIT:r>' , '<STR_LIT>' , '<STR_LIT>' , _to_report ] <EOL> ] <EOL> def __init__ ( self , name , * args , ** kwargs ) : <EOL> super ( FindFlakyTestsOptions , self ) . __init__ ( * args , ** kwargs ) <EOL> self . _name = name <EOL> self . synopsis = "<STR_LIT>" . format ( name ) <EOL> def parseArgs ( self , * suites ) : <EOL> """<STR_LIT>""" <EOL> if not suites : <EOL> raise UsageError ( '<STR_LIT>' ) <EOL> self [ '<STR_LIT>' ] = suites <EOL> def find_flaky_tests_main ( args , base_path , top_level , stdout = None , <EOL> stderr = None ) : <EOL> """<STR_LIT>""" <EOL> stdout = sys . stdout if stdout is None else stdout <EOL> stderr = sys . stderr if stderr is None else stderr <EOL> options = FindFlakyTestsOptions ( base_path . basename ( ) ) <EOL> try : <EOL> options . parseOptions ( args ) <EOL> except UsageError as e : <EOL> stdout . write ( str ( options ) ) <EOL> stdout . write ( "<STR_LIT:\n>" ) <EOL> stderr . write ( "<STR_LIT>" . format ( base_path . basename ( ) , e ) ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> flaky_tests = find_flaky_tests ( options [ '<STR_LIT>' ] ) <EOL> _REPORTS [ options [ '<STR_LIT>' ] ] ( stdout , flaky_tests ) </s>
