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