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

from abc import ABC, abstractmethod
from collections.abc import Iterable
import argparse
import csv
from dataclasses import dataclass, field, fields
import difflib
from enum import Enum
from math import isclose
import os
from pathlib import Path
import re
import subprocess
import multiprocessing as mp
import sys
import time
from typing import Optional, Tuple, List, Dict, Callable, Iterable, get_origin
import shutil

sys.path.insert(0, str(Path(__file__).parent / "junit-xml"))
from junit_xml import TestCase, TestSuite, to_xml_report_string  # nopep8


class ParseError(RuntimeError):
    """A custom exception for failed parsing."""

    def __init__(self, message):
        super().__init__(message)


class CaseInsensitiveEnumAction(argparse.Action):
    """Action to convert input values to lower case prior to converting to an Enum type"""

    def __init__(self, option_strings, dest, type, default, **kwargs):
        if not issubclass(type, Enum):
            raise ValueError(f"{type} must be an Enum")
        # store provided enum type
        self.enum_type = type
        if isinstance(default, self.enum_type):
            pass
        elif isinstance(default, str):
            default = self.enum_type(default.lower())
        elif isinstance(default, Iterable):
            default = [self.enum_type(v.lower()) for v in default]
        else:
            raise argparse.ArgumentTypeError("Invalid value type, must be str or iterable")
        # prevent automatic type conversion
        super().__init__(option_strings, dest, default=default, **kwargs)

    def __call__(self, parser, namespace, values, option_string=None):
        if isinstance(values, self.enum_type):
            pass
        elif isinstance(values, str):
            values = self.enum_type(values.lower())
        elif isinstance(values, Iterable):
            values = [self.enum_type(v.lower()) for v in values]
        else:
            raise argparse.ArgumentTypeError("Invalid value type, must be str or iterable")
        setattr(namespace, self.dest, values)


@dataclass
class TestSpec:
    """Dataclass storing information about a single test case"""
    name: str = field(default_factory=str)
    csv_rtol: float = -1
    csv_ztol: float = -1
    cgns_tol: float = -1
    only: List = field(default_factory=list)
    args: List = field(default_factory=list)
    key_values: Dict = field(default_factory=dict)


class RunMode(Enum):
    """Enumeration of run modes, either `RunMode.TAP` or `RunMode.JUNIT`"""
    TAP = 'tap'
    JUNIT = 'junit'

    def __str__(self):
        return self.value

    def __repr__(self):
        return self.value


