• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# SPDX-License-Identifier: GPL-2.0-or-later
3# Copyright (c) 2025 Cyril Hrubis <chrubis@suse.cz>
4# Copyright (c) 2025 Andrea Cervesato <andrea.cervesato@suse.com>
5"""
6This script parses JSON results from kirk and LTP metadata in order to
7calculate timeouts for tests based on the results file.
8It can also patch tests automatically and replace the calculated timeout.
9"""
10
11import re
12import os
13import json
14import argparse
15
16# The test runtime is multiplied by this to get a timeout
17TIMEOUT_MUL = 1.2
18
19
20def _sed(fname, expr, replace):
21    """
22    Pythonic version of sed command.
23    """
24    content = []
25    matcher = re.compile(expr)
26
27    with open(fname, 'r', encoding="utf-8") as data:
28        for line in data:
29            match = matcher.search(line)
30            if not match:
31                content.append(line)
32            else:
33                content.append(replace)
34
35    with open(fname, 'w', encoding="utf-8") as data:
36        data.writelines(content)
37
38
39def _patch(ltp_dir, fname, new_timeout, override):
40    """
41    If `override` is True, it patches a test file, searching for timeout and
42    replacing it with `new_timeout`.
43    """
44    orig_timeout = None
45    file_path = os.path.join(ltp_dir, fname)
46
47    with open(file_path, 'r', encoding="utf-8") as c_source:
48        matcher = re.compile(r'\s*.timeout\s*=\s*(\d+).')
49        for line in c_source:
50            match = matcher.search(line)
51            if not match:
52                continue
53
54            timeout = match.group(1)
55            orig_timeout = int(timeout)
56
57    if orig_timeout:
58        if orig_timeout < new_timeout or override:
59            print(f"CHANGE {fname} timeout {orig_timeout} -> {new_timeout}")
60            _sed(file_path, r".timeout = [0-9]*,\n",
61                 f"\t.timeout = {new_timeout},\n")
62        else:
63            print(f"KEEP   {fname} timeout {orig_timeout} (new {new_timeout})")
64    else:
65        print(f"ADD    {fname} timeout {new_timeout}")
66        _sed(file_path,
67             "static struct tst_test test = {",
68             "static struct tst_test test = {\n"
69             f"\t.timeout = {new_timeout},\n")
70
71
72def _patch_all(ltp_dir, timeouts, override):
73    """
74    Patch all tests.
75    """
76    for timeout in timeouts:
77        if timeout['path']:
78            _patch(ltp_dir, timeout['path'], timeout['timeout'], override)
79
80
81def _print_table(timeouts):
82    """
83    Print the timeouts table.
84    """
85    timeouts.sort(key=lambda x: x['timeout'], reverse=True)
86
87    total = 0
88
89    print("Old library tests\n-----------------\n")
90    for timeout in timeouts:
91        if not timeout['newlib']:
92            print(f"{timeout['name']:30s} {timeout['timeout']}")
93            total += 1
94
95    print(f"\n\t{total} tests in total")
96
97    total = 0
98
99    print("\nNew library tests\n-----------------\n")
100    for timeout in timeouts:
101        if timeout['newlib']:
102            print(f"{timeout['name']:30s} {timeout['timeout']}")
103            total += 1
104
105    print(f"\n\t{total} tests in total")
106
107
108def _parse_data(ltp_dir, results_path):
109    """
110    Parse results data and metadata, then it generates timeouts data.
111    """
112    timeouts = []
113    results = None
114    metadata = None
115
116    with open(results_path, 'r', encoding="utf-8") as file:
117        results = json.load(file)
118
119    metadata_path = os.path.join(ltp_dir, 'metadata', 'ltp.json')
120    with open(metadata_path, 'r', encoding="utf-8") as file:
121        metadata = json.load(file)
122
123    for test in results['results']:
124        name = test['test_fqn']
125        duration = test['test']['duration']
126
127        # if test runs for all_filesystems, normalize runtime to one filesystem
128        filesystems = max(1, test['test']['log'].count('TINFO: Formatting /'))
129
130        # check if test is new library test
131        test_is_newlib = name in metadata['tests']
132
133        # store test file path
134        path = None
135        if test_is_newlib:
136            path = metadata['tests'][name]['fname']
137
138        test_has_runtime = False
139        if test_is_newlib:
140            # filter out tests with runtime
141            test_has_runtime = 'runtime' in metadata['tests'][name]
142
143            # timer tests define runtime dynamically in timer library
144            test_has_runtime = 'sample' in metadata['tests'][name]
145
146        # select tests that does not have runtime and which are executed
147        # for a long time
148        if not test_has_runtime and duration >= 0.5:
149            data = {}
150            data["name"] = name
151            data["timeout"] = int(TIMEOUT_MUL * duration/filesystems + 0.5)
152            data["newlib"] = test_is_newlib
153            data["path"] = path
154
155            timeouts.append(data)
156
157    return timeouts
158
159
160def _file_exists(filepath):
161    """
162    Check if the given file path exists.
163    """
164    if not os.path.isfile(filepath):
165        raise argparse.ArgumentTypeError(
166            f"The file '{filepath}' does not exist.")
167    return filepath
168
169
170def _dir_exists(dirpath):
171    """
172    Check if the given directory path exists.
173    """
174    if not os.path.isdir(dirpath):
175        raise argparse.ArgumentTypeError(
176            f"The directory '{dirpath}' does not exist.")
177    return dirpath
178
179
180def run():
181    """
182    Entry point of the script.
183    """
184    parser = argparse.ArgumentParser(
185        description="Script to calculate LTP tests timeouts")
186
187    parser.add_argument(
188        '-l',
189        '--ltp-dir',
190        type=_dir_exists,
191        help='LTP source code directory',
192        default='..')
193
194    parser.add_argument(
195        '-r',
196        '--results',
197        type=_file_exists,
198        required=True,
199        help='kirk results.json file location')
200
201    parser.add_argument(
202        '-o',
203        '--override',
204        default=False,
205        action='store_true',
206        help='Always override test timeouts')
207
208    parser.add_argument(
209        '-p',
210        '--patch',
211        default=False,
212        action='store_true',
213        help='Patch tests with updated timeout')
214
215    parser.add_argument(
216        '-t',
217        '--print-table',
218        default=True,
219        action='store_true',
220        help='Print table with suggested timeouts')
221
222    args = parser.parse_args()
223
224    timeouts = _parse_data(args.ltp_dir, args.results)
225
226    if args.print_table:
227        _print_table(timeouts)
228
229    if args.patch:
230        _patch_all(args.ltp_dir, timeouts, args.override)
231
232
233if __name__ == "__main__":
234    run()
235