1#!/usr/bin/python 2# 3# tap2subunit: convert a tap stream to a subunit stream. 4# Extract from the subunit source: 5# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net> 6# 7# Licensed under either the Apache License, Version 2.0 or the BSD 3-clause 8# license at the users choice. A copy of both licenses are available in the 9# project source as Apache-2.0 and BSD. You may not use this file except in 10# compliance with one of these two licences. 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT 14# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 15# license you chose for the specific language governing permissions and 16# limitations under that license. 17# 18 19 20import re 21import sys 22 23def TAP2SubUnit(tap, subunit): 24 """Filter a TAP pipe into a subunit pipe. 25 26 :param tap: A tap pipe/stream/file object. 27 :param subunit: A pipe/stream/file object to write subunit results to. 28 :return: The exit code to exit with. 29 """ 30 BEFORE_PLAN = 0 31 AFTER_PLAN = 1 32 SKIP_STREAM = 2 33 state = BEFORE_PLAN 34 plan_start = 1 35 plan_stop = 0 36 def _skipped_test(subunit, plan_start): 37 # Some tests were skipped. 38 subunit.write('test: test %d\n' % plan_start) 39 subunit.write('error: test %d [\n' % plan_start) 40 subunit.write('test missing from TAP output\n') 41 subunit.write(']\n') 42 return plan_start + 1 43 # Test data for the next test to emit 44 test_name = None 45 log = [] 46 result = None 47 def _emit_test(): 48 "write out a test" 49 if test_name is None: 50 return 51 subunit.write("test: %s\n" % test_name) 52 if not log: 53 subunit.write("%s: %s\n" % (result, test_name)) 54 else: 55 subunit.write("%s: %s [\n" % (result, test_name)) 56 if log: 57 for line in log: 58 subunit.write("%s\n" % line) 59 subunit.write("]\n") 60 del log[:] 61 for line in tap: 62 if state == BEFORE_PLAN: 63 match = re.match("(\d+)\.\.(\d+)\s*(?:\#\s+(.*))?\n", line) 64 if match: 65 state = AFTER_PLAN 66 _, plan_stop, comment = match.groups() 67 plan_stop = int(plan_stop) 68 if plan_start > plan_stop and plan_stop == 0: 69 # skipped file 70 state = SKIP_STREAM 71 subunit.write("test: file skip\n") 72 subunit.write("skip: file skip [\n") 73 subunit.write("%s\n" % comment) 74 subunit.write("]\n") 75 continue 76 # not a plan line, or have seen one before 77 match = re.match("(ok|not ok)(?:\s+(\d+)?)?(?:\s+([^#]*[^#\s]+)\s*)?(?:\s+#\s+(TODO|SKIP|skip|todo)(?:\s+(.*))?)?\n", line) 78 if match: 79 # new test, emit current one. 80 _emit_test() 81 status, number, description, directive, directive_comment = match.groups() 82 if status == 'ok': 83 result = 'success' 84 else: 85 result = "failure" 86 if description is None: 87 description = '' 88 else: 89 description = ' ' + description 90 if directive is not None: 91 if directive.upper() == 'TODO': 92 result = 'xfail' 93 elif directive.upper() == 'SKIP': 94 result = 'skip' 95 if directive_comment is not None: 96 log.append(directive_comment) 97 if number is not None: 98 number = int(number) 99 while plan_start < number: 100 plan_start = _skipped_test(subunit, plan_start) 101 test_name = "test %d%s" % (plan_start, description) 102 plan_start += 1 103 continue 104 match = re.match("Bail out\!(?:\s*(.*))?\n", line) 105 if match: 106 reason, = match.groups() 107 if reason is None: 108 extra = '' 109 else: 110 extra = ' %s' % reason 111 _emit_test() 112 test_name = "Bail out!%s" % extra 113 result = "error" 114 state = SKIP_STREAM 115 continue 116 match = re.match("\#.*\n", line) 117 if match: 118 log.append(line[:-1]) 119 continue 120 subunit.write(line) 121 _emit_test() 122 while plan_start <= plan_stop: 123 # record missed tests 124 plan_start = _skipped_test(subunit, plan_start) 125 return 0 126 127 128sys.exit(TAP2SubUnit(sys.stdin, sys.stdout)) 129