scripts

 1#!/usr/bin/env python3
 2
 3import gzip
 4import sys
 5import json
 6
 7
 8def main(argv: list[str]):
 9    original_times_by_test = {}
10    final_times_by_test = {}
11    id_by_test = {}
12    with gzip.open(argv[1]) as gzip_f:
13        first_timestamp = -1.0
14        for line in gzip_f:
15            entry = json.loads(line)
16            if first_timestamp == -1:
17                first_timestamp = entry["timestamp"]
18            if "payload" not in entry:
19                pass
20            elif "program_output" not in entry["payload"]:
21                pass
22            else:
23                duration = entry["timestamp"] - first_timestamp
24                payload = entry["payload"]["program_output"]["data"].strip()
25                prints = [
26                    "[RUNNING]",
27                    "[SKIPPED]",
28                    "[PASSED]",
29                    "[FAILED]",
30                    "[TIMED_OUT]",
31                ]
32                for i, p in enumerate(prints):
33                    if p not in payload:
34                        continue
35
36                    test = payload.split()[1]
37                    if test in id_by_test and id_by_test[test] != entry["id"]:
38                        raise RuntimeError(
39                            "test already run by a different test: {}".format(
40                                id_by_test[test]
41                            )
42                        )
43
44                    id_by_test[test] = entry["id"]
45                    if i == 0:
46                        original_times_by_test[test] = duration
47                    else:
48                        final_times_by_test[test] = duration
49
50                    break
51
52    for test, orig in original_times_by_test.items():
53        final = final_times_by_test[test]
54        id = id_by_test[test]
55        print("{} {} {}".format(test, id, final - orig))
56
57
58if __name__ == "__main__":
59    main(sys.argv)