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