xunit2subunit filter to convert xunitxml to subunit

Makefile.am

@@ -63,7 +63,8 @@ dist_bin_SCRIPTS = \
 	filters/subunit2gtk \
 	filters/subunit2junitxml \
 	filters/subunit2pyunit \
-	filters/tap2subunit
+	filters/tap2subunit \
+        filters/xunit2subunit
 
 TESTS = $(check_PROGRAMS)
 

filters/xunit2subunit

@@ -0,0 +1,87 @@
+#!/usr/bin/env python
+#  subunit: extensions to python unittest to get test results from subprocesses.
+#  Hewlett Packard Enterprise (c) 2017
+#
+#  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.
+#
+
+"""A filter that reads a xunitXML stream and outputs a subunit stream.
+"""
+
+import datetime
+import sys
+import math
+
+from subunit.iso8601 import UTC
+from subunit.v2 import StreamResultToBytes
+from datetime import timedelta
+from xml.etree import ElementTree as ET
+
+STATUS_CODES = frozenset([
+    'exists',
+    'fail',
+    'skip',
+    'success',
+    'uxsuccess',
+    'xfail',
+])
+
+
+def xunit2subunit(xunitxml_input=sys.stdin, output=sys.stdout):
+    output = StreamResultToBytes(output)
+    tree = ET.parse(xunitxml_input)
+    for testcase in tree.findall('.//testcase'):
+        test_id = testcase.attrib.get('name')
+        test_metadata = None
+        test_status = None
+        skipped = None
+        failure = None
+        test_time = to_timedelta(testcase.attrib.get('time'))
+        skipped = testcase.find('.//skipped')
+        failure = testcase.find('.//failure')
+        if skipped is not None : test_status = "skip"
+        elif failure is not None :test_status = "fail"
+        else : test_status = "success"
+        write_test(output, test_id, test_status, test_metadata, test_time)
+
+def write_test(output, test_id, test_status, metadatas, test_time):
+    write_status = output.status
+    kwargs = {}
+    if metadatas:
+        if 'tags' in metadatas:
+            tags = metadatas['tags']
+            kwargs['test_tags'] = tags.split(',')
+        if 'attrs' in metadatas:
+            test_id = test_id + '[' + metadatas['attrs'] + ']'
+    kwargs['test_id'] = test_id
+    if test_status in STATUS_CODES:
+        kwargs['test_status'] = test_status
+    kwargs['timestamp'] = create_timestamp()
+    kwargs['test_status'] = "inprogress"
+    write_status(**kwargs)
+    kwargs['timestamp'] = create_timestamp() + test_time
+    kwargs['test_status'] = test_status
+    write_status(**kwargs)
+
+def to_timedelta(value):
+    if value is None:
+        return None
+    sec = float(value)
+    if math.isnan(sec):
+        return None
+    return timedelta(seconds=sec)
+
+def create_timestamp():
+    return datetime.datetime.now(UTC)
+
+if __name__ == '__main__':
+    sys.exit(xunit2subunit(sys.stdin, sys.stdout))

setup.py

@@ -78,6 +78,7 @@ def _get_version_from_file(filename, start_of_line, split_marker):
         'filters/subunit2junitxml',
         'filters/subunit2pyunit',
         'filters/tap2subunit',
+        'filters/xunit2subunit',
     ],
     **extra
 )