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 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': ['mypy', '--config-file', 'mypy.ini', '--exclude', '_test.py$', '--exclude', 'qemu_configs/', '.'], 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[None], 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]) -> None: 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