1
#!/usr/bin/env python3
2
import argparse
3
import json
4
import re
5
import subprocess
6
import sys
7
import time
8
from dataclasses import dataclass
9
from pathlib import Path
10
from typing import Any, Optional
11
from xml.etree import ElementTree
12
 
13
EXCLUDED_TESTS = {
14
    # The WPT compliant implementation is in urlpattern-standard@
15
    "urlpattern@"
16
}
17
 
18
 
19
@dataclass
20
class Options:
21
    config: bool
22
    report: bool
23
    stats: bool
24
 
25
 
26
def main() -> None:
27
    cmd = argparse.ArgumentParser()
28
    cmd.add_argument("--update-config", action="store_true")
29
    cmd.add_argument("--write-report", type=Path)
30
    cmd.add_argument("--print-stats", action="store_true")
31
    cmd.add_argument("logs_dir", type=Path)
32
 
33
    args = cmd.parse_args()
34
    options = Options(args.update_config, bool(args.write_report), args.print_stats)
35
    logs = parse_logs(args.logs_dir, options)
36
 
37
    if args.print_stats:
38
        print(stats_table(logs))
39
 
40
    if args.write_report:
41
        # TODO(soon): Elapsed time will not be accurate. Figure out if it matters to us, and how to fix.
42
        now = int(time.time())
43
        time_interval = TimeInterval(now, now)
44
 
45
        report = wpt_report(logs, time_interval)
46
        if report:
47
            with args.write_report.open("w") as fp:
48
                json.dump(report, fp, indent=2)
49
 
50
    # TODO(soon): Implement the config option to update test configs
51
 
52
 
53
@dataclass
54
class TimeInterval:
55
    start: int = 0
56
    end: int = 0
57
 
58
 
59
@dataclass
60
class Log:
61
    # TypeScript test config used to configure WPT tests
62
    config: Optional[str] = None
63
 
64
    # JSON report in WPT's format
65
    report: Optional[dict[str, Any]] = None
66
 
67
    # Summary stats in JSON
68
    stats: Optional[list[Any]] = None
69
 
70
 
71
def parse_logs(logs_dir: Path, options: Options) -> list[Log]:
72
    return [
73
        parse_log(log_file, options)
74
        for log_file in sorted(logs_dir.glob("**/src/wpt/**/*@/test.xml"))
75
    ]
76
 
77
 
78
def parse_log(log_file: Path, options: Options) -> Log:
79
    log = Log()
80
 
81
    if log_file.parent.name in EXCLUDED_TESTS:
82
        return log
83
 
84
    parser = ElementTree.XMLParser(encoding="utf-8")
85
    root = ElementTree.parse(log_file, parser=parser)
86
    text = root.find("testsuite").find("system-out").text
87
 
88
    start_log = re.search(r"\[ TEST \] .*:zzz_results\n", text)
89
    if not start_log:
90
        return log
91
 
92
    end_log = re.search(r"^.*\[ (PASS|FAIL) \].*:zzz_results", text, re.MULTILINE)
93
 
94
    if not end_log:
95
        return log
96
 
97
    results_log = text[start_log.end(0) : end_log.start(0)]
98
    items = results_log.split("***")
99
 
100
    if options.config:
101
        # Removes a message for the user that we don't need
102
        items.pop(0)
103
        items.pop(0)
104
        log.config = items.pop(0).strip()
105
 
106
    if options.report:
107
        log.report = json.loads(items.pop(0))
108
 
109
    if options.stats:
110
        log.stats = json.loads(items.pop(0))
111
 
112
    return log
113
 
114
 
115
def stats_table(logs: list[Log]) -> str:
116
    cells = []
117
 
118
    for log in logs:
119
        if not log.stats:
120
            continue
121
 
122
        cells.append(
123
            [f"<td>{log.stats[0]}</td>"]
124
            + [f"<td align='right'>{value}</td>" for value in log.stats[1:]]
125
        )
126
 
127
    if not cells:
128
        return ""
129
 
130
    cells_html = "\n".join(f"<tr>{' '.join(row)}</tr>" for row in cells)
131
 
132
    return f"""## WPT statistics
133
134
<table>
135
    <tr>
136
        <th>Module</th>
137
        <th colspan="4">Coverage</th>
138
        <th colspan="5">Pass</th>
139
    </tr>
140
    <tr>
141
        <th></th>
142
        <th>OK</th>
143
        <th>Disabled</th>
144
        <th>Total</th>
145
        <th>% OK</th>
146
        <th>Pass</th>
147
        <th>Fail</th>
148
        <th>Disabled</th>
149
        <th>Total</th>
150
        <th>% Pass</th>
151
    </tr>
152
    {cells_html}
153
</table>
154
155
This table shows how workerd performs for each listed [Web Platform Tests](https://github.com/web-platform-tests/wpt) module."""
156
 
157
 
158
def cmd_output(cmd: list[str]) -> str:
159
    return subprocess.run(cmd, capture_output=True, check=True).stdout.decode().strip()
160
 
161
 
162
def get_os() -> str:
163
    """
164
    Return one of three expected values
165
    <https://github.com/web-platform-tests/wpt/blob/1c6ff12/tools/wptrunner/wptrunner/tests/test_update.py#L953-L958>
166
    """
167
 
168
    if sys.platform == "darwin":
169
        return "mac"
170
    elif sys.platform == "linux":
171
        return "linux"
172
    elif sys.platform == "win32":
173
        return "win"
174
    else:
175
        raise ValueError("Unsupported os type")
176
 
177
 
178
def wpt_report(logs: list[Log], time_interval: TimeInterval) -> dict[str, Any]:
179
    all_results = [
180
        result for log in logs if log.report for result in log.report["results"]
181
    ]
182
 
183
    if not all_results:
184
        return {}
185
 
186
    return {
187
        "time_start": time_interval.start,
188
        "time_end": time_interval.end,
189
        "run_info": {
190
            "product": "workerd",
191
            "browser_channel": "experimental",
192
            "browser_version": cmd_output(["git", "describe", "--tags", "--always"]),
193
            "revision": cmd_output(["git", "rev-parse", "HEAD"]),
194
            "os": get_os(),
195
        },
196
        "results": all_results,
197
    }
198
 
199
 
200
if __name__ == "__main__":
201
    main()