120 lines
5.9 KiB
Python
120 lines
5.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Reproducible standard-library Python line and GCC core branch coverage.
|
|
|
|
No ROS emulation is labeled a middleware integration test. GCC branch counts
|
|
include compiler-generated exception/short-circuit branches; the scenario
|
|
campaign is a separate functional metric. Run from any working directory.
|
|
"""
|
|
import argparse
|
|
import gzip
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import trace
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
CORE_TESTS = tuple(path.stem for path in sorted((ROOT / 'core/tests').glob('*_test.cpp')))
|
|
|
|
|
|
def executable_lines(path):
|
|
# Python 3.14 can include artificial bytecode positions without a source
|
|
# line; these are not executable source statements and cannot be sorted.
|
|
return {line for line in trace._find_executable_linenos(str(path))
|
|
if type(line) is int and line > 0}
|
|
|
|
|
|
def production(path):
|
|
relative = Path(path).resolve().relative_to(ROOT)
|
|
return relative.parts[0] in ('coordinator', 'robobrain', 'navigation_gateway', 'ros2') and not any(
|
|
p in ('tests', 'test', 'tools', '__pycache__') for p in relative.parts) and relative.name not in ('setup.py',)
|
|
|
|
|
|
def python_lines(destination):
|
|
os.chdir(ROOT)
|
|
sys.path[:0] = [str(ROOT), str(ROOT / 'tests')]
|
|
suite = unittest.defaultTestLoader.discover(str(ROOT / 'tests'), pattern='test_*.py')
|
|
tracer = trace.Trace(count=True, trace=False, ignoredirs=[sys.prefix, sys.exec_prefix])
|
|
# Trace.runfunc covers the executing test thread. Background thread lines
|
|
# are deliberately not counted as covered without an observed trace event.
|
|
result = tracer.runfunc(unittest.TextTestRunner(verbosity=1).run, suite)
|
|
counts = tracer.results().counts
|
|
observed = {}
|
|
resolved = {}
|
|
for (filename, line), count in counts.items():
|
|
if count <= 0:
|
|
continue
|
|
if filename not in resolved:
|
|
resolved[filename] = Path(filename).resolve()
|
|
observed.setdefault(resolved[filename], set()).add(line)
|
|
files = []
|
|
for path in sorted(ROOT.rglob('*.py')):
|
|
if not production(path):
|
|
continue
|
|
statements = executable_lines(path)
|
|
hit = observed.get(path, set()) & statements
|
|
files.append(dict(path=str(path.relative_to(ROOT)), statements=len(statements),
|
|
covered=len(hit), missing=sorted(statements-hit)))
|
|
data = dict(metric='Python executable source lines; main test thread only',
|
|
tests_run=result.testsRun, failures=len(result.failures), errors=len(result.errors),
|
|
statements=sum(f['statements'] for f in files), covered=sum(f['covered'] for f in files), files=files)
|
|
(destination / 'python_lines.json').write_text(json.dumps(data, indent=2)+'\n')
|
|
if not result.wasSuccessful():
|
|
raise RuntimeError('Python suite failed under instrumentation')
|
|
return data
|
|
|
|
|
|
def cpp_branches(destination):
|
|
build = destination / 'gcc'; build.mkdir(exist_ok=True)
|
|
# Each execution is a fresh measurement, not an accumulation of old runs.
|
|
for p in build.glob('*.gcda'): p.unlink()
|
|
compiler = os.environ.get('CXX', 'g++')
|
|
flags = ['-std=c++17', '-O0', '-g', '--coverage', '-Wall', '-Wextra', '-Werror',
|
|
'-pthread', '-UNDEBUG', '-I'+str(ROOT / 'core/include')]
|
|
objects = []
|
|
for source in ('core', 'workflow'):
|
|
obj = build / (source+'.o')
|
|
subprocess.run([compiler, *flags, '-c', str(ROOT / 'core/src' / (source+'.cpp')), '-o', str(obj)], check=True)
|
|
objects.append(str(obj))
|
|
runs = []
|
|
for name in CORE_TESTS:
|
|
exe = build / name
|
|
subprocess.run([compiler, *flags, str(ROOT / 'core/tests' / (name+'.cpp')), *objects, '-o', str(exe)], check=True)
|
|
completed = subprocess.run([str(exe)], check=True, capture_output=True, text=True)
|
|
runs.append(dict(test=name, exit_code=completed.returncode, stdout=completed.stdout))
|
|
for obj in objects:
|
|
subprocess.run(['gcov', '-j', '-b', '-c', obj], cwd=build, check=True, stdout=subprocess.DEVNULL)
|
|
files = []
|
|
for archive in build.glob('*.gcov.json.gz'):
|
|
with gzip.open(archive, 'rt') as stream: data=json.load(stream)
|
|
for source in data['files']:
|
|
path=Path(source['file']).resolve()
|
|
if path.parent != ROOT / 'core/src': continue
|
|
lines=source['lines'];branches=[b for line in lines for b in line.get('branches', [])]
|
|
files.append(dict(path=str(path.relative_to(ROOT)), statements=len(lines),
|
|
covered=sum(line['count']>0 for line in lines),
|
|
branches=len(branches), branches_taken=sum(b['count']>0 for b in branches),
|
|
missing_lines=[line['line_number'] for line in lines if not line['count']],
|
|
lines_with_untaken_branches=[line['line_number'] for line in lines if any(not b['count'] for b in line.get('branches', []))]))
|
|
data=dict(metric='GCC source branches including exception and short-circuit arcs', files=files, runs=runs,
|
|
statements=sum(f['statements'] for f in files), covered=sum(f['covered'] for f in files),
|
|
branches=sum(f['branches'] for f in files), branches_taken=sum(f['branches_taken'] for f in files))
|
|
(destination / 'cpp_branches.json').write_text(json.dumps(data, indent=2)+'\n')
|
|
return data
|
|
|
|
|
|
def main():
|
|
parser=argparse.ArgumentParser();parser.add_argument('--output', default=str(ROOT / 'build/coverage'))
|
|
args=parser.parse_args();out=Path(args.output).resolve();out.mkdir(parents=True, exist_ok=True)
|
|
py=python_lines(out);cpp=cpp_branches(out)
|
|
summary=dict(python={k:v for k,v in py.items() if k!='files'},
|
|
cpp={k:v for k,v in cpp.items() if k not in ('files','runs')},
|
|
ros_wire_execution=False, hardware_execution=False)
|
|
(out / 'summary.json').write_text(json.dumps(summary, indent=2)+'\n')
|
|
print(json.dumps(summary, indent=2))
|
|
|
|
|
|
if __name__=='__main__': main()
|