class SuiteSpec(ABC):
    """Abstract Base Class defining the required interface for running a test suite"""
    @abstractmethod
    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
        """
        raise NotImplementedError

    @abstractmethod
    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
        """
        raise NotImplementedError

    @abstractmethod
    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
        """
        raise NotImplementedError

    @property
    def test_failure_artifacts_path(self) -> Path:
        """Path to test failure artifacts"""
        return Path('build') / 'test_failure_artifacts'

    @property
    def cgns_tol(self):
        """Absolute tolerance for CGNS diff"""
        return getattr(self, '_cgns_tol', 1.0e-12)

    @cgns_tol.setter
    def cgns_tol(self, val):
        self._cgns_tol = val

    @property
    def csv_ztol(self):
        """Keyword arguments to be passed to diff_csv()"""
        return getattr(self, '_csv_ztol', 3e-10)

    @csv_ztol.setter
    def csv_ztol(self, val):
        self._csv_ztol = val

    @property
    def csv_rtol(self):
        """Keyword arguments to be passed to diff_csv()"""
        return getattr(self, '_csv_rtol', 1e-6)

    @csv_rtol.setter
    def csv_rtol(self, val):
        self._csv_rtol = val

    @property
    def csv_comment_diff_fn(self):  # -> Any | Callable[..., None]:
        return getattr(self, '_csv_comment_diff_fn', None)

    @csv_comment_diff_fn.setter
    def csv_comment_diff_fn(self, test_fn):
        self._csv_comment_diff_fn = test_fn

    @property
    def csv_comment_str(self):
        return getattr(self, '_csv_comment_str', '#')

    @csv_comment_str.setter
    def csv_comment_str(self, comment_str):
        self._csv_comment_str = comment_str

    def post_test_hook(self, test: str, spec: TestSpec, backend: str) -> None:
        """Function callback ran after each test case

        Args:
            test (str): Name of test
            spec (TestSpec): Test case specification
        """
        pass

    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
        """
        return None

    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 unexpected error
        """
        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


def has_cgnsdiff() -> bool:
    """Check whether `cgnsdiff` is an executable program in the current environment

    Returns:
        bool: True if `cgnsdiff` is found
    """
    my_env: dict = os.environ.copy()
    proc = subprocess.run('cgnsdiff',
                          shell=True,
                          stdout=subprocess.PIPE,
                          stderr=subprocess.PIPE,
                          env=my_env)
    return 'not found' not in proc.stderr.decode('utf-8')


def contains_any(base: str, substrings: List[str]) -> bool:
    """Helper function, checks if any of the substrings are included in the base string

    Args:
        base (str): Base string to search in
        substrings (List[str]): List of potential substrings

    Returns:
        bool: True if any substrings are included in base string
    """
    return any((sub in base for sub in substrings))


def startswith_any(base: str, prefixes: List[str]) -> bool:
    """Helper function, checks if the base string is prefixed by any of `prefixes`

    Args:
        base (str): Base string to search
        prefixes (List[str]): List of potential prefixes

    Returns:
        bool: True if base string is prefixed by any of the prefixes
    """
    return any((base.startswith(prefix) for prefix in prefixes))


def find_matching(line: str, open: str = '(', close: str = ')') -> Tuple[int, int]:
    """Find the start and end positions of the first outer paired delimeters

    Args:
        line (str): Line to search
        open (str, optional): Opening delimiter, must be different than `close`. Defaults to '('.
        close (str, optional): Closing delimeter, must be different than `open`. Defaults to ')'.

    Raises:
        RuntimeError: If open or close is not a single character
        RuntimeError: If open and close are the same characters

    Returns:
        Tuple[int]: If matching delimeters are found, return indices in `list`. Otherwise, return end < start.
    """
    if len(open) != 1 or len(close) != 1:
        raise RuntimeError("`open` and `close` must be single characters")
    if open == close:
        raise RuntimeError("`open` and `close` must be different characters")
    start: int = line.find(open)
    if start < 0:
        return -1, -1
    count: int = 1
    for i in range(start + 1, len(line)):
        if line[i] == open:
            count += 1
        if line[i] == close:
            count -= 1
            if count == 0:
                return start, i
    return start, -1


def parse_test_line(line: str, fallback_name: str = '') -> TestSpec:
    """Parse a single line of TESTARGS and CLI arguments into a `TestSpec` object

    Args:
        line (str): String containing TESTARGS specification and CLI arguments

    Returns:
        TestSpec: Parsed specification of test case
    """
    test_fields = fields(TestSpec)
    field_names = [f.name for f in test_fields]
    known: Dict = dict()
    other: Dict = dict()
    if line[0] == "(":
        # have key/value pairs to parse
        start, end = find_matching(line)
        if end < start:
            raise ParseError(f"Mismatched parentheses in TESTCASE: {line}")

        keyvalues_str = line[start:end + 1]
        keyvalues_pattern = re.compile(r'''
            (?:\(\s*|\s*,\s*)   # start with open parentheses or comma, no capture
            ([A-Za-z]+[\w\-]+)  # match key starting with alpha, containing alphanumeric, _, or -; captured as Group 1
            \s*=\s*             # key is followed by = (whitespace ignored)
            (?:                 # uncaptured group for OR
              "((?:[^"]|\\")+)" #   match quoted value (any internal " must be escaped as \"); captured as Group 2
            | ([^=]+)           #   OR match unquoted value (no equals signs allowed); captured as Group 3
            )                   # end uncaptured group for OR
            \s*(?=,|\))         # lookahead for either next comma or closing parentheses
        ''', re.VERBOSE)

        for match in re.finditer(keyvalues_pattern, keyvalues_str):
            if not match:  # empty
                continue
            key = match.group(1)
            value = match.group(2) if match.group(2) else match.group(3)
            try:
                index = field_names.index(key)
                if key == "only":  # weird bc only is a list
                    value = [constraint.strip() for constraint in value.split(',')]
                try:
                    # TODO: stop supporting python <=3.8
                    known[key] = test_fields[index].type(value)  # type: ignore
                except TypeError:
                    # TODO: this is still liable to fail for complex types
                    known[key] = get_origin(test_fields[index].type)(value)  # type: ignore
            except ValueError:
                other[key] = value

        line = line[end + 1:]

    if not 'name' in known.keys():
        known['name'] = fallback_name

    args_pattern = re.compile(r'''
        \s+(            # remove leading space
            (?:"[^"]+") # match quoted CLI option
          | (?:[\S]+)   # match anything else that is space separated
        )
    ''', re.VERBOSE)
    args: List[str] = re.findall(args_pattern, line)
    for k, v in other.items():
        print(f"warning, unknown TESTCASE option for test '{known['name']}': {k}={v}")
    return TestSpec(**known, key_values=other, args=args)


def get_test_args(source_file: Path) -> List[TestSpec]:
    """Parse all test cases from a given source file

    Args:
        source_file (Path): Path to source file

    Raises:
        RuntimeError: Errors if source file extension is unsupported

    Returns:
        List[TestSpec]: List of parsed `TestSpec` objects, or a list containing a single, default `TestSpec` if none were found
    """
    comment_str: str = ''
    if source_file.suffix in ['.c', '.cc', '.cpp']:
        comment_str = '//'
    elif source_file.suffix in ['.py']:
        comment_str = '#'
    elif source_file.suffix in ['.usr']:
        comment_str = 'C_'
    elif source_file.suffix in ['.f90']:
        comment_str = '! '
    else:
        raise RuntimeError(f'Unrecognized extension for file: {source_file}')

    return [parse_test_line(line.strip(comment_str).removeprefix("TESTARGS"), source_file.stem)
            for line in source_file.read_text().splitlines()
            if line.startswith(f'{comment_str}TESTARGS')] or [TestSpec(source_file.stem, args=['{ceed_resource}'])]


def diff_csv(test_csv: Path, true_csv: Path, zero_tol: float, rel_tol: float,
             comment_str: str = '#', comment_func: Optional[Callable[[str, str], Optional[str]]] = None) -> str:
    """Compare CSV results against an expected CSV file with tolerances

    Args:
        test_csv (Path): Path to output CSV results
        true_csv (Path): Path to expected CSV results
        zero_tol (float): Tolerance below which values are considered to be zero.
        rel_tol (float): Relative tolerance for comparing non-zero values.
        comment_str (str, optional): String to denoting commented line
        comment_func (Callable, optional): Function to determine if test and true line are different

    Returns:
        str: Diff output between result and expected CSVs
    """
    test_lines: List[str] = test_csv.read_text().splitlines()
    true_lines: List[str] = true_csv.read_text().splitlines()
    # Files should not be empty
    if len(test_lines) == 0:
        return f'No lines found in test output {test_csv}'
    if len(true_lines) == 0:
        return f'No lines found in test source {true_csv}'
    if len(test_lines) != len(true_lines):
        return f'Number of lines in {test_csv} and {true_csv} do not match'

    # Process commented lines
    uncommented_lines: List[int] = []
    for n, (test_line, true_line) in enumerate(zip(test_lines, true_lines)):
        if test_line[0] == comment_str and true_line[0] == comment_str:
            if comment_func:
                output = comment_func(test_line, true_line)
                if output:
                    return output
        elif test_line[0] == comment_str and true_line[0] != comment_str:
            return f'Commented line found in {test_csv} at line {n} but not in {true_csv}'
        elif test_line[0] != comment_str and true_line[0] == comment_str:
            return f'Commented line found in {true_csv} at line {n} but not in {test_csv}'
        else:
            uncommented_lines.append(n)

    # Remove commented lines
    test_lines = [test_lines[line] for line in uncommented_lines]
    true_lines = [true_lines[line] for line in uncommented_lines]

    test_reader: csv.DictReader = csv.DictReader(test_lines)
    true_reader: csv.DictReader = csv.DictReader(true_lines)
    if not test_reader.fieldnames:
        return f'No CSV columns found in test output {test_csv}'
    if not true_reader.fieldnames:
        return f'No CSV columns found in test source {true_csv}'
    if test_reader.fieldnames != true_reader.fieldnames:
        return ''.join(difflib.unified_diff([f'{test_lines[0]}\n'], [f'{true_lines[0]}\n'],
                       tofile='found CSV columns', fromfile='expected CSV columns'))

    diff_lines: List[str] = list()
    for test_line, true_line in zip(test_reader, true_reader):
        for key in test_reader.fieldnames:
            # Check if the value is numeric
            try:
                true_val: float = float(true_line[key])
                test_val: float = float(test_line[key])
                true_zero: bool = abs(true_val) < zero_tol
                test_zero: bool = abs(test_val) < zero_tol
                fail: bool = False
                if true_zero:
                    fail = not test_zero
                else:
                    fail = not isclose(test_val, true_val, rel_tol=rel_tol)
                if fail:
                    diff_lines.append(f'column: {key}, expected: {true_val}, got: {test_val}')
            except ValueError:
                if test_line[key] != true_line[key]:
                    diff_lines.append(f'column: {key}, expected: {true_line[key]}, got: {test_line[key]}')

    return '\n'.join(diff_lines)


def diff_cgns(test_cgns: Path, true_cgns: Path, cgns_tol: float) -> str:
    """Compare CGNS results against an expected CGSN file with tolerance

    Args:
        test_cgns (Path): Path to output CGNS file
        true_cgns (Path): Path to expected CGNS file
        cgns_tol (float): Tolerance for comparing floating-point values

    Returns:
        str: Diff output between result and expected CGNS files
    """
    my_env: dict = os.environ.copy()

    run_args: List[str] = ['cgnsdiff', '-d', '-t', f'{cgns_tol}', str(test_cgns), str(true_cgns)]
    proc = subprocess.run(' '.join(run_args),
                          shell=True,
                          stdout=subprocess.PIPE,
                          stderr=subprocess.PIPE,
                          env=my_env)

    return proc.stderr.decode('utf-8') + proc.stdout.decode('utf-8')


def diff_ascii(test_file: Path, true_file: Path, backend: str) -> str:
    """Compare ASCII results against an expected ASCII file

    Args:
        test_file (Path): Path to output ASCII file
        true_file (Path): Path to expected ASCII file

    Returns:
        str: Diff output between result and expected ASCII files
    """
    tmp_backend: str = backend.replace('/', '-')
    true_str: str = true_file.read_text().replace('{ceed_resource}', tmp_backend)
    diff = list(difflib.unified_diff(test_file.read_text().splitlines(keepends=True),
                                     true_str.splitlines(keepends=True),
                                     fromfile=str(test_file),
                                     tofile=str(true_file)))
    return ''.join(diff)


def test_case_output_string(test_case: TestCase, spec: TestSpec, mode: RunMode,
                            backend: str, test: str, index: int, verbose: bool) -> str:
    output_str = ''
    if mode is RunMode.TAP:
        # print incremental output if TAP mode
        if test_case.is_skipped():
            output_str += f'    ok {index} - {spec.name}, {backend} # SKIP {test_case.skipped[0]["message"]}\n'
        elif test_case.is_failure() or test_case.is_error():
            output_str += f'    not ok {index} - {spec.name}, {backend} ({test_case.elapsed_sec} s)\n'
        else:
            output_str += f'    ok {index} - {spec.name}, {backend} ({test_case.elapsed_sec} s)\n'
        if test_case.is_failure() or test_case.is_error() or verbose:
            output_str += f'      ---\n'
            if spec.only:
                output_str += f'      only: {",".join(spec.only)}\n'
            output_str += f'      args: {test_case.args}\n'
            if spec.csv_ztol > 0:
                output_str += f'      csv_ztol: {spec.csv_ztol}\n'
            if spec.csv_rtol > 0:
                output_str += f'      csv_rtol: {spec.csv_rtol}\n'
            if spec.cgns_tol > 0:
                output_str += f'      cgns_tol: {spec.cgns_tol}\n'
            for k, v in spec.key_values.items():
                output_str += f'      {k}: {v}\n'
            if test_case.is_error():
                output_str += f'      error: {test_case.errors[0]["message"]}\n'
            if test_case.is_failure():
                output_str += f'      failures:\n'
                for i, failure in enumerate(test_case.failures):
                    output_str += f'        -\n'
                    output_str += f'          message: {failure["message"]}\n'
                    if failure["output"]:
                        out = failure["output"].strip().replace('\n', '\n            ')
                        output_str += f'          output: |\n            {out}\n'
            output_str += f'      ...\n'
    else:
        # print error or failure information if JUNIT mode
        if test_case.is_error() or test_case.is_failure():
            output_str += f'Test: {test} {spec.name}\n'
            output_str += f'  $ {test_case.args}\n'
            if test_case.is_error():
                output_str += 'ERROR: {}\n'.format((test_case.errors[0]['message'] or 'NO MESSAGE').strip())
                output_str += 'Output: \n{}\n'.format((test_case.errors[0]['output'] or 'NO MESSAGE').strip())
            if test_case.is_failure():
                for failure in test_case.failures:
                    output_str += 'FAIL: {}\n'.format((failure['message'] or 'NO MESSAGE').strip())
                    output_str += 'Output: \n{}\n'.format((failure['output'] or 'NO MESSAGE').strip())
    return output_str


def save_failure_artifact(suite_spec: SuiteSpec, file: Path) -> Path:
    """Attach a file to a test case

    Args:
        test_case (TestCase): Test case to attach the file to
        file (Path): Path to the file to attach
    """
    save_path: Path = suite_spec.test_failure_artifacts_path / file.name
    shutil.copyfile(file, save_path)
    return save_path


def run_test(index: int, test: str, spec: TestSpec, backend: str,
             mode: RunMode, nproc: int, suite_spec: SuiteSpec, verbose: bool = False) -> TestCase:
    """Run a single test case and backend combination

    Args:
        index (int): Index of backend for current spec
        test (str): Path to test
        spec (TestSpec): Specification of test case
        backend (str): CEED backend
        mode (RunMode): Output mode
        nproc (int): Number of MPI processes to use when running test case
        suite_spec (SuiteSpec): Specification of test suite
        verbose (bool, optional): Print detailed output for all runs, not just failures. Defaults to False.

    Returns:
        TestCase: Test case result
    """
    source_path: Path = suite_spec.get_source_path(test)
    run_args: List = [f'{suite_spec.get_run_path(test)}', *map(str, spec.args)]

    if '{ceed_resource}' in run_args:
        run_args[run_args.index('{ceed_resource}')] = backend
    for i, arg in enumerate(run_args):
        if '{ceed_resource}' in arg:
            run_args[i] = arg.replace('{ceed_resource}', backend.replace('/', '-'))
    if '{nproc}' in run_args:
        run_args[run_args.index('{nproc}')] = f'{nproc}'
    elif nproc > 1 and source_path.suffix != '.py':
        run_args = ['mpiexec', '-n', f'{nproc}', *run_args]

    # run test
    skip_reason: Optional[str] = suite_spec.check_pre_skip(test, spec, backend, nproc)
    if skip_reason:
        test_case: TestCase = TestCase(f'{test}, "{spec.name}", n{nproc}, {backend}',
                                       elapsed_sec=0,
                                       timestamp=time.strftime('%Y-%m-%d %H:%M:%S %Z', time.localtime()),
                                       stdout='',
                                       stderr='',
                                       category=spec.name,)
        test_case.add_skipped_info(skip_reason)
    else:
        start: float = time.time()
        proc = subprocess.run(' '.join(str(arg) for arg in run_args),
                              shell=True,
                              stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE,
                              env=my_env)

        test_case = TestCase(f'{test}, "{spec.name}", n{nproc}, {backend}',
                             classname=source_path.parent,
                             elapsed_sec=time.time() - start,
                             timestamp=time.strftime('%Y-%m-%d %H:%M:%S %Z', time.localtime(start)),
                             stdout=proc.stdout.decode('utf-8'),
                             stderr=proc.stderr.decode('utf-8'),
                             allow_multiple_subelements=True,
                             category=spec.name,)
        ref_csvs: List[Path] = []
        ref_ascii: List[Path] = []
        output_files: List[str] = [arg.split(':')[1] for arg in run_args if arg.startswith('ascii:')]
        if output_files:
            ref_csvs = [suite_spec.get_output_path(test, file)
                        for file in output_files if file.endswith('.csv')]
            ref_ascii = [suite_spec.get_output_path(test, file)
                         for file in output_files if not file.endswith('.csv')]
        ref_cgns: List[Path] = []
        output_files = [arg.split(':')[1] for arg in run_args if arg.startswith('cgns:')]
        if output_files:
            ref_cgns = [suite_spec.get_output_path(test, file) for file in output_files]
        ref_stdout: Path = suite_spec.get_output_path(test, test + '.out')
        suite_spec.post_test_hook(test, spec, backend)

    # check allowed failures
    if not test_case.is_skipped() and test_case.stderr:
        skip_reason: Optional[str] = suite_spec.check_post_skip(test, spec, backend, test_case.stderr)
        if skip_reason:
            test_case.add_skipped_info(skip_reason)

    # check required failures
    if not test_case.is_skipped():
        required_message, did_fail = suite_spec.check_required_failure(
            test, spec, backend, test_case.stderr)
        if required_message and did_fail:
            test_case.status = f'fails with required: {required_message}'
        elif required_message:
            test_case.add_failure_info(f'required failure missing: {required_message}')

    # classify other results
    if not test_case.is_skipped() and not test_case.status:
        if test_case.stderr:
            test_case.add_failure_info('stderr', test_case.stderr)
        if proc.returncode != 0:
            test_case.add_error_info(f'returncode = {proc.returncode}')
        if ref_stdout.is_file():
            diff = list(difflib.unified_diff(ref_stdout.read_text().splitlines(keepends=True),
                                             test_case.stdout.splitlines(keepends=True),
                                             fromfile=str(ref_stdout),
                                             tofile='New'))
            if diff:
                test_case.add_failure_info('stdout', output=''.join(diff))
        elif test_case.stdout and not suite_spec.check_allowed_stdout(test):
            test_case.add_failure_info('stdout', output=test_case.stdout)
        # expected CSV output
        for ref_csv in ref_csvs:
            csv_name = ref_csv.name
            out_file = Path.cwd() / csv_name
            if not ref_csv.is_file():
                # remove _{ceed_backend} from path name
                ref_csv = (ref_csv.parent / ref_csv.name.rsplit('_', 1)[0]).with_suffix('.csv')
            if not ref_csv.is_file():
                test_case.add_failure_info('csv', output=f'{ref_csv} not found')
            elif not out_file.is_file():
                test_case.add_failure_info('csv', output=f'{out_file} not found')
            else:
                csv_ztol: float = spec.csv_ztol if spec.csv_ztol > 0 else suite_spec.csv_ztol
                csv_rtol: float = spec.csv_rtol if spec.csv_rtol > 0 else suite_spec.csv_rtol
                diff = diff_csv(
                    out_file,
                    ref_csv,
                    csv_ztol,
                    csv_rtol,
                    suite_spec.csv_comment_str,
                    suite_spec.csv_comment_diff_fn)
                if diff:
                    save_path: Path = suite_spec.test_failure_artifacts_path / csv_name
                    shutil.move(out_file, save_path)
                    test_case.add_failure_info(f'csv: {save_path}', output=diff)
                else:
                    out_file.unlink()
        # expected CGNS output
        for ref_cgn in ref_cgns:
            cgn_name = ref_cgn.name
            out_file = Path.cwd() / cgn_name
            if not ref_cgn.is_file():
                # remove _{ceed_backend} from path name
                ref_cgn = (ref_cgn.parent / ref_cgn.name.rsplit('_', 1)[0]).with_suffix('.cgns')
            if not ref_cgn.is_file():
                test_case.add_failure_info('cgns', output=f'{ref_cgn} not found')
            elif not out_file.is_file():
                test_case.add_failure_info('cgns', output=f'{out_file} not found')
            else:
                cgns_tol = spec.cgns_tol if spec.cgns_tol > 0 else suite_spec.cgns_tol
                diff = diff_cgns(out_file, ref_cgn, cgns_tol=cgns_tol)
                if diff:
                    save_path: Path = suite_spec.test_failure_artifacts_path / cgn_name
                    shutil.move(out_file, save_path)
                    test_case.add_failure_info(f'cgns: {save_path}', output=diff)
                else:
                    out_file.unlink()
        # expected ASCII output
        for ref_file in ref_ascii:
            ref_name = ref_file.name
            out_file = Path.cwd() / ref_name
            if not ref_file.is_file():
                # remove _{ceed_backend} from path name
                ref_file = (ref_file.parent / ref_file.name.rsplit('_', 1)[0]).with_suffix(ref_file.suffix)
            if not ref_file.is_file():
                test_case.add_failure_info('ascii', output=f'{ref_file} not found')
            elif not out_file.is_file():
                test_case.add_failure_info('ascii', output=f'{out_file} not found')
            else:
                diff = diff_ascii(out_file, ref_file, backend)
                if diff:
                    save_path: Path = suite_spec.test_failure_artifacts_path / ref_name
                    shutil.move(out_file, save_path)
                    test_case.add_failure_info(f'ascii: {save_path}', output=diff)
                else:
                    out_file.unlink()

    # store result
    test_case.args = ' '.join(str(arg) for arg in run_args)
    output_str = test_case_output_string(test_case, spec, mode, backend, test, index, verbose)

    return test_case, output_str


def init_process():
    """Initialize multiprocessing process"""
    # set up error handler
    global my_env
    my_env = os.environ.copy()
    my_env['CEED_ERROR_HANDLER'] = 'exit'


def run_tests(test: str, ceed_backends: List[str], mode: RunMode, nproc: int,
              suite_spec: SuiteSpec, pool_size: int = 1, search: str = ".*", verbose: bool = False) -> TestSuite:
    """Run all test cases for `test` with each of the provided `ceed_backends`

    Args:
        test (str): Name of test
        ceed_backends (List[str]): List of libCEED backends
        mode (RunMode): Output mode, either `RunMode.TAP` or `RunMode.JUNIT`
        nproc (int): Number of MPI processes to use when running each test case
        suite_spec (SuiteSpec): Object defining required methods for running tests
        pool_size (int, optional): Number of processes to use when running tests in parallel. Defaults to 1.
        search (str, optional): Regular expression used to match tests. Defaults to ".*".
        verbose (bool, optional): Print detailed output for all runs, not just failures. Defaults to False.

    Returns:
        TestSuite: JUnit `TestSuite` containing results of all test cases
    """
    test_specs: List[TestSpec] = [
        t for t in get_test_args(suite_spec.get_source_path(test)) if re.search(search, t.name, re.IGNORECASE)
    ]
    suite_spec.test_failure_artifacts_path.mkdir(parents=True, exist_ok=True)
    if mode is RunMode.TAP:
        print('TAP version 13')
        print(f'1..{len(test_specs)}')

    with mp.Pool(processes=pool_size, initializer=init_process) as pool:
        async_outputs: List[List[mp.pool.AsyncResult]] = [
            [pool.apply_async(run_test, (i, test, spec, backend, mode, nproc, suite_spec, verbose))
             for (i, backend) in enumerate(ceed_backends, start=1)]
            for spec in test_specs
        ]

        test_cases = []
        for (i, subtest) in enumerate(async_outputs, start=1):
            is_new_subtest = True
            subtest_ok = True
            for async_output in subtest:
                test_case, print_output = async_output.get()
                test_cases.append(test_case)
                if is_new_subtest and mode == RunMode.TAP:
                    is_new_subtest = False
                    print(f'# Subtest: {test_case.category}')
                    print(f'    1..{len(ceed_backends)}')
                print(print_output, end='')
                if test_case.is_failure() or test_case.is_error():
                    subtest_ok = False
            if mode == RunMode.TAP:
                print(f'{"" if subtest_ok else "not "}ok {i} - {test_case.category}')

    return TestSuite(test, test_cases)


def write_junit_xml(test_suite: TestSuite, batch: str = '') -> None:
    """Write a JUnit XML file containing the results of a `TestSuite`

    Args:
        test_suite (TestSuite): JUnit `TestSuite` to write
        batch (str): Name of JUnit batch, defaults to empty string
    """
    output_file = Path('build') / (f'{test_suite.name}{batch}.junit')
    output_file.write_text(to_xml_report_string([test_suite]))


def has_failures(test_suite: TestSuite) -> bool:
    """Check whether any test cases in a `TestSuite` failed

    Args:
        test_suite (TestSuite): JUnit `TestSuite` to check

    Returns:
        bool: True if any test cases failed
    """
    return any(c.is_failure() or c.is_error() for c in test_suite.test_cases)