-
Notifications
You must be signed in to change notification settings - Fork 49
/
Copy pathjunit.py
executable file
·150 lines (132 loc) · 7.18 KB
/
junit.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
#!/usr/bin/env python3
import os
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '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:]] for line in open(file).readlines()
if line.startswith('//TESTARGS')], [])
elif os.path.splitext(file)[1] == '.usr':
return sum([[line.split()[2:]] for line in open(file).readlines()
if line.startswith('C 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')
def get_testargs(test):
source = get_source(test)
if source is None:
return [['{ceed_resource}']]
return parse_testargs(source)
def check_required_failure(case, stderr, required):
if required in stderr:
case.status = 'fails with required: {}'.format(required)
else:
case.add_failure_info('required: {}'.format(required))
def contains_any(resource, substrings):
return any((sub in resource for sub in substrings))
def skip_rule(test, resource):
return any((
test.startswith('fluids-') and contains_any(resource, ['occa', 'gpu']) and not contains_any(resource, ['/gpu/cuda/gen']),
test.startswith('solids-') and contains_any(resource, ['occa']),
test.startswith('petsc-multigrid') and contains_any(resource, ['occa']),
test.startswith('t506') and contains_any(resource, ['occa']),
test.startswith('t507') and contains_any(resource, ['occa']),
))
def run(test, backends):
import subprocess
import time
import difflib
allargs = get_testargs(test)
testcases = []
for args in allargs:
for ceed_resource in backends:
rargs = [os.path.join('build', test)] + args.copy()
rargs[rargs.index('{ceed_resource}')] = ceed_resource
if skip_rule(test, ceed_resource):
case = TestCase('{} {}'.format(test, ceed_resource),
elapsed_sec=0,
timestamp=time.strftime('%Y-%m-%d %H:%M:%S %Z', time.localtime(start)),
stdout='',
stderr='')
case.add_skipped_info('Pre-run skip rule')
else:
start = time.time()
proc = subprocess.run(rargs,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
proc.stdout = proc.stdout.decode('utf-8')
proc.stderr = proc.stderr.decode('utf-8')
case = TestCase('{} {}'.format(test, ceed_resource),
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')
if not case.is_skipped() and proc.stderr:
if 'OCCA backend failed to use' in proc.stderr:
case.add_skipped_info('occa mode not supported {} {}'.format(test, ceed_resource))
elif 'Backend does not implement' in proc.stderr:
case.add_skipped_info('not implemented {} {}'.format(test, ceed_resource))
elif 'Can only provide to HOST memory' in proc.stderr:
case.add_skipped_info('device memory not supported {} {}'.format(test, ceed_resource))
if not case.is_skipped():
if test[:4] in 't110 t111 t112 t113 t114'.split():
check_required_failure(case, proc.stderr, 'Cannot grant CeedVector array access')
if test[:4] in 't115'.split():
check_required_failure(case, proc.stderr, 'Cannot grant CeedVector read-only array access, the access lock is already in use')
if test[:4] in 't116'.split():
check_required_failure(case, proc.stderr, 'Cannot destroy CeedVector, the access lock is in use')
if test[:4] in 't117'.split():
check_required_failure(case, proc.stderr, 'Cannot restore CeedVector array access, access was not granted')
if test[:4] in 't118'.split():
check_required_failure(case, proc.stderr, 'Cannot sync CeedVector, the access lock is already in use')
if test[:4] in 't215'.split():
check_required_failure(case, proc.stderr, 'Cannot destroy CeedElemRestriction, a process has read access to the offset data')
if test[:4] in 't303'.split():
check_required_failure(case, proc.stderr, 'Length of input/output vectors incompatible with basis dimensions')
if not case.is_skipped() and not case.status:
if proc.stderr:
case.add_failure_info('stderr', proc.stderr)
elif proc.returncode != 0:
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'))
if diff:
case.add_failure_info('stdout', output=''.join(diff))
elif proc.stdout and test[:4] not in 't003':
case.add_failure_info('stdout', output=proc.stdout)
testcases.append(case)
return TestSuite(test, testcases)
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser('Test runner with JUnit output')
parser.add_argument('--output', help='Output file to write test', default=None)
parser.add_argument('--gather', help='Gather all *.junit files into XML', action='store_true')
parser.add_argument('test', help='Test executable', nargs='?')
args = parser.parse_args()
if args.gather:
gather()
else:
backends = os.environ['BACKENDS'].split()
result = run(args.test, backends)
output = (os.path.join('build', args.test + '.junit')
if args.output is None
else args.output)
with open(output, 'w') as fd:
TestSuite.to_file(fd, [result])