cachepc-linux

Fork of AMDESE/linux with modifications for CachePC side-channel attack
git clone https://git.sinitax.com/sinitax/cachepc-linux
Log | Files | Refs | README | LICENSE | sfeed.txt

run_checks.py (2348B)

      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': ['/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:])