<s> from oslo_utils import encodeutils <EOL> from subunit import v2 <EOL> def total_seconds ( td ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( td , "<STR_LIT>" ) : <EOL> s = td . total_seconds ( ) <EOL> else : <EOL> s = ( td . microseconds + <EOL> ( td . seconds + td . days * <NUM_LIT> * <NUM_LIT> ) * <NUM_LIT:10> ** <NUM_LIT:6> ) / <NUM_LIT> ** <NUM_LIT:6> <EOL> return "<STR_LIT>" % s <EOL> def preparse_input_args ( func ) : <EOL> def inner ( self , test_id = None , test_status = None , test_tags = None , <EOL> runnable = True , file_name = None , file_bytes = None , eof = False , <EOL> mime_type = None , route_code = None , timestamp = None ) : <EOL> if not test_id : <EOL> return <EOL> if ( test_id . startswith ( "<STR_LIT>" ) or <EOL> test_id . startswith ( "<STR_LIT>" ) ) : <EOL> test_id = test_id [ test_id . find ( "<STR_LIT:(>" ) + <NUM_LIT:1> : - <NUM_LIT:1> ] <EOL> if test_id . find ( "<STR_LIT:[>" ) > - <NUM_LIT:1> : <EOL> tags = test_id . split ( "<STR_LIT:[>" ) [ <NUM_LIT:1> ] [ : - <NUM_LIT:1> ] . split ( "<STR_LIT:U+002C>" ) <EOL> else : <EOL> tags = [ ] <EOL> if mime_type : <EOL> mime_type , charset = mime_type . split ( "<STR_LIT>" ) [ : <NUM_LIT:2> ] <EOL> charset = charset . split ( "<STR_LIT:=>" ) [ <NUM_LIT:1> ] <EOL> else : <EOL> charset = None <EOL> func ( self , test_id , test_status , tags , file_name , file_bytes , <EOL> mime_type , timestamp , charset ) <EOL> return inner <EOL> class SubunitV2StreamResult ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , expected_failures = None ) : <EOL> self . _tests = { } <EOL> self . _expected_failures = expected_failures or { } <EOL> self . _timestamps = { } <EOL> self . _first_timestamp = None <EOL> self . _last_timestamp = None <EOL> self . _unknown_entities = { } <EOL> self . _is_parsed = False <EOL> def _post_parse ( self ) : <EOL> for test_id in self . _unknown_entities : <EOL> matcher = lambda i : i == test_id or i . startswith ( "<STR_LIT>" % test_id ) <EOL> known_ids = filter ( matcher , self . _tests ) <EOL> for id_ in known_ids : <EOL> if self . _tests [ id_ ] [ "<STR_LIT:status>" ] == "<STR_LIT>" : <EOL> self . _tests [ id_ ] [ "<STR_LIT:status>" ] = ( <EOL> self . _unknown_entities [ test_id ] [ "<STR_LIT:status>" ] ) <EOL> if self . _unknown_entities [ test_id ] . get ( "<STR_LIT>" ) : <EOL> self . _tests [ id_ ] [ "<STR_LIT>" ] = ( <EOL> self . _unknown_entities [ test_id ] [ "<STR_LIT>" ] ) <EOL> elif self . _unknown_entities [ test_id ] . get ( "<STR_LIT>" ) : <EOL> self . _tests [ id_ ] [ "<STR_LIT>" ] = ( <EOL> self . _unknown_entities [ test_id ] [ "<STR_LIT>" ] ) <EOL> for test_id in self . _expected_failures : <EOL> if self . _tests . get ( test_id ) : <EOL> if self . _tests [ test_id ] [ "<STR_LIT:status>" ] == "<STR_LIT>" : <EOL> self . _tests [ test_id ] [ "<STR_LIT:status>" ] = "<STR_LIT>" <EOL> if self . _expected_failures [ test_id ] : <EOL> self . _tests [ test_id ] [ "<STR_LIT>" ] = ( <EOL> self . _expected_failures [ test_id ] ) <EOL> elif self . _tests [ test_id ] [ "<STR_LIT:status>" ] == "<STR_LIT:success>" : <EOL> self . _tests [ test_id ] [ "<STR_LIT:status>" ] = "<STR_LIT>" <EOL> for test_id in self . _tests : <EOL> for file_name in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> if file_name in self . _tests [ test_id ] : <EOL> self . _tests [ test_id ] [ file_name ] = ( <EOL> encodeutils . safe_decode ( <EOL> self . _tests [ test_id ] [ file_name ] ) ) <EOL> self . _is_parsed = True <EOL> @ property <EOL> def tests ( self ) : <EOL> if not self . _is_parsed : <EOL> self . _post_parse ( ) <EOL> return self . _tests <EOL> @ property <EOL> def total ( self ) : <EOL> total_time = <NUM_LIT:0> <EOL> if self . _first_timestamp : <EOL> total_time = total_seconds ( <EOL> self . _last_timestamp - self . _first_timestamp ) <EOL> return { "<STR_LIT>" : len ( self . tests ) , <EOL> "<STR_LIT:time>" : total_time , <EOL> "<STR_LIT>" : len ( self . filter_tests ( "<STR_LIT>" ) ) , <EOL> "<STR_LIT>" : len ( self . filter_tests ( "<STR_LIT>" ) ) , <EOL> "<STR_LIT:success>" : len ( self . filter_tests ( "<STR_LIT:success>" ) ) , <EOL> "<STR_LIT>" : len ( self . filter_tests ( "<STR_LIT>" ) ) , <EOL> "<STR_LIT>" : len ( self . filter_tests ( "<STR_LIT>" ) ) } <EOL> @ preparse_input_args <EOL> def status ( self , test_id = None , test_status = None , tags = None , <EOL> file_name = None , file_bytes = None , mime_type = None , <EOL> timestamp = None , charset = None ) : <EOL> if test_status == "<STR_LIT>" : <EOL> self . _tests [ test_id ] = { "<STR_LIT:status>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : ( test_id . split ( "<STR_LIT:[>" ) [ <NUM_LIT:0> ] <EOL> if test_id . find ( "<STR_LIT:[>" ) > - <NUM_LIT:1> <EOL> else test_id ) , <EOL> "<STR_LIT:time>" : "<STR_LIT>" % <NUM_LIT:0> } <EOL> if tags : <EOL> self . _tests [ test_id ] [ "<STR_LIT>" ] = tags <EOL> elif test_id in self . _tests : <EOL> if test_status == "<STR_LIT>" : <EOL> self . _timestamps [ test_id ] = timestamp <EOL> elif test_status : <EOL> self . _tests [ test_id ] [ "<STR_LIT:time>" ] = total_seconds ( <EOL> timestamp - self . _timestamps [ test_id ] ) <EOL> self . _tests [ test_id ] [ "<STR_LIT:status>" ] = test_status <EOL> else : <EOL> if file_name in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> if file_name not in self . _tests [ test_id ] : <EOL> self . _tests [ test_id ] [ file_name ] = file_bytes <EOL> else : <EOL> self . _tests [ test_id ] [ file_name ] += file_bytes <EOL> else : <EOL> self . _unknown_entities . setdefault ( test_id , { "<STR_LIT:name>" : test_id } ) <EOL> self . _unknown_entities [ test_id ] [ "<STR_LIT:status>" ] = test_status <EOL> if file_name in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> if file_name not in self . _unknown_entities [ test_id ] : <EOL> self . _unknown_entities [ test_id ] [ file_name ] = file_bytes <EOL> else : <EOL> self . _unknown_entities [ test_id ] [ file_name ] += file_bytes <EOL> if timestamp : <EOL> if not self . _first_timestamp : <EOL> self . _first_timestamp = timestamp <EOL> self . _last_timestamp = timestamp <EOL> def filter_tests ( self , status ) : <EOL> """<STR_LIT>""" <EOL> filtered_tests = { } <EOL> for test in self . tests : <EOL> if self . tests [ test ] [ "<STR_LIT:status>" ] == status : <EOL> filtered_tests [ test ] = self . tests [ test ] <EOL> return filtered_tests <EOL> def parse_results_file ( filename , expected_failures = None ) : <EOL> with open ( filename , "<STR_LIT:rb>" ) as source : <EOL> results = SubunitV2StreamResult ( expected_failures ) <EOL> v2 . ByteStreamToStreamResult ( <EOL> source = source , non_subunit_name = "<STR_LIT>" ) . run ( results ) <EOL> return results </s>
