behavior-tree/tools/coverage_report.py

#!/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', '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()