selftest/tap2subunit

   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
  20 import re
  21 import sys
  22
  23 def 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(r"(\d+)\.\.(\d+)\s*(?:\#\s+(.*))?"
  64                              "\n", line)
  65             if match:
  66                 state = AFTER_PLAN
  67                 _, plan_stop, comment = match.groups()
  68                 plan_stop = int(plan_stop)
  69                 if plan_start > plan_stop and plan_stop == 0:
  70                     # skipped file
  71                     state = SKIP_STREAM
  72                     subunit.write("test: file skip\n")
  73                     subunit.write("skip: file skip [\n")
  74                     subunit.write("%s\n" % comment)
  75                     subunit.write("]\n")
  76                 continue
  77         # not a plan line, or have seen one before
  78         match = re.match(r"(ok|not ok)(?:\s+(\d+)?)?(?:\s+([^#]*[^#\s]+)\s*)?(?:\s+#\s+(TODO|SKIP|skip|todo)(?:\s+(.*))?)?"
  79                          "\n", line)
  80         if match:
  81             # new test, emit current one.
  82             _emit_test()
  83             status, number, description, directive, directive_comment = match.groups()
  84             if status == 'ok':
  85                 result = 'success'
  86             else:
  87                 result = "failure"
  88             if description is None:
  89                 description = ''
  90             else:
  91                 description = ' ' + description
  92             if directive is not None:
  93                 if directive.upper() == 'TODO':
  94                     result = 'xfail'
  95                 elif directive.upper() == 'SKIP':
  96                     result = 'skip'
  97                 if directive_comment is not None:
  98                     log.append(directive_comment)
  99             if number is not None:
 100                 number = int(number)
 101                 while plan_start < number:
 102                     plan_start = _skipped_test(subunit, plan_start)
 103             test_name = "test %d%s" % (plan_start, description)
 104             plan_start += 1
 105             continue
 106         match = re.match(r"Bail out\!(?:\s*(.*))?"
 107                          "\n", line)
 108         if match:
 109             reason, = match.groups()
 110             if reason is None:
 111                 extra = ''
 112             else:
 113                 extra = ' %s' % reason
 114             _emit_test()
 115             test_name = "Bail out!%s" % extra
 116             result = "error"
 117             state = SKIP_STREAM
 118             continue
 119         match = re.match(r"\#.*"
 120                          "\n", line)
 121         if match:
 122             log.append(line[:-1])
 123             continue
 124         subunit.write(line)
 125     _emit_test()
 126     while plan_start <= plan_stop:
 127         # record missed tests
 128         plan_start = _skipped_test(subunit, plan_start)
 129     return 0
 130
 131
 132 sys.exit(TAP2SubUnit(sys.stdin, sys.stdout))