Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Change os.path to pathlib and simplify parsing #1263

Merged
merged 2 commits into from
Jul 20, 2023
Merged

Change os.path to pathlib and simplify parsing #1263

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
96cb76b
Change os.path to pathlib and simplify parsing
zatkins-dev Jul 19, 2023
def07d8
support for testargs without parens/name
zatkins-dev Jul 19, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
162 changes: 86 additions & 76 deletions tests/junit.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,65 +1,81 @@
#!/usr/bin/env python3

from dataclasses import dataclass, field
import difflib
from itertools import combinations
import os
from pathlib import Path
import re
import subprocess
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), 'junit-xml')))
import time
sys.path.insert(0, str(Path(__file__).parent / "junit-xml"))
from junit_xml import TestCase, TestSuite


def parse_testargs(file):
if os.path.splitext(file)[1] in ['.c', '.cpp']:
return sum([[[line.split()[1:], [line.split()[0].strip('//TESTARGS(name=').strip(')')]]]
for line in open(file).readlines()
if line.startswith('//TESTARGS')], [])
elif os.path.splitext(file)[1] == '.usr':
return sum([[[line.split()[1:], [line.split()[0].strip('C_TESTARGS(name=').strip(')')]]]
for line in open(file).readlines()
if line.startswith('C_TESTARGS')], [])
elif os.path.splitext(file)[1] in ['.f90']:
return sum([[[line.split()[1:], [line.split()[0].strip('C_TESTARGS(name=').strip(')')]]]
for line in open(file).readlines()
if line.startswith('! TESTARGS')], [])
raise RuntimeError('Unrecognized extension for file: {}'.format(file))


def get_source(test):
if test.startswith('petsc-'):
return os.path.join('examples', 'petsc', test[6:] + '.c')
elif test.startswith('mfem-'):
return os.path.join('examples', 'mfem', test[5:] + '.cpp')
elif test.startswith('nek-'):
return os.path.join('examples', 'nek', 'bps', test[4:] + '.usr')
elif test.startswith('fluids-'):
return os.path.join('examples', 'fluids', test[7:] + '.c')
elif test.startswith('solids-'):
return os.path.join('examples', 'solids', test[7:] + '.c')
elif test.startswith('ex'):
return os.path.join('examples', 'ceed', test + '.c')
elif test.endswith('-f'):
return os.path.join('tests', test + '.f90')
@dataclass
class TestSpec:
name: str
only: list = field(default_factory=list)
args: list = field(default_factory=list)


def parse_test_line(line: str) -> TestSpec:
args = line.strip().split()
test_args = args[0][args[0].index('TESTARGS(')+9:args[0].rindex(')')]
# transform 'name="myname",only="serial,int32"' into {'name': 'myname', 'only': 'serial,int32'}
test_args = dict([''.join(t).split('=') for t in re.findall(r"""([^,=]+)(=)"([^"]*)\"""", test_args)])
constraints = test_args['only'].split(',') if 'only' in test_args else []
if len(args) > 1:
return TestSpec(name=test_args['name'], only=constraints, args=args[1:])
else:
return os.path.join('tests', test + '.c')
return TestSpec(name=test_args['name'], only=constraints)


def get_testargs(file : Path) -> list[TestSpec]:
if file.suffix in ['.c', '.cpp']: comment_str = '//'
elif file.suffix in ['.py']: comment_str = '#'
elif file.suffix in ['.usr']: comment_str = 'C_'
elif file.suffix in ['.f90']: comment_str = '! '
else: raise RuntimeError(f'Unrecognized extension for file: {file}')

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

def get_testargs(source):
args = parse_testargs(source)
if not args:
return [(['{ceed_resource}'], [''])]
return args

def get_source(test: str) -> Path:
prefix, rest = test.split('-', 1)
if prefix == 'petsc':
return (Path('examples') / 'petsc' / rest).with_suffix('.c')
elif prefix == 'mfem':
return (Path('examples') / 'mfem' / rest).with_suffix('.cpp')
elif prefix == 'nek':
return (Path('examples') / 'nek' / 'bps' / rest).with_suffix('.usr')
elif prefix == 'fluids':
return (Path('examples') / 'fluids' / rest).with_suffix('.c')
elif prefix == 'solids':
return (Path('examples') / 'solids' / rest).with_suffix('.c')
elif test.startswith('ex'):
return (Path('examples') / 'ceed' / test).with_suffix('.c')
elif test.endswith('-f'):
return (Path('tests') / test).with_suffix('.f90')
else:
return (Path('tests') / test).with_suffix('.c')


def check_required_failure(test_case, stderr, required):
def check_required_failure(test_case: TestCase, stderr: str, required: str) -> None:
if required in stderr:
test_case.status = 'fails with required: {}'.format(required)
else:
test_case.add_failure_info('required: {}'.format(required))


def contains_any(resource, substrings):
def contains_any(resource: str, substrings: list[str]) -> bool:
return any((sub in resource for sub in substrings))


def skip_rule(test, resource):
def skip_rule(test: str, resource: str) -> bool:
return any((
test.startswith('t4') and contains_any(resource, ['occa']),
test.startswith('t5') and contains_any(resource, ['occa']),
Expand All @@ -71,31 +87,28 @@ def skip_rule(test, resource):
test.startswith('solids-') and contains_any(resource, ['occa']),
test.startswith('t318') and contains_any(resource, ['/gpu/cuda/ref']),
test.startswith('t506') and contains_any(resource, ['/gpu/cuda/shared']),
))
))


