| # SPDX-License-Identifier: GPL-2.0+ |
| # |
| # Copyright (c) 2016 Google, Inc |
| # |
| |
| from contextlib import contextmanager |
| import glob |
| import multiprocessing |
| import os |
| import sys |
| import unittest |
| |
| from patman import command |
| |
| from io import StringIO |
| |
| use_concurrent = True |
| try: |
| from concurrencytest.concurrencytest import ConcurrentTestSuite |
| from concurrencytest.concurrencytest import fork_for_tests |
| except: |
| use_concurrent = False |
| |
| |
| def RunTestCoverage(prog, filter_fname, exclude_list, build_dir, required=None, |
| extra_args=None): |
| """Run tests and check that we get 100% coverage |
| |
| Args: |
| prog: Program to run (with be passed a '-t' argument to run tests |
| filter_fname: Normally all *.py files in the program's directory will |
| be included. If this is not None, then it is used to filter the |
| list so that only filenames that don't contain filter_fname are |
| included. |
| exclude_list: List of file patterns to exclude from the coverage |
| calculation |
| build_dir: Build directory, used to locate libfdt.py |
| required: List of modules which must be in the coverage report |
| extra_args (str): Extra arguments to pass to the tool before the -t/test |
| arg |
| |
| Raises: |
| ValueError if the code coverage is not 100% |
| """ |
| # This uses the build output from sandbox_spl to get _libfdt.so |
| path = os.path.dirname(prog) |
| if filter_fname: |
| glob_list = glob.glob(os.path.join(path, '*.py')) |
| glob_list = [fname for fname in glob_list if filter_fname in fname] |
| else: |
| glob_list = [] |
| glob_list += exclude_list |
| glob_list += ['*libfdt.py', '*site-packages*', '*dist-packages*'] |
| glob_list += ['*concurrencytest*'] |
| test_cmd = 'test' if 'binman' in prog or 'patman' in prog else '-t' |
| prefix = '' |
| if build_dir: |
| prefix = 'PYTHONPATH=$PYTHONPATH:%s/sandbox_spl/tools ' % build_dir |
| cmd = ('%spython3-coverage run ' |
| '--omit "%s" %s %s %s -P1' % (prefix, ','.join(glob_list), |
| prog, extra_args or '', test_cmd)) |
| os.system(cmd) |
| stdout = command.Output('python3-coverage', 'report') |
| lines = stdout.splitlines() |
| if required: |
| # Convert '/path/to/name.py' just the module name 'name' |
| test_set = set([os.path.splitext(os.path.basename(line.split()[0]))[0] |
| for line in lines if '/etype/' in line]) |
| missing_list = required |
| missing_list.discard('__init__') |
| missing_list.difference_update(test_set) |
| if missing_list: |
| print('Missing tests for %s' % (', '.join(missing_list))) |
| print(stdout) |
| ok = False |
| |
| coverage = lines[-1].split(' ')[-1] |
| ok = True |
| print(coverage) |
| if coverage != '100%': |
| print(stdout) |
| print("Type 'python3-coverage html' to get a report in " |
| 'htmlcov/index.html') |
| print('Coverage error: %s, but should be 100%%' % coverage) |
| ok = False |
| if not ok: |
| raise ValueError('Test coverage failure') |
| |
| |
| # Use this to suppress stdout/stderr output: |
| # with capture_sys_output() as (stdout, stderr) |
| # ...do something... |
| @contextmanager |
| def capture_sys_output(): |
| capture_out, capture_err = StringIO(), StringIO() |
| old_out, old_err = sys.stdout, sys.stderr |
| try: |
| sys.stdout, sys.stderr = capture_out, capture_err |
| yield capture_out, capture_err |
| finally: |
| sys.stdout, sys.stderr = old_out, old_err |
| |
| |
| def ReportResult(toolname:str, test_name: str, result: unittest.TestResult): |
| """Report the results from a suite of tests |
| |
| Args: |
| toolname: Name of the tool that ran the tests |
| test_name: Name of test that was run, or None for all |
| result: A unittest.TestResult object containing the results |
| """ |
| # Remove errors which just indicate a missing test. Since Python v3.5 If an |
| # ImportError or AttributeError occurs while traversing name then a |
| # synthetic test that raises that error when run will be returned. These |
| # errors are included in the errors accumulated by result.errors. |
| if test_name: |
| errors = [] |
| |
| for test, err in result.errors: |
| if ("has no attribute '%s'" % test_name) not in err: |
| errors.append((test, err)) |
| result.testsRun -= 1 |
| result.errors = errors |
| |
| print(result) |
| for test, err in result.errors: |
| print(test.id(), err) |
| for test, err in result.failures: |
| print(err, result.failures) |
| if result.skipped: |
| print('%d %s test%s SKIPPED:' % (len(result.skipped), toolname, |
| 's' if len(result.skipped) > 1 else '')) |
| for skip_info in result.skipped: |
| print('%s: %s' % (skip_info[0], skip_info[1])) |
| if result.errors or result.failures: |
| print('%s tests FAILED' % toolname) |
| return 1 |
| return 0 |
| |
| |
| def RunTestSuites(result, debug, verbosity, test_preserve_dirs, processes, |
| test_name, toolpath, test_class_list): |
| """Run a series of test suites and collect the results |
| |
| Args: |
| result: A unittest.TestResult object to add the results to |
| debug: True to enable debugging, which shows a full stack trace on error |
| verbosity: Verbosity level to use (0-4) |
| test_preserve_dirs: True to preserve the input directory used by tests |
| so that it can be examined afterwards (only useful for debugging |
| tests). If a single test is selected (in args[0]) it also preserves |
| the output directory for this test. Both directories are displayed |
| on the command line. |
| processes: Number of processes to use to run tests (None=same as #CPUs) |
| test_name: Name of test to run, or None for all |
| toolpath: List of paths to use for tools |
| test_class_list: List of test classes to run |
| """ |
| for module in []: |
| suite = doctest.DocTestSuite(module) |
| suite.run(result) |
| |
| sys.argv = [sys.argv[0]] |
| if debug: |
| sys.argv.append('-D') |
| if verbosity: |
| sys.argv.append('-v%d' % verbosity) |
| if toolpath: |
| for path in toolpath: |
| sys.argv += ['--toolpath', path] |
| |
| suite = unittest.TestSuite() |
| loader = unittest.TestLoader() |
| for module in test_class_list: |
| # Test the test module about our arguments, if it is interested |
| if hasattr(module, 'setup_test_args'): |
| setup_test_args = getattr(module, 'setup_test_args') |
| setup_test_args(preserve_indir=test_preserve_dirs, |
| preserve_outdirs=test_preserve_dirs and test_name is not None, |
| toolpath=toolpath, verbosity=verbosity) |
| if test_name: |
| try: |
| suite.addTests(loader.loadTestsFromName(test_name, module)) |
| except AttributeError: |
| continue |
| else: |
| suite.addTests(loader.loadTestsFromTestCase(module)) |
| if use_concurrent and processes != 1: |
| concurrent_suite = ConcurrentTestSuite(suite, |
| fork_for_tests(processes or multiprocessing.cpu_count())) |
| concurrent_suite.run(result) |
| else: |
| suite.run(result) |