tools/testing/kunit/run_checks.py

   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
  10 from concurrent import futures
  11 import datetime
  12 import os
  13 import shutil
  14 import subprocess
  15 import sys
  16 import textwrap
  17 from typing import Dict, List, Sequence, Tuple
  18
  19 ABS_TOOL_PATH = os.path.abspath(os.path.dirname(__file__))
  20 TIMEOUT = datetime.timedelta(minutes=5).total_seconds()
  21
  22 commands: 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`
  31 necessary_deps : Dict[str, str] = {
  32         'pytype': 'pytype',
  33         'mypy': 'mypy',
  34 }
  35
  36 def 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('{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
  76 def run_cmd(argv: Sequence[str]):
  77         subprocess.check_output(argv, stderr=subprocess.STDOUT, cwd=ABS_TOOL_PATH, timeout=TIMEOUT)
  78
  79
  80 if __name__ == '__main__':
  81         main(sys.argv[1:])