def run(test, backends, mode):
import subprocess
import time
import difflib
def run(test: str, backends: list[str], mode: str) -> TestSuite:
source = get_source(test)
all_args = get_testargs(source)
test_specs = get_testargs(source)

if mode.lower() == "tap":
print('1..' + str(len(all_args) * len(backends)))
print('1..' + str(len(test_specs) * len(backends)))

test_cases = []
my_env = os.environ.copy()
my_env["CEED_ERROR_HANDLER"] = 'exit'
index = 1
for args, name in all_args:
for spec in test_specs:
for ceed_resource in backends:
rargs = [os.path.join('build', test)] + args.copy()
rargs = [str(Path('build') / test), *spec.args]
rargs[rargs.index('{ceed_resource}')] = ceed_resource

# run test
if skip_rule(test, ceed_resource):
test_case = TestCase('{} {}'.format(test, ceed_resource),
test_case = TestCase(f'{test} {ceed_resource}',
elapsed_sec=0,
timestamp=time.strftime('%Y-%m-%d %H:%M:%S %Z', time.localtime()),
stdout='',
Expand All @@ -110,13 +123,13 @@ def run(test, backends, mode):
proc.stdout = proc.stdout.decode('utf-8')
proc.stderr = proc.stderr.decode('utf-8')

test_case = TestCase('{} {} {}'.format(test, *name, ceed_resource),
classname=os.path.dirname(source),
test_case = TestCase(f'{test} {spec.name} {ceed_resource}',
classname=source.parent,
elapsed_sec=time.time() - start,
timestamp=time.strftime('%Y-%m-%d %H:%M:%S %Z', time.localtime(start)),
stdout=proc.stdout,
stderr=proc.stderr)
ref_stdout = os.path.join('tests/output', test + '.out')
ref_stdout = (Path('tests') / 'output' / test).with_suffix('.out')

# check for allowed errors
if not test_case.is_skipped() and proc.stderr:
Expand All @@ -133,41 +146,40 @@ def run(test, backends, mode):

# check required failures
if not test_case.is_skipped():
if test[:4] in 't006 t007'.split():
if test[:4] in ['t006', 't007']:
check_required_failure(test_case, proc.stderr, 'No suitable backend:')
if test[:4] in 't008'.split():
if test[:4] in ['t008']:
check_required_failure(test_case, proc.stderr, 'Available backend resources:')
if test[:4] in 't110 t111 t112 t113 t114'.split():
if test[:4] in ['t110', 't111', 't112', 't113', 't114']:
check_required_failure(test_case, proc.stderr, 'Cannot grant CeedVector array access')
if test[:4] in 't115'.split():
if test[:4] in ['t115']:
check_required_failure(test_case, proc.stderr, 'Cannot grant CeedVector read-only array access, the access lock is already in use')
if test[:4] in 't116'.split():
if test[:4] in ['t116']:
check_required_failure(test_case, proc.stderr, 'Cannot destroy CeedVector, the writable access lock is in use')
if test[:4] in 't117'.split():
if test[:4] in ['t117']:
check_required_failure(test_case, proc.stderr, 'Cannot restore CeedVector array access, access was not granted')
if test[:4] in 't118'.split():
if test[:4] in ['t118']:
check_required_failure(test_case, proc.stderr, 'Cannot sync CeedVector, the access lock is already in use')
if test[:4] in 't215'.split():
if test[:4] in ['t215']:
check_required_failure(test_case, proc.stderr, 'Cannot destroy CeedElemRestriction, a process has read access to the offset data')
if test[:4] in 't303'.split():
if test[:4] in ['t303']:
check_required_failure(test_case, proc.stderr, 'Length of input/output vectors incompatible with basis dimensions')
if test[:4] in 't408'.split():
if test[:4] in ['t408']:
check_required_failure(test_case, proc.stderr, 'CeedQFunctionContextGetData(): Cannot grant CeedQFunctionContext data access, a process has read access')
if test[:4] in 't409'.split() and contains_any(ceed_resource, ['memcheck']):
if test[:4] in ['t409'] and contains_any(ceed_resource, ['memcheck']):
check_required_failure(test_case, proc.stderr, 'Context data changed while accessed in read-only mode')

# classify other results
if not test_case.is_skipped() and not test_case.status:
if proc.stderr:
test_case.add_failure_info('stderr', proc.stderr)
elif proc.returncode != 0:
test_case.add_error_info('returncode = {}'.format(proc.returncode))
elif os.path.isfile(ref_stdout):
with open(ref_stdout) as ref:
diff = list(difflib.unified_diff(ref.readlines(),
proc.stdout.splitlines(keepends=True),
fromfile=ref_stdout,
tofile='New'))
test_case.add_error_info(f'returncode = {proc.returncode}')
elif ref_stdout.is_file():
diff = list(difflib.unified_diff(ref_stdout.read_text().splitlines(keepends=True),
proc.stdout.splitlines(keepends=True),
fromfile=str(ref_stdout),
tofile='New'))
if diff:
test_case.add_failure_info('stdout', output=''.join(diff))
elif proc.stdout and test[:4] not in 't003':
Expand Down Expand Up @@ -235,11 +247,9 @@ def run(test, backends, mode):
junit_batch = '-' + os.environ['JUNIT_BATCH']
except:
pass
output = (os.path.join('build', args.test + junit_batch + '.junit')
if args.output is None
else args.output)
output = Path('build') / (args.test + junit_batch + '.junit') if args.output is None else Path(args.output)

with open(output, 'w') as fd:
with output.open('w') as fd:
TestSuite.to_file(fd, [result])
elif args.mode.lower() != "tap":
raise Exception("output mode not recognized")
Expand Down