|
| 1 | +from __future__ import unicode_literals, print_function |
| 2 | +from contextlib import contextmanager |
| 3 | +from xml.etree import ElementTree as etree |
| 4 | +from xml.sax.saxutils import quoteattr |
| 5 | + |
| 6 | +import six |
| 7 | + |
| 8 | + |
| 9 | +CNAME_TAGS = ('system-out', 'skipped', 'error', 'failure') |
| 10 | +CNAME_PATTERN = '<![CDATA[{}]]>' |
| 11 | +TAG_PATTERN = '<{tag}{attrs}>{text}</{tag}>' |
| 12 | + |
| 13 | + |
| 14 | +@contextmanager |
| 15 | +def patch_etree_cname(etree): |
| 16 | + """ |
| 17 | + Patch ElementTree's _serialize_xml function so that it will |
| 18 | + write text as CDATA tag for tags tags defined in CNAME_TAGS. |
| 19 | +
|
| 20 | + >>> import re |
| 21 | + >>> from xml.etree import ElementTree |
| 22 | + >>> xml_string = ''' |
| 23 | + ... <testsuite name="nosetests" tests="1" errors="0" failures="0" skip="0"> |
| 24 | + ... <testcase classname="some.class.Foo" name="test_system_out" time="0.001"> |
| 25 | + ... <system-out>Some output here</system-out> |
| 26 | + ... </testcase> |
| 27 | + ... <testcase classname="some.class.Foo" name="test_skipped" time="0.001"> |
| 28 | + ... <skipped type="unittest.case.SkipTest" message="Skipped">Skipped</skipped> |
| 29 | + ... </testcase> |
| 30 | + ... <testcase classname="some.class.Foo" name="test_error" time="0.001"> |
| 31 | + ... <error type="KeyError" message="Error here">Error here</error> |
| 32 | + ... </testcase> |
| 33 | + ... <testcase classname="some.class.Foo" name="test_failure" time="0.001"> |
| 34 | + ... <failure type="AssertionError" message="Failure here">Failure here</failure> |
| 35 | + ... </testcase> |
| 36 | + ... </testsuite> |
| 37 | + ... ''' |
| 38 | + >>> tree = ElementTree.fromstring(xml_string) |
| 39 | + >>> with patch_etree_cname(ElementTree): |
| 40 | + ... saved = str(ElementTree.tostring(tree)) |
| 41 | + >>> systemout = re.findall(r'(<system-out>.*?</system-out>)', saved)[0] |
| 42 | + >>> print(systemout) |
| 43 | + <system-out><![CDATA[Some output here]]></system-out> |
| 44 | + >>> skipped = re.findall(r'(<skipped.*?</skipped>)', saved)[0] |
| 45 | + >>> print(skipped) |
| 46 | + <skipped message="Skipped" type="unittest.case.SkipTest"><![CDATA[Skipped]]></skipped> |
| 47 | + >>> error = re.findall(r'(<error.*?</error>)', saved)[0] |
| 48 | + >>> print(error) |
| 49 | + <error message="Error here" type="KeyError"><![CDATA[Error here]]></error> |
| 50 | + >>> failure = re.findall(r'(<failure.*?</failure>)', saved)[0] |
| 51 | + >>> print(failure) |
| 52 | + <failure message="Failure here" type="AssertionError"><![CDATA[Failure here]]></failure> |
| 53 | + """ |
| 54 | + original_serialize = etree._serialize_xml |
| 55 | + |
| 56 | + def _serialize_xml(write, elem, *args, **kwargs): |
| 57 | + if elem.tag in CNAME_TAGS: |
| 58 | + attrs = ' '.join( |
| 59 | + ['{}={}'.format(k, quoteattr(v)) |
| 60 | + for k, v in sorted(elem.attrib.items())] |
| 61 | + ) |
| 62 | + attrs = ' ' + attrs if attrs else '' |
| 63 | + text = CNAME_PATTERN.format(elem.text) |
| 64 | + write(TAG_PATTERN.format( |
| 65 | + tag=elem.tag, |
| 66 | + attrs=attrs, |
| 67 | + text=text |
| 68 | + )) |
| 69 | + else: |
| 70 | + original_serialize(write, elem, *args, **kwargs) |
| 71 | + |
| 72 | + etree._serialize_xml = etree._serialize['xml'] = _serialize_xml |
| 73 | + |
| 74 | + yield |
| 75 | + |
| 76 | + etree._serialize_xml = etree._serialize['xml'] = original_serialize |
| 77 | + |
| 78 | + |
| 79 | +def merge_trees(*trees): |
| 80 | + """ |
| 81 | + Merge all given XUnit ElementTrees into a single ElementTree. |
| 82 | + This combines all of the children test-cases and also merges |
| 83 | + all of the metadata of how many tests were executed, etc. |
| 84 | + """ |
| 85 | + first_tree = trees[0] |
| 86 | + first_root = first_tree.getroot() |
| 87 | + |
| 88 | + if len(trees) == 0: |
| 89 | + return first_tree |
| 90 | + |
| 91 | + for tree in trees[1:]: |
| 92 | + root = tree.getroot() |
| 93 | + |
| 94 | + # append children elements (testcases) |
| 95 | + first_root.extend(root.getchildren()) |
| 96 | + |
| 97 | + # combine root attributes which stores the number |
| 98 | + # of executed tests, skipped tests, etc |
| 99 | + for key, value in first_root.attrib.items(): |
| 100 | + if not value.isdigit(): |
| 101 | + continue |
| 102 | + combined = six.text_type(int(value) + int(root.attrib.get(key, '0'))) |
| 103 | + first_root.set(key, combined) |
| 104 | + |
| 105 | + return first_tree |
| 106 | + |
| 107 | + |
| 108 | +def merge_xunit(files, output, callback=None): |
| 109 | + """ |
| 110 | + Merge the given xunit xml files into a single output xml file. |
| 111 | +
|
| 112 | + If callback is not None, it will be called with the merged ElementTree |
| 113 | + before the output file is written (useful for applying other fixes to |
| 114 | + the merged file). This can either modify the element tree in place (and |
| 115 | + return None) or return a completely new ElementTree to be written. |
| 116 | + """ |
| 117 | + trees = [] |
| 118 | + |
| 119 | + for f in files: |
| 120 | + trees.append(etree.parse(f)) |
| 121 | + |
| 122 | + merged = merge_trees(*trees) |
| 123 | + |
| 124 | + if callback is not None: |
| 125 | + result = callback(merged) |
| 126 | + if result is not None: |
| 127 | + merged = result |
| 128 | + |
| 129 | + with patch_etree_cname(etree): |
| 130 | + merged.write(output, encoding='utf-8', xml_declaration=True) |
0 commit comments