1#!/usr/bin/env python3
2# SPDX-License-Identifier: GPL-2.0
3#
4# This file runs some basic checks to verify kunit works.
5# It is only of interest if you're making changes to KUnit itself.
6#
7# Copyright (C) 2021, Google LLC.
8# Author: Daniel Latypov <dlatypov@google.com.com>
9
10from concurrent import futures
11import datetime
12import os
13import shutil
14import subprocess
15import sys
16import textwrap
17from typing import Dict, List, Sequence, Tuple
18
19ABS_TOOL_PATH = os.path.abspath(os.path.dirname(__file__))
20TIMEOUT = datetime.timedelta(minutes=5).total_seconds()
21
22commands: Dict[str, Sequence[str]] = {
23	'kunit_tool_test.py': ['./kunit_tool_test.py'],
24	'kunit smoke test': ['./kunit.py', 'run', '--kunitconfig=lib/kunit', '--build_dir=kunit_run_checks'],
25	'pytype': ['/bin/sh', '-c', 'pytype *.py'],
26	'mypy': ['/bin/sh', '-c', 'mypy *.py'],
27}
28
29# The user might not have mypy or pytype installed, skip them if so.
30# Note: you can install both via `$ pip install mypy pytype`
31necessary_deps : Dict[str, str] = {
32	'pytype': 'pytype',
33	'mypy': 'mypy',
34}
35
36def main(argv: Sequence[str]) -> None:
37	if argv:
38		raise RuntimeError('This script takes no arguments')
39
40	future_to_name: Dict[futures.Future, str] = {}
41	executor = futures.ThreadPoolExecutor(max_workers=len(commands))
42	for name, argv in commands.items():
43		if name in necessary_deps and shutil.which(necessary_deps[name]) is None:
44			print(f'{name}: SKIPPED, {necessary_deps[name]} not in $PATH')
45			continue
46		f = executor.submit(run_cmd, argv)
47		future_to_name[f] = name
48
49	has_failures = False
50	print(f'Waiting on {len(future_to_name)} checks ({", ".join(future_to_name.values())})...')
51	for f in  futures.as_completed(future_to_name.keys()):
52		name = future_to_name[f]
53		ex = f.exception()
54		if not ex:
55			print(f'{name}: PASSED')
56			continue
57
58		has_failures = True
59		if isinstance(ex, subprocess.TimeoutExpired):
60			print(f'{name}: TIMED OUT')
61		elif isinstance(ex, subprocess.CalledProcessError):
62			print(f'{name}: FAILED')
63		else:
64			print(f'{name}: unexpected exception: {ex}')
65			continue
66
67		output = ex.output
68		if output:
69			print(textwrap.indent(output.decode(), '> '))
70	executor.shutdown()
71
72	if has_failures:
73		sys.exit(1)
74
75
76def run_cmd(argv: Sequence[str]):
77	subprocess.check_output(argv, stderr=subprocess.STDOUT, cwd=ABS_TOOL_PATH, timeout=TIMEOUT)
78
79
80if __name__ == '__main__':
81	main(sys.argv[1:])
82