testing/kunit/run_checks.py

0001 #!/usr/bin/env python3
0002 # SPDX-License-Identifier: GPL-2.0
0003 #
0004 # This file runs some basic checks to verify kunit works.
0005 # It is only of interest if you're making changes to KUnit itself.
0006 #
0007 # Copyright (C) 2021, Google LLC.
0008 # Author: Daniel Latypov <dlatypov@google.com.com>
0009
0010 from concurrent import futures
0011 import datetime
0012 import os
0013 import shutil
0014 import subprocess
0015 import sys
0016 import textwrap
0017 from typing import Dict, List, Sequence
0018
0019 ABS_TOOL_PATH = os.path.abspath(os.path.dirname(__file__))
0020 TIMEOUT = datetime.timedelta(minutes=5).total_seconds()
0021
0022 commands: Dict[str, Sequence[str]] = {
0023     'kunit_tool_test.py': ['./kunit_tool_test.py'],
0024     'kunit smoke test': ['./kunit.py', 'run', '--kunitconfig=lib/kunit', '--build_dir=kunit_run_checks'],
0025     'pytype': ['/bin/sh', '-c', 'pytype *.py'],
0026     'mypy': ['/bin/sh', '-c', 'mypy *.py'],
0027 }
0028
0029 # The user might not have mypy or pytype installed, skip them if so.
0030 # Note: you can install both via `$ pip install mypy pytype`
0031 necessary_deps : Dict[str, str] = {
0032     'pytype': 'pytype',
0033     'mypy': 'mypy',
0034 }
0035
0036 def main(argv: Sequence[str]) -> None:
0037     if argv:
0038         raise RuntimeError('This script takes no arguments')
0039
0040     future_to_name: Dict[futures.Future, str] = {}
0041     executor = futures.ThreadPoolExecutor(max_workers=len(commands))
0042     for name, argv in commands.items():
0043         if name in necessary_deps and shutil.which(necessary_deps[name]) is None:
0044             print(f'{name}: SKIPPED, {necessary_deps[name]} not in $PATH')
0045             continue
0046         f = executor.submit(run_cmd, argv)
0047         future_to_name[f] = name
0048
0049     has_failures = False
0050     print(f'Waiting on {len(future_to_name)} checks ({", ".join(future_to_name.values())})...')
0051     for f in  futures.as_completed(future_to_name.keys()):
0052         name = future_to_name[f]
0053         ex = f.exception()
0054         if not ex:
0055             print(f'{name}: PASSED')
0056             continue
0057
0058         has_failures = True
0059         if isinstance(ex, subprocess.TimeoutExpired):
0060             print(f'{name}: TIMED OUT')
0061         elif isinstance(ex, subprocess.CalledProcessError):
0062             print(f'{name}: FAILED')
0063         else:
0064             print(f'{name}: unexpected exception: {ex}')
0065             continue
0066
0067         output = ex.output
0068         if output:
0069             print(textwrap.indent(output.decode(), '> '))
0070     executor.shutdown()
0071
0072     if has_failures:
0073         sys.exit(1)
0074
0075
0076 def run_cmd(argv: Sequence[str]):
0077     subprocess.check_output(argv, stderr=subprocess.STDOUT, cwd=ABS_TOOL_PATH, timeout=TIMEOUT)
0078
0079
0080 if __name__ == '__main__':
0081     main(sys.argv[1:])