File: parse_stdout.py

package info (click to toggle)
dask.distributed 2022.12.1%2Bds.1-3
  • links: PTS, VCS
  • area: main
  • in suites: bookworm
  • size: 10,164 kB
  • sloc: python: 81,938; javascript: 1,549; makefile: 228; sh: 100
file content (114 lines) | stat: -rw-r--r-- 4,210 bytes parent folder | download
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
"""On Windows, pytest-timeout kills off the whole test suite, leaving no junit report
behind. Parse the stdout of pytest to generate one.
"""
from __future__ import annotations

import html
import re
import sys
from collections import Counter, defaultdict
from collections.abc import Iterable
from datetime import datetime

OUTCOMES = {
    "PASSED",
    "FAILED",
    # Test timeout. Marked as a variant of FAILED in the junit report
    None,
    # Setup failed or teardown failed.
    # In the latter case, if the test also failed, show both a FAILED and an ERROR line.
    "ERROR",
    # @pytest.mark.skip, @pytest.mark.skipif, or raise pytest.skip()
    "SKIPPED",
    # Reported as a variant of SKIPPED in the junit report
    "XFAIL",
    # These appear respectively before and after another status. Ignore.
    "RERUN",
    "LEAKED",
}


def parse_rows(rows: Iterable[str]) -> list[tuple[str, str, set[str | None]]]:
    match = re.compile(
        r"(distributed/.*test.*)::([^ ]*)"
        r"( (.*)(PASSED|FAILED|ERROR|SKIPPED|XFAIL|RERUN|LEAKED).*| )$"
    )

    out: defaultdict[tuple[str, str], set[str | None]] = defaultdict(set)

    for row in rows:
        m = match.match(row)
        if not m:
            continue

        fname = m.group(1)
        clsname = fname.replace("/", ".").replace(".py", "").replace("::", ".")

        tname = m.group(2).strip()
        if m.group(4) and "]" in m.group(4):
            tname += " " + m.group(4).split("]")[0] + "]"

        outcome = m.group(5)
        assert outcome in OUTCOMES
        if outcome not in {"RERUN", "LEAKED"}:
            out[clsname, tname].add(outcome)

    return [(clsname, tname, outcomes) for (clsname, tname), outcomes in out.items()]


def build_xml(rows: list[tuple[str, str, set[str | None]]]) -> None:
    cnt = Counter(outcome for _, _, outcomes in rows for outcome in outcomes)
    timestamp = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")

    # We could have used ElementTree but it feels like overkill here
    print('<?xml version="1.0" encoding="utf-8"?>')
    print("<testsuites>")
    print(
        '<testsuite name="distributed" '
        f'errors="{cnt["ERROR"]}" failures="{cnt["FAILED"] + cnt[None]}" '
        f'skipped="{cnt["SKIPPED"] + cnt["XFAIL"]}" tests="{sum(cnt.values())}" '
        f'time="0.0" timestamp="{timestamp}" hostname="">'
    )

    for clsname, tname, outcomes in rows:
        clsname = html.escape(clsname)
        tname = html.escape(tname)
        print(f'<testcase classname="{clsname}" name="{tname}" time="0.0"', end="")
        if outcomes == {"PASSED"}:
            print(" />")
        elif outcomes == {"FAILED"}:
            print('><failure message=""></failure></testcase>')
        elif outcomes == {None}:
            print('><failure message="pytest-timeout exceeded"></failure></testcase>')
        elif outcomes == {"ERROR"}:
            print('><error message="failed on setup"></error></testcase>')
        elif outcomes == {"PASSED", "ERROR"}:
            print('><error message="failed on teardown"></error></testcase>')
        elif outcomes == {"FAILED", "ERROR"}:
            print(
                '><failure message=""></failure></testcase>\n'
                f'<testcase classname="{clsname}" name="{tname}" time="0.0">'
                '<error message="failed on teardown"></error></testcase>'
            )
        elif outcomes == {"SKIPPED"}:
            print('><skipped type="pytest.skip" message="skip"></skipped></testcase>')
        elif outcomes == {"XFAIL"}:
            print('><skipped type="pytest.xfail" message="xfail"></skipped></testcase>')
        else:  # pragma: nocover
            # This should be unreachable. We would normally raise ValueError, except
            # that a crash in this script would be pretty much invisible.
            print(
                f' />\n<testcase classname="parse_stdout" name="build_xml" time="0.0">'
                f'><failure message="Unexpected {outcomes=}"></failure></testcase>'
            )

    print("</testsuite>")
    print("</testsuites>")


def main() -> None:  # pragma: nocover
    build_xml(parse_rows(sys.stdin))


if __name__ == "__main__":
    main()  # pragma: nocover