xref: /honee/tests/junit.py (revision 3a98a9868128cddbae29bd879e7c98a077a4f2e8)

#!/usr/bin/env python3
from junit_common import *


def create_argparser() -> argparse.ArgumentParser:
    """Creates argument parser to read command line arguments

    Returns:
        argparse.ArgumentParser: Created `ArgumentParser`
    """
    parser = argparse.ArgumentParser('Test runner with JUnit and TAP output')
    parser.add_argument(
        '-c',
        '--ceed-backends',
        type=str,
        nargs='*',
        default=['/cpu/self'],
        help='libCEED backend to use with convergence tests')
    parser.add_argument(
        '-m',
        '--mode',
        type=RunMode,
        action=CaseInsensitiveEnumAction,
        help='Output mode, junit or tap',
        default=RunMode.JUNIT)
    parser.add_argument('-n', '--nproc', type=int, default=1, help='number of MPI processes')
    parser.add_argument('-b', '--junit-batch', type=str, default='', help='Name of JUnit batch for output file')
    parser.add_argument('-np', '--pool-size', type=int, default=1, help='Number of test cases to run in parallel')
    parser.add_argument('--smartredis-dir', type=str, default='', help='path to SmartSim library, if present')
    parser.add_argument('--has-torch', type=bool, default=False, help='Whether to build with torch')
    parser.add_argument('-s', '--search', type=str, default='.*',
                        help='Search string to filter tests, using `re` package format')
    parser.add_argument('-v', '--verbose', action='store_true', default=False,
                        help='print details for all runs, not just failures')
    parser.add_argument('test', help='Test executable', nargs='?')

    return parser


def diff_csv_comment_function(test_line: str, true_line: str) -> Optional[str]:
    test_line_split = test_line.split(':')[0]
    true_line_split = true_line.split(':')[0]
    diff_output = ''.join(difflib.unified_diff([test_line_split + '\n'],
                                               [true_line_split + '\n'],
                                               tofile='test created file',
                                               fromfile='expected output'))
    return diff_output if diff_output else None


class HoneeSuiteSpec(SuiteSpec):
    def __init__(self, has_torch: bool):
        self.has_torch: bool = has_torch
        self.csv_rtol = 1e-9
        self.csv_comment_diff_fn = diff_csv_comment_function
        self.csv_comment_str = '#'

    def get_source_path(self, test: str) -> Path:
        """Compute path to test source file

        Args:
            test (str): Name of test

        Returns:
            Path: Path to source file
        """
        if test.startswith('navierstokes'):
            return (Path('examples') / 'navierstokes').with_suffix('.c')
        else:
            return (Path('tests') / test).with_suffix('.c')

    # get path to executable
    def get_run_path(self, test: str) -> Path:
        """Compute path to built test executable file

        Args:
            test (str): Name of test

        Returns:
            Path: Path to test executable
        """
        return Path('build') / test

    def get_output_path(self, test: str, output_file: str) -> Path:
        """Compute path to expected output file

        Args:
            test (str): Name of test
            output_file (str): File name of output file

        Returns:
            Path: Path to expected output file
        """
        return Path('tests') / 'output' / output_file

    def check_pre_skip(self, test: str, spec: TestSpec, resource: str, nproc: int) -> Optional[str]:
        """Check if a test case should be skipped prior to running, returning the reason for skipping

        Args:
            test (str): Name of test
            spec (TestSpec): Test case specification
            resource (str): libCEED backend
            nproc (int): Number of MPI processes to use when running test case

        Returns:
            Optional[str]: Skip reason, or `None` if test case should not be skipped
        """
        for condition in spec.only:
            if (condition == 'cpu') and ('gpu' in resource):
                return 'CPU only test with GPU backend'
            if condition == 'torch' and not self.has_torch:
                return 'PyTorch only test without USE_TORCH=1'

    def check_post_skip(self, test: str, spec: TestSpec, resource: str, stderr: str) -> Optional[str]:
        """Check if a test case should be allowed to fail, based on its stderr output

        Args:
            test (str): Name of test
            spec (TestSpec): Test case specification
            resource (str): libCEED backend
            stderr (str): Standard error output from test case execution

        Returns:
            Optional[str]: Skip reason, or `None` if unexpeced error
        """
        if 'No SYCL devices of the requested type are available' in stderr:
            return f'SYCL device type not available'
        elif 'Loading meshes requires CGNS support. Reconfigure using --with-cgns-dir' in stderr:
            return f'CGNS not installed in PETSc for {test}, {spec.name}'
        elif 'You may need to add --download-ctetgen or --download-tetgen' in stderr:
            return f'Tet mesh generator not installed for {test}, {spec.name}'
        return None

    def check_required_failure(self, test: str, spec: TestSpec, resource: str, stderr: str) -> Tuple[str, bool]:
        """Check whether a test case is expected to fail and if it failed expectedly

        Args:
            test (str): Name of test
            spec (TestSpec): Test case specification
            resource (str): libCEED backend
            stderr (str): Standard error output from test case execution

        Returns:
            tuple[str, bool]: Tuple of the expected failure string and whether it was present in `stderr`
        """
        return '', True

    def check_allowed_stdout(self, test: str) -> bool:
        """Check whether a test is allowed to print console output

        Args:
            test (str): Name of test

        Returns:
            bool: True if the test is allowed to print console output
        """
        return False


if __name__ == '__main__':
    args = create_argparser().parse_args()

    # run tests
    if 'smartsim' in args.test:
        has_smartsim: bool = args.smartredis_dir and Path(args.smartredis_dir).is_dir()
        test_cases = []
        test_directory = Path(__file__).parent / 'smartsim_test_dir'

        if args.mode is RunMode.TAP:
            print(f'1..1')
        if has_smartsim:
            from smartsim_regression_framework import SmartSimTest

            test_framework = SmartSimTest(test_directory)
            test_framework.setup()

            is_new_subtest = True
            subtest_ok = True
            for i, backend in enumerate(args.ceed_backends):
                test_cases.append(test_framework.test_junit(backend))
                if is_new_subtest and args.mode == RunMode.TAP:
                    is_new_subtest = False
                    print(f'# Subtest: {test_cases[0].category}')
                    print(f'    1..{len(args.ceed_backends)}')
                print(
                    test_case_output_string(
                        test_cases[i],
                        TestSpec("SmartSim Tests"),
                        args.mode,
                        backend,
                        '',
                        i,
                        verbose=args.verbose))
            if args.mode == RunMode.TAP:
                print(f'{"" if subtest_ok else "not "}ok 1 - {test_cases[0].category}')
            test_framework.teardown()
        elif args.mode is RunMode.TAP:
            print(f'ok 1 - # SKIP SmartSim not installed')
        result: TestSuite = TestSuite('SmartSim Tests', test_cases)
        if has_failures(result):
            shutil.copytree(
                test_directory,
                HoneeSuiteSpec(
                    args.has_torch).test_failure_artifacts_path,
                dirs_exist_ok=True)
    else:
        result: TestSuite = run_tests(
            args.test,
            args.ceed_backends,
            args.mode,
            args.nproc,
            HoneeSuiteSpec(args.has_torch),
            args.pool_size,
            search=args.search,
            verbose=args.verbose)

    # write output and check for failures
    if args.mode is RunMode.JUNIT:
        write_junit_xml(result, args.junit_batch)
        if has_failures(result):
            sys.exit(1)