2010-03-30 00:25:25 +04:00
# Python module for parsing and generating the Subunit protocol
# (Samba-specific)
# Copyright (C) 2008-2009 Jelmer Vernooij <jelmer@samba.org>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
__all__ = [ ' parse_results ' ]
import re
2010-03-30 14:46:26 +04:00
import sys
2010-03-30 16:36:25 +04:00
import subunit
2010-09-13 22:53:54 +04:00
import testtools
2010-03-30 00:25:25 +04:00
import time
VALID_RESULTS = [ ' success ' , ' successful ' , ' failure ' , ' fail ' , ' skip ' , ' knownfail ' , ' error ' , ' xfail ' , ' skip-testsuite ' , ' testsuite-failure ' , ' testsuite-xfail ' , ' testsuite-success ' , ' testsuite-error ' ]
def parse_results ( msg_ops , statistics , fh ) :
expected_fail = 0
open_tests = [ ]
while fh :
l = fh . readline ( )
2010-03-30 02:30:52 +04:00
if l == " " :
break
2010-04-11 03:39:06 +04:00
parts = l . split ( None , 1 )
if not len ( parts ) == 2 or not l . startswith ( parts [ 0 ] ) :
2010-08-26 04:25:44 +04:00
msg_ops . output_msg ( l )
2010-04-11 03:39:06 +04:00
continue
command = parts [ 0 ] . rstrip ( " : " )
arg = parts [ 1 ]
if command in ( " test " , " testing " ) :
msg_ops . control_msg ( l )
msg_ops . start_test ( arg . rstrip ( ) )
open_tests . append ( arg . rstrip ( ) )
elif command == " time " :
2010-03-30 00:25:25 +04:00
msg_ops . control_msg ( l )
2010-03-30 02:30:52 +04:00
grp = re . match (
2010-09-05 05:14:44 +04:00
' ( \ d+)-( \ d+)-( \ d+) ( \ d+):( \ d+):([.0-9]+) \n ' , arg )
if grp is None :
grp = re . match (
' ( \ d+)-( \ d+)-( \ d+) ( \ d+):( \ d+):([.0-9]+)Z \n ' , arg )
if grp is None :
print " Unable to parse time line: %s " % arg
if grp is not None :
msg_ops . report_time ( time . mktime ( ( int ( grp . group ( 1 ) ) , int ( grp . group ( 2 ) ) , int ( grp . group ( 3 ) ) , int ( grp . group ( 4 ) ) , int ( grp . group ( 5 ) ) , int ( float ( grp . group ( 6 ) ) ) , 0 , 0 , 0 ) ) )
2010-04-11 03:39:06 +04:00
elif command in VALID_RESULTS :
2010-03-30 00:25:25 +04:00
msg_ops . control_msg ( l )
2010-04-11 03:39:06 +04:00
result = command
grp = re . match ( " (.*?)( \ [)?([ \t ]*)( multipart)? \n " , arg )
( testname , hasreason ) = ( grp . group ( 1 ) , grp . group ( 2 ) )
2010-03-30 00:25:25 +04:00
if hasreason :
reason = " "
# reason may be specified in next lines
terminated = False
while fh :
l = fh . readline ( )
2010-03-30 02:30:52 +04:00
if l == " " :
break
2010-03-30 00:25:25 +04:00
msg_ops . control_msg ( l )
if l == " ] \n " :
terminated = True
break
else :
reason + = l
2010-08-26 05:50:08 +04:00
2010-03-30 00:25:25 +04:00
if not terminated :
statistics [ ' TESTS_ERROR ' ] + = 1
2010-03-30 02:59:04 +04:00
msg_ops . end_test ( testname , " error " , True ,
2010-03-30 00:25:25 +04:00
" reason ( %s ) interrupted " % result )
return 1
2010-03-30 02:30:52 +04:00
else :
reason = None
2010-03-30 00:25:25 +04:00
if result in ( " success " , " successful " ) :
2010-04-11 00:35:57 +04:00
try :
open_tests . remove ( testname )
2010-05-30 19:05:07 +04:00
except ValueError :
2010-04-11 00:35:57 +04:00
statistics [ ' TESTS_ERROR ' ] + = 1
msg_ops . end_test ( testname , " error " , True ,
" Test was never started " )
else :
statistics [ ' TESTS_EXPECTED_OK ' ] + = 1
msg_ops . end_test ( testname , " success " , False , reason )
2010-03-30 00:25:25 +04:00
elif result in ( " xfail " , " knownfail " ) :
2010-04-11 00:35:57 +04:00
try :
open_tests . remove ( testname )
2010-05-30 19:05:07 +04:00
except ValueError :
2010-04-11 00:35:57 +04:00
statistics [ ' TESTS_ERROR ' ] + = 1
msg_ops . end_test ( testname , " error " , True ,
" Test was never started " )
else :
statistics [ ' TESTS_EXPECTED_FAIL ' ] + = 1
msg_ops . end_test ( testname , " xfail " , False , reason )
expected_fail + = 1
2010-03-30 00:25:25 +04:00
elif result in ( " failure " , " fail " ) :
2010-04-11 00:35:57 +04:00
try :
open_tests . remove ( testname )
2010-05-30 19:05:07 +04:00
except ValueError :
2010-04-11 00:35:57 +04:00
statistics [ ' TESTS_ERROR ' ] + = 1
msg_ops . end_test ( testname , " error " , True ,
" Test was never started " )
else :
statistics [ ' TESTS_UNEXPECTED_FAIL ' ] + = 1
msg_ops . end_test ( testname , " failure " , True , reason )
2010-03-30 00:25:25 +04:00
elif result == " skip " :
statistics [ ' TESTS_SKIP ' ] + = 1
# Allow tests to be skipped without prior announcement of test
last = open_tests . pop ( )
if last is not None and last != testname :
open_tests . append ( testname )
2010-03-30 02:59:04 +04:00
msg_ops . end_test ( testname , " skip " , False , reason )
2010-03-30 00:25:25 +04:00
elif result == " error " :
statistics [ ' TESTS_ERROR ' ] + = 1
2010-04-11 00:35:57 +04:00
try :
open_tests . remove ( testname )
2010-05-30 19:05:07 +04:00
except ValueError :
2010-04-11 00:35:57 +04:00
pass
2010-03-30 02:59:04 +04:00
msg_ops . end_test ( testname , " error " , True , reason )
2010-03-30 00:25:25 +04:00
elif result == " skip-testsuite " :
msg_ops . skip_testsuite ( testname )
elif result == " testsuite-success " :
msg_ops . end_testsuite ( testname , " success " , reason )
elif result == " testsuite-failure " :
msg_ops . end_testsuite ( testname , " failure " , reason )
elif result == " testsuite-xfail " :
msg_ops . end_testsuite ( testname , " xfail " , reason )
elif result == " testsuite-error " :
msg_ops . end_testsuite ( testname , " error " , reason )
2010-08-26 04:25:44 +04:00
else :
raise AssertionError ( " Recognized but unhandled result %r " %
result )
2010-04-11 03:39:06 +04:00
elif command == " testsuite " :
msg_ops . start_testsuite ( arg . strip ( ) )
elif command == " progress " :
arg = arg . strip ( )
2010-03-30 16:36:25 +04:00
if arg == " pop " :
msg_ops . progress ( None , subunit . PROGRESS_POP )
elif arg == " push " :
msg_ops . progress ( None , subunit . PROGRESS_PUSH )
elif arg [ 0 ] in ' +- ' :
msg_ops . progress ( int ( arg ) , subunit . PROGRESS_CUR )
else :
msg_ops . progress ( int ( arg ) , subunit . PROGRESS_SET )
2010-03-30 00:25:25 +04:00
else :
msg_ops . output_msg ( l )
while open_tests :
2010-03-30 02:59:04 +04:00
msg_ops . end_test ( open_tests . pop ( ) , " error " , True ,
2010-03-30 00:25:25 +04:00
" was started but never finished! " )
statistics [ ' TESTS_ERROR ' ] + = 1
if statistics [ ' TESTS_ERROR ' ] > 0 :
return 1
if statistics [ ' TESTS_UNEXPECTED_FAIL ' ] > 0 :
return 1
return 0
2010-03-30 02:59:04 +04:00
class SubunitOps ( object ) :
2010-03-30 00:25:25 +04:00
2010-03-30 02:59:04 +04:00
def start_test ( self , testname ) :
print " test: %s " % testname
2010-03-30 00:25:25 +04:00
2010-03-30 02:59:04 +04:00
def end_test ( self , name , result , reason = None ) :
if reason :
print " %s : %s [ " % ( result , name )
2010-08-26 05:50:08 +04:00
print reason
2010-03-30 02:59:04 +04:00
print " ] "
else :
print " %s : %s " % ( result , name )
2010-03-30 00:25:25 +04:00
2010-03-30 02:59:04 +04:00
def skip_test ( self , name , reason = None ) :
self . end_test ( name , " skip " , reason )
2010-03-30 00:25:25 +04:00
2010-03-30 02:59:04 +04:00
def fail_test ( self , name , reason = None ) :
self . end_test ( name , " fail " , reason )
2010-03-30 00:25:25 +04:00
2010-03-30 02:59:04 +04:00
def success_test ( self , name , reason = None ) :
self . end_test ( name , " success " , reason )
2010-03-30 00:25:25 +04:00
2010-03-30 02:59:04 +04:00
def xfail_test ( self , name , reason = None ) :
self . end_test ( name , " xfail " , reason )
2010-03-30 00:25:25 +04:00
2010-03-30 02:59:04 +04:00
def report_time ( self , t ) :
2010-03-30 14:46:26 +04:00
( year , mon , mday , hour , min , sec , wday , yday , isdst ) = time . localtime ( t )
print " time: %04d - %02d - %02d %02d : %02d : %02d " % ( year , mon , mday , hour , min , sec )
2010-03-30 00:25:25 +04:00
2010-03-30 16:36:25 +04:00
def progress ( self , offset , whence ) :
if whence == subunit . PROGRESS_CUR and offset > - 1 :
prefix = " + "
elif whence == subunit . PROGRESS_PUSH :
prefix = " "
offset = " push "
elif whence == subunit . PROGRESS_POP :
prefix = " "
offset = " pop "
else :
prefix = " "
print " progress: %s %s " % ( prefix , offset )
2010-03-30 02:59:04 +04:00
# The following are Samba extensions:
def start_testsuite ( self , name ) :
print " testsuite: %s " % name
2010-03-30 00:25:25 +04:00
2010-03-30 02:59:04 +04:00
def skip_testsuite ( self , name , reason = None ) :
if reason :
print " skip-testsuite: %s [ \n %s \n ] " % ( name , reason )
else :
print " skip-testsuite: %s " % name
2010-03-30 00:25:25 +04:00
2010-03-30 02:59:04 +04:00
def end_testsuite ( self , name , result , reason = None ) :
if reason :
2010-03-30 14:46:26 +04:00
print " testsuite- %s : %s [ " % ( result , name )
2010-03-30 02:59:04 +04:00
print " %s " % reason
print " ] "
else :
2010-03-30 14:46:26 +04:00
print " testsuite- %s : %s " % ( result , name )
2010-03-30 00:25:25 +04:00
2010-03-30 14:46:26 +04:00
def read_test_regexes ( name ) :
2010-04-06 05:55:10 +04:00
ret = { }
2010-03-30 14:46:26 +04:00
f = open ( name , ' r ' )
try :
for l in f :
l = l . strip ( )
if l == " " or l [ 0 ] == " # " :
continue
if " # " in l :
( regex , reason ) = l . split ( " # " , 1 )
2010-04-06 05:55:10 +04:00
ret [ regex . strip ( ) ] = reason . strip ( )
2010-03-30 14:46:26 +04:00
else :
2010-04-06 05:55:10 +04:00
ret [ l ] = None
2010-03-30 14:46:26 +04:00
finally :
f . close ( )
2010-04-06 05:55:10 +04:00
return ret
2010-03-30 14:46:26 +04:00
def find_in_list ( regexes , fullname ) :
2010-04-06 05:55:10 +04:00
for regex , reason in regexes . iteritems ( ) :
2010-03-30 14:46:26 +04:00
if re . match ( regex , fullname ) :
if reason is None :
return " "
return reason
return None
2010-09-13 22:53:54 +04:00
class FilterOps ( testtools . testresult . TestResult ) :
2010-03-30 14:46:26 +04:00
def control_msg ( self , msg ) :
pass # We regenerate control messages, so ignore this
def report_time ( self , time ) :
self . _ops . report_time ( time )
2010-03-30 16:36:25 +04:00
def progress ( self , delta , whence ) :
self . _ops . progress ( delta , whence )
2010-03-30 14:46:26 +04:00
def output_msg ( self , msg ) :
if self . output is None :
sys . stdout . write ( msg )
else :
self . output + = msg
def start_test ( self , testname ) :
if self . prefix is not None :
testname = self . prefix + testname
if self . strip_ok_output :
self . output = " "
self . _ops . start_test ( testname )
def end_test ( self , testname , result , unexpected , reason ) :
if self . prefix is not None :
testname = self . prefix + testname
if result in ( " fail " , " failure " ) and not unexpected :
result = " xfail "
self . xfail_added + = 1
self . total_xfail + = 1
xfail_reason = find_in_list ( self . expected_failures , testname )
if xfail_reason is not None and result in ( " fail " , " failure " ) :
result = " xfail "
self . xfail_added + = 1
self . total_xfail + = 1
reason + = xfail_reason
if result in ( " fail " , " failure " ) :
self . fail_added + = 1
self . total_fail + = 1
if result == " error " :
self . error_added + = 1
self . total_error + = 1
if self . strip_ok_output :
if result not in ( " success " , " xfail " , " skip " ) :
print self . output
self . output = None
self . _ops . end_test ( testname , result , reason )
def skip_testsuite ( self , name , reason = None ) :
self . _ops . skip_testsuite ( name , reason )
def start_testsuite ( self , name ) :
self . _ops . start_testsuite ( name )
self . error_added = 0
self . fail_added = 0
self . xfail_added = 0
def end_testsuite ( self , name , result , reason = None ) :
xfail = False
if self . xfail_added > 0 :
xfail = True
if self . fail_added > 0 or self . error_added > 0 :
xfail = False
if xfail and result in ( " fail " , " failure " ) :
result = " xfail "
if self . fail_added > 0 and result != " failure " :
result = " failure "
if reason is None :
reason = " Subunit/Filter Reason "
reason + = " \n failures[ %d ] " % self . fail_added
if self . error_added > 0 and result != " error " :
result = " error "
if reason is None :
reason = " Subunit/Filter Reason "
reason + = " \n errors[ %d ] " % self . error_added
self . _ops . end_testsuite ( name , result , reason )
def __init__ ( self , prefix , expected_failures , strip_ok_output ) :
self . _ops = SubunitOps ( )
self . output = None
self . prefix = prefix
self . expected_failures = expected_failures
self . strip_ok_output = strip_ok_output
self . xfail_added = 0
self . total_xfail = 0
self . total_error = 0
self . total_fail = 0