#!/usr/bin/env python3
# Copyright 2026 WebAssembly Community Group participants
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Multi-metric differential test coverage reporter for Binaryen.
Intersects Git or Jujutsu (`jj`) diff hunks with Clang/LLVM source-based code
coverage (`llvm-cov export`) to report Line, Region, Branch, MC/DC, and Function
coverage specifically for the code modified in a change. With -a / --annotate,
also prints the diff annotated with coverage information.
Prints instructions for configuring a coverage reporting build and collecting
coverage information if a coverage build cannot be found automatically.
"""
import argparse
import fnmatch
import glob
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
from dataclasses import dataclass, field
SCRIPT_PATH = os.path.relpath(os.path.abspath(__file__))
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
CPP_EXTENSIONS = {'.c', '.cpp', '.h', '.inc'}
HUNK_RE = re.compile(r'^@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@')
UNREACHABLE_RE = re.compile(
r'\b(WASM_UNREACHABLE|WASM_BUILTIN_UNREACHABLE|__builtin_unreachable)\b'
r'|LCOV_EXCL_LINE|GCOVR_EXCL_LINE|COV_EXCL_LINE',
)
DEFAULT_LABEL_RE = re.compile(r'^\s*default\s*:\s*(//.*|/\*.*\*/)?$')
CLOSING_BRACE_RE = re.compile(r'^\s*\}\s*(//.*|/\*.*\*/)?$')
# CounterMappingRegion::RegionKind values in llvm-cov JSON export.
REGION_KIND_CODE = 0
@dataclass
class Segment:
line: int
col: int
count: int
has_count: bool
is_region_entry: bool
is_gap_region: bool
@dataclass
class LineStat:
line: int
mapped: bool
count: int
@dataclass
class CodeRegion:
line_start: int
col_start: int
line_end: int
col_end: int
count: int
@dataclass
class BranchRecord:
line_start: int
col_start: int
line_end: int
col_end: int
true_count: int
false_count: int
@dataclass
class MCDCRecord:
line_start: int
col_start: int
line_end: int
col_end: int
conditions: list[bool]
@dataclass
class FunctionSpan:
name: str
demangled_name: str
line_start: int
line_end: int
count: int
@dataclass
class MetricCounts:
covered: int = 0
total: int = 0
def add(self, covered: int, total: int):
self.covered += covered
self.total += total
@property
def percent(self) -> float | None:
if self.total == 0:
return None
return (self.covered * 100.0) / self.total
def format_summary(self) -> str:
if self.total == 0:
return 'N/A (0 in diff)'
return f'{self.covered} / {self.total} ({self.percent:.1f}%)'
@dataclass
class FileDiffCoverage:
rel_path: str
abs_path: str
diff_lines: list[int]
hunks: list[tuple[int, int]]
lines: MetricCounts = field(default_factory=MetricCounts)
regions: MetricCounts = field(default_factory=MetricCounts)
branches: MetricCounts = field(default_factory=MetricCounts)
mcdc: MetricCounts = field(default_factory=MetricCounts)
functions: MetricCounts = field(default_factory=MetricCounts)
non_executable_diff_lines: int = 0
excluded_unreachable_lines: int = 0
line_stats: dict[int, LineStat] = field(default_factory=dict)
uncovered_functions: list[FunctionSpan] = field(default_factory=list)
uncovered_lines: list[int] = field(default_factory=list)
uncovered_subregions: list[CodeRegion] = field(default_factory=list)
uncovered_branches: list[BranchRecord] = field(default_factory=list)
uncovered_mcdc: list[MCDCRecord] = field(default_factory=list)
in_coverage_map: bool = True
def find_llvm_tool(tool_name: str, explicit_path: str | None = None) -> str | None:
if explicit_path:
return explicit_path if shutil.which(explicit_path) or os.path.isfile(explicit_path) else None
llvm_version = os.environ.get('LLVM_VERSION')
if llvm_version:
found = shutil.which(f'{tool_name}-{llvm_version}')
if found:
return found
found = shutil.which(tool_name)
if found:
return found
# Search PATH for versioned binaries (e.g. llvm-cov-22) and pick the newest.
pattern = re.compile(rf'^{re.escape(tool_name)}-(\d+)$')
versioned: list[tuple[int, str]] = []
for path_dir in os.environ.get('PATH', '').split(os.pathsep):
if not path_dir or not os.path.isdir(path_dir):
continue
try:
for entry in os.listdir(path_dir):
m = pattern.match(entry)
if m:
full = os.path.join(path_dir, entry)
if os.path.isfile(full) and os.access(full, os.X_OK):
versioned.append((int(m.group(1)), full))
except OSError:
continue
if versioned:
versioned.sort(key=lambda item: item[0], reverse=True)
return versioned[0][1]
return None
def parse_cmake_cache(cache_path: str) -> dict[str, str]:
entries: dict[str, str] = {}
if not os.path.isfile(cache_path):
return entries
with open(cache_path, errors='replace') as f:
for raw_line in f:
line = raw_line.strip()
if not line or line.startswith(('#', '//')):
continue
if ':' in line and '=' in line:
key_type, val = line.split('=', 1)
key = key_type.split(':', 1)[0]
entries[key] = val
return entries
def check_build_dir_coverage_config(build_dir: str) -> tuple[bool, str, dict[str, str]]:
"""Return (is_valid, reason, cmake_cache_dict)."""
cache_path = os.path.join(build_dir, 'CMakeCache.txt')
if not os.path.isfile(cache_path):
return False, f"No CMakeCache.txt found in '{build_dir}'.", {}
cache = parse_cmake_cache(cache_path)
cxx = cache.get('CMAKE_CXX_COMPILER', '')
build_type = cache.get('CMAKE_BUILD_TYPE', 'Release').upper()
cxx_flags = ' '.join([
cache.get('CMAKE_CXX_FLAGS', ''),
cache.get(f'CMAKE_CXX_FLAGS_{build_type}', ''),
])
c_flags = ' '.join([
cache.get('CMAKE_C_FLAGS', ''),
cache.get(f'CMAKE_C_FLAGS_{build_type}', ''),
])
cxx_base = os.path.basename(cxx)
if 'clang' not in cxx_base:
return (
False,
f"Build in '{build_dir}' uses compiler '{cxx or 'unknown'}' instead of Clang "
"(Clang is required for source-based coverage mapping).",
cache,
)
required_flags = ('-fprofile-instr-generate', '-fcoverage-mapping')
missing_cxx = [fl for fl in required_flags if fl not in cxx_flags]
missing_c = [fl for fl in required_flags if fl not in c_flags]
if missing_cxx or missing_c:
return (
False,
f"Build in '{build_dir}' is missing coverage flags "
f"({', '.join(required_flags)}) in CMAKE_C_FLAGS / CMAKE_CXX_FLAGS.",
cache,
)
return True, 'OK', cache
def discover_build_dir(explicit_dir: str | None) -> tuple[str, bool, str, dict[str, str]]:
"""Find a coverage-configured build directory or report why candidates failed."""
if explicit_dir:
abs_dir = os.path.abspath(explicit_dir)
ok, reason, cache = check_build_dir_coverage_config(abs_dir)
return abs_dir, ok, reason, cache
env_dir = os.environ.get('BINARYEN_BUILD_DIR')
if env_dir:
abs_dir = os.path.abspath(env_dir)
ok, reason, cache = check_build_dir_coverage_config(abs_dir)
return abs_dir, ok, reason, cache
candidates = [
os.path.join(REPO_ROOT, 'out', 'cov'),
os.path.join(REPO_ROOT, 'build', 'cov'),
os.path.join(REPO_ROOT, 'cov'),
REPO_ROOT,
os.path.join(REPO_ROOT, 'out', 'debug'),
os.path.join(REPO_ROOT, 'build', 'debug'),
os.path.join(REPO_ROOT, 'debug'),
os.path.join(REPO_ROOT, 'out'),
os.path.join(REPO_ROOT, 'build'),
]
first_existing = None
for cand in candidates:
ok, reason, cache = check_build_dir_coverage_config(cand)
if ok:
return cand, True, reason, cache
if first_existing is None and os.path.isfile(os.path.join(cand, 'CMakeCache.txt')):
first_existing = (cand, False, reason, cache)
if first_existing:
return first_existing
default_cov_dir = os.path.join(REPO_ROOT, 'out', 'cov')
return default_cov_dir, False, f"No coverage build directory found (checked '{default_cov_dir}' and repo root).", {}
def print_setup_instructions(reason: str, suggested_build_dir: str = 'out/cov'):
rel_build = os.path.relpath(os.path.abspath(suggested_build_dir), REPO_ROOT)
if rel_build == '.':
rel_build = 'out/cov'
print(f'Error: Build is not configured for coverage reporting.\n Reason: {reason}\n', file=sys.stderr)
print(
f"""If you already have a coverage build directory, specify it with:
-B <dir> / --build-dir <dir> (or set BINARYEN_BUILD_DIR=<dir>)
To configure a new Clang coverage build:
1. Configure CMake with Clang source-based coverage flags (e.g. in '{rel_build}'):
cmake -S . -B {rel_build} -G Ninja \\
-DCMAKE_BUILD_TYPE=Debug \\
-DCMAKE_C_COMPILER=clang \\
-DCMAKE_CXX_COMPILER=clang++ \\
-DCMAKE_C_FLAGS="-fprofile-instr-generate -fcoverage-mapping -fcoverage-mcdc" \\
-DCMAKE_CXX_FLAGS="-fprofile-instr-generate -fcoverage-mapping -fcoverage-mcdc"
2. Build the targets exercised by your tests (e.g. wasm-opt, binaryen-unittests):
ninja -C {rel_build} wasm-opt binaryen-unittests
3. Run this tool with your test(s) to collect profiles and report diff coverage in one step:
{SCRIPT_PATH} -B {rel_build} --lit test/lit/passes/<your-test>.wast
{SCRIPT_PATH} -B {rel_build} --gtest "*YourTest*"
{SCRIPT_PATH} -B {rel_build} --run "{rel_build}/bin/wasm-opt ..."
""",
file=sys.stderr,
)
def print_missing_profile_instructions(build_dir: str):
rel_build = os.path.relpath(build_dir, REPO_ROOT)
bin_prefix = f'{rel_build}/bin' if rel_build != '.' else './bin'
b_flag = f'-B {rel_build} ' if rel_build not in {'.', 'out/cov'} else ''
print(
f"Error: No coverage profile data (.profraw or .profdata) found in '{rel_build}'.\n",
file=sys.stderr,
)
print(
f"""Run tests through coverage-diff.py to automatically generate and merge profiles:
# Run specific lit test(s):
{SCRIPT_PATH} {b_flag}--lit test/lit/passes/<your-test>.wast
# Run GTest unit tests (all or filtered):
{SCRIPT_PATH} {b_flag}--gtest "*YourFilter*"
# Run an arbitrary command:
{SCRIPT_PATH} {b_flag}--run "{bin_prefix}/wasm-opt ..."
Or run tests manually with LLVM_PROFILE_FILE set, then re-run coverage-diff.py:
LLVM_PROFILE_FILE="$(pwd)/{rel_build}/coverage/profraw/%8m.profraw" {bin_prefix}/binaryen-lit test/lit/...
{SCRIPT_PATH} {b_flag.rstrip()}
""",
file=sys.stderr,
)
def find_object_file_for_source(build_dir: str, rel_path: str) -> str | None:
"""Locate the compiled .o file for a C/C++ source file in build_dir."""
if rel_path.endswith(('.h', '.inc')):
return None
# 1. Standard libbinaryen object path: CMakeFiles/binaryen.dir/<rel_path>.o
cand = os.path.join(build_dir, 'CMakeFiles', 'binaryen.dir', f'{rel_path}.o')
if os.path.isfile(cand):
return cand
# 2. Subdirectory target object path: <dir>/CMakeFiles/*.dir/<basename>.o
rel_dir = os.path.dirname(rel_path)
base_name = os.path.basename(rel_path)
matches = glob.glob(os.path.join(build_dir, rel_dir, 'CMakeFiles', '*.dir', f'{base_name}.o'))
if not matches:
matches = glob.glob(
os.path.join(build_dir, 'src', 'tools', 'CMakeFiles', '*.dir', '**', f'{base_name}.o'),
recursive=True,
)
return matches[0] if matches else None
def find_coverage_objects(
build_dir: str,
explicit_objects: list[str] | None,
diff_rel_paths: list[str],
used_gtest: bool,
) -> list[str]:
if explicit_objects:
return [os.path.abspath(obj) for obj in explicit_objects]
# Fast path: if all modified files in the diff are .c/.cpp compilation units
# with corresponding .o files in build_dir, pass those .o files directly to
# llvm-cov export (~0.15s) instead of parsing all 300MB+ of libbinaryen.so (~12s).
obj_files: list[str] = []
all_have_obj = bool(diff_rel_paths)
for rel_path in diff_rel_paths:
obj_file = find_object_file_for_source(build_dir, rel_path)
if obj_file:
obj_files.append(obj_file)
else:
all_have_obj = False
break
if all_have_obj and obj_files:
return obj_files
objects: list[str] = []
lib_dir = os.path.join(build_dir, 'lib')
for lib_name in ('libbinaryen.so', 'libbinaryen.dylib'):
lib_path = os.path.join(lib_dir, lib_name)
if os.path.isfile(lib_path):
objects.append(lib_path)
bin_dir = os.path.join(build_dir, 'bin')
needed_bins: set[str] = set()
# If static build (no shared libbinaryen), we need at least wasm-opt / binaryen-unittests.
if not objects:
needed_bins.update(['wasm-opt', 'binaryen-unittests'])
if used_gtest:
needed_bins.add('binaryen-unittests')
for rel_path in diff_rel_paths:
parts = rel_path.replace('\\', '/').split('/')
if len(parts) >= 3 and parts[0] == 'src' and parts[1] == 'tools':
tool_entry = parts[2]
if os.path.splitext(tool_entry)[1]:
needed_bins.add(os.path.splitext(tool_entry)[0])
else:
needed_bins.add(tool_entry)
elif len(parts) >= 2 and parts[0] == 'test' and parts[1] == 'gtest':
needed_bins.add('binaryen-unittests')
if rel_path.endswith(('.h', '.inc')):
# Header changes might only be instantiated inside unit tests or tools.
needed_bins.update(['wasm-opt', 'binaryen-unittests'])
for bin_name in sorted(needed_bins):
bin_path = os.path.join(bin_dir, bin_name)
if os.path.isfile(bin_path) and os.access(bin_path, os.X_OK):
objects.append(bin_path)
# Fallback: if still empty, look for any executable in bin/
if not objects and os.path.isdir(bin_dir):
for entry in sorted(os.listdir(bin_dir)):
if entry == 'binaryen-lit':
continue
full = os.path.join(bin_dir, entry)
if os.path.isfile(full) and os.access(full, os.X_OK):
objects.append(full)
break
return objects
def remove_file_quietly(path: str):
try:
os.remove(path)
except OSError:
pass
def clear_coverage_profiles(build_dir: str) -> int:
"""Remove all .profraw and .profdata files in build_dir and return count removed."""
cov_dir = os.path.join(build_dir, 'coverage')
patterns = [
os.path.join(cov_dir, 'profraw', '*.profraw'),
os.path.join(cov_dir, '*.profraw'),
os.path.join(cov_dir, '*.profdata'),
os.path.join(build_dir, '*.profraw'),
os.path.join(build_dir, 'default.profdata'),
os.path.join(build_dir, 'out', 'test', '**', '*.profraw'),
os.path.join(REPO_ROOT, '*.profraw'),
]
removed = 0
seen: set[str] = set()
for pat in patterns:
for path in glob.glob(pat, recursive=True):
if path not in seen and os.path.isfile(path):
seen.add(path)
try:
os.remove(path)
removed += 1
except OSError:
pass
return removed
def run_test_commands(
build_dir: str,
lit_tests: list[str] | None,
gtest_filter: str | None,
custom_cmds: list[str] | None,
append_profiles: bool,
) -> tuple[str, list[str]]:
"""Execute requested tests with LLVM_PROFILE_FILE set and return profraw dir & list."""
profraw_dir = os.path.join(build_dir, 'coverage', 'profraw')
os.makedirs(profraw_dir, exist_ok=True)
if not append_profiles:
clear_coverage_profiles(build_dir)
env = os.environ.copy()
env['LLVM_PROFILE_FILE'] = os.path.join(profraw_dir, '%8m.profraw')
bin_dir = os.path.join(build_dir, 'bin')
env['PATH'] = f"{bin_dir}{os.pathsep}{env.get('PATH', '')}"
commands: list[list[str] | str] = []
if lit_tests:
lit_script = os.path.join(bin_dir, 'binaryen-lit')
if not os.path.isfile(lit_script):
rel_build = os.path.relpath(build_dir, REPO_ROOT)
print(
f"Error: '{lit_script}' not found. Configure and build '{rel_build}' first.",
file=sys.stderr,
)
sys.exit(2)
commands.append([sys.executable, lit_script, '-vv', *lit_tests])
if gtest_filter is not None:
gtest_bin = os.path.join(bin_dir, 'binaryen-unittests')
if not os.path.isfile(gtest_bin):
rel_build = os.path.relpath(build_dir, REPO_ROOT)
print(
f"Error: '{gtest_bin}' not found. Build it with: ninja -C {rel_build} binaryen-unittests",
file=sys.stderr,
)
sys.exit(2)
cmd = [gtest_bin]
if gtest_filter and gtest_filter != '*':
cmd.append(f'--gtest_filter={gtest_filter}')
commands.append(cmd)
if custom_cmds:
commands.extend(custom_cmds)
for cmd in commands:
if isinstance(cmd, list):
display = shlex.join(cmd)
res = subprocess.run(cmd, cwd=REPO_ROOT, env=env)
else:
display = cmd
res = subprocess.run(cmd, cwd=REPO_ROOT, env=env, shell=True)
if res.returncode != 0:
print(f'Error: Test command failed (exit code {res.returncode}): {display}', file=sys.stderr)
sys.exit(res.returncode)
profraw_files = sorted(glob.glob(os.path.join(profraw_dir, '*.profraw')))
return profraw_dir, profraw_files
def resolve_profdata(
build_dir: str,
llvm_profdata: str,
explicit_profdata: str | None,
profraw_files: list[str] | None,
objects: list[str] | None = None,
) -> str | None:
if explicit_profdata:
return os.path.abspath(explicit_profdata) if os.path.isfile(explicit_profdata) else None
cov_dir = os.path.join(build_dir, 'coverage')
profdata_path = os.path.join(cov_dir, 'coverage.profdata')
newest_obj_mtime = max(
(os.path.getmtime(o) for o in (objects or []) if os.path.isfile(o)),
default=0.0,
)
if profraw_files is None:
# Search standard locations for existing .profraw files.
search_patterns = [
os.path.join(cov_dir, 'profraw', '*.profraw'),
os.path.join(cov_dir, '*.profraw'),
os.path.join(build_dir, '*.profraw'),
os.path.join(build_dir, 'out', 'test', '**', '*.profraw'),
os.path.join(REPO_ROOT, '*.profraw'),
]
found: list[str] = []
for pat in search_patterns:
found.extend(glob.glob(pat, recursive=True))
profraw_files = sorted(set(found))
# Automatically discard any .profraw files recorded before the target objects were rebuilt.
if profraw_files and newest_obj_mtime > 0:
fresh_profraw: list[str] = []
for p in profraw_files:
if os.path.getmtime(p) < newest_obj_mtime:
remove_file_quietly(p)
else:
fresh_profraw.append(p)
if len(fresh_profraw) != len(profraw_files):
remove_file_quietly(profdata_path)
profraw_files = fresh_profraw
if profraw_files:
os.makedirs(cov_dir, exist_ok=True)
need_merge = not os.path.isfile(profdata_path)
if not need_merge:
profdata_mtime = os.path.getmtime(profdata_path)
need_merge = any(os.path.getmtime(p) > profdata_mtime for p in profraw_files)
if need_merge:
cmd = [llvm_profdata, 'merge', '-sparse', *profraw_files, '-o', profdata_path]
res = subprocess.run(cmd, capture_output=True, text=True)
if res.returncode != 0:
print(f'Error merging profile data with {llvm_profdata}:\n{res.stderr}', file=sys.stderr)
sys.exit(2)
return profdata_path
if os.path.isfile(profdata_path):
if newest_obj_mtime > 0 and os.path.getmtime(profdata_path) < newest_obj_mtime:
remove_file_quietly(profdata_path)
else:
return profdata_path
root_profdata = os.path.join(build_dir, 'default.profdata')
if os.path.isfile(root_profdata):
if newest_obj_mtime > 0 and os.path.getmtime(root_profdata) < newest_obj_mtime:
remove_file_quietly(root_profdata)
else:
return root_profdata
return None
def detect_vcs(explicit_vcs: str | None = None) -> str:
if explicit_vcs and explicit_vcs != 'auto':
return explicit_vcs
cur = os.path.abspath(REPO_ROOT)
while True:
if os.path.isdir(os.path.join(cur, '.jj')) and shutil.which('jj'):
return 'jj'
if os.path.exists(os.path.join(cur, '.git')):
return 'git'
parent = os.path.dirname(cur)
if parent == cur:
break
cur = parent
return 'jj' if shutil.which('jj') and not shutil.which('git') else 'git'
def matches_file_patterns(rel_path: str, file_patterns: list[str] | None) -> bool:
if not file_patterns:
return True
rel_norm = rel_path.replace('\\', '/')
base_name = os.path.basename(rel_norm)
target_real = os.path.realpath(os.path.join(REPO_ROOT, rel_norm))
for raw_pat in file_patterns:
pat = raw_pat.replace('\\', '/')
while pat.startswith('./'):
pat = pat[2:]
if pat in {rel_norm, base_name}:
return True
if fnmatch.fnmatch(rel_norm, pat) or fnmatch.fnmatch(base_name, pat):
return True
if not any(ch in raw_pat for ch in '*?[') and os.path.realpath(raw_pat) == target_real:
return True
return False
def parse_unified_diff(
diff_stdout: str,
file_patterns: list[str] | None = None,
) -> dict[str, tuple[list[int], list[tuple[int, int]]]]:
file_diffs: dict[str, tuple[set[int], list[tuple[int, int]]]] = {}
current_file: str | None = None
for raw_line in diff_stdout.splitlines():
if raw_line.startswith('diff --git '):
current_file = None
elif raw_line.startswith('+++ b/'):
rel_path = raw_line[6:].strip()
ext = os.path.splitext(rel_path)[1]
abs_path = os.path.join(REPO_ROOT, rel_path)
if (
ext in CPP_EXTENSIONS
and os.path.isfile(abs_path)
and not rel_path.startswith('third_party/')
and matches_file_patterns(rel_path, file_patterns)
):
current_file = rel_path
if current_file not in file_diffs:
file_diffs[current_file] = (set(), [])
else:
current_file = None
elif raw_line.startswith('+++ /dev/null'):
current_file = None
elif current_file and raw_line.startswith('@@ '):
m = HUNK_RE.match(raw_line)
if m:
start_line = int(m.group(1))
count = int(m.group(2)) if m.group(2) is not None else 1
if count > 0:
lines_set, hunks = file_diffs[current_file]
lines_set.update(range(start_line, start_line + count))
hunks.append((start_line, count))
return {
rel: (sorted(lines_set), hunks)
for rel, (lines_set, hunks) in sorted(file_diffs.items())
if lines_set
}
def get_jj_diff_output(
base_ref: str | None,
revisions: str | None,
staged_only: bool,
) -> tuple[str, str]:
if staged_only:
print(
'Error: --staged is only supported with Git (Jujutsu has no staging area; use -r @ or --base @-).',
file=sys.stderr,
)
sys.exit(2)
if not shutil.which('jj'):
print("Error: 'jj' executable not found in PATH.", file=sys.stderr)
sys.exit(2)
jj_prefix = ['jj', '--no-pager', '--color=never']
if revisions:
diff_cmd = [*jj_prefix, 'diff', '--git', '--context=0', f'--revisions={revisions}']
desc = f'jj revisions {revisions}'
elif base_ref:
diff_cmd = [*jj_prefix, 'diff', '--git', '--context=0', f'--from={base_ref}', '--to=@']
desc = f'changes vs {base_ref}'
else:
mb_res = subprocess.run(
[*jj_prefix, 'log', '--no-graph', '-r', 'latest(heads(::(trunk() ~ root()) & ::@))', '-T', 'commit_id.short(10)'],
cwd=REPO_ROOT,
capture_output=True,
text=True,
)
base_commit = mb_res.stdout.strip().splitlines()[0].strip() if mb_res.returncode == 0 and mb_res.stdout.strip() else ''
if base_commit:
rev_res = subprocess.run(
[*jj_prefix, 'log', '--no-graph', '-r', '(trunk() ~ root())..@', '-T', 'change_id.short(8) ++ "\\n"'],
cwd=REPO_ROOT,
capture_output=True,
text=True,
)
if rev_res.returncode == 0 and rev_res.stdout.strip():
diff_cmd = [*jj_prefix, 'diff', '--git', '--context=0', '--revisions=(trunk() ~ root())..@']
else:
diff_cmd = [*jj_prefix, 'diff', '--git', '--context=0', f'--from={base_commit}', '--to=@']
desc = f'branch changes vs trunk() ({base_commit})'
else:
wc_res = subprocess.run(
[*jj_prefix, 'log', '--no-graph', '-r', '@ & ~empty()', '-T', 'change_id.short(8)'],
cwd=REPO_ROOT,
capture_output=True,
text=True,
)
if wc_res.returncode == 0 and wc_res.stdout.strip():
diff_cmd = [*jj_prefix, 'diff', '--git', '--context=0', '--revisions=@']
desc = 'working-copy changes (@)'
else:
diff_cmd = [*jj_prefix, 'diff', '--git', '--context=0', '--revisions=@-']
desc = 'parent commit changes (@-)'
res = subprocess.run(diff_cmd, cwd=REPO_ROOT, capture_output=True, text=True)
if res.returncode != 0:
print(f'Error running jj diff:\n{res.stderr}', file=sys.stderr)
sys.exit(2)
return desc, res.stdout
def get_git_diff_output(
base_ref: str | None,
revisions: str | None,
staged_only: bool,
) -> tuple[str, str]:
if staged_only:
diff_cmd = ['git', 'diff', '--cached', '-U0', '--no-color']
desc = 'staged changes (git diff --cached)'
elif revisions:
if '..' in revisions:
diff_cmd = ['git', 'diff', '-U0', '--no-color', '--end-of-options', revisions]
else:
diff_cmd = ['git', 'diff', '-U0', '--no-color', '--end-of-options', f'{revisions}~1', revisions]
desc = f'git revisions {revisions}'
elif base_ref:
if base_ref.startswith('-'):
print(f"Error: Invalid base ref '{base_ref}'.", file=sys.stderr)
sys.exit(2)
diff_cmd = ['git', 'diff', '-U0', '--no-color', base_ref]
desc = f'changes vs {base_ref}'
else:
# Check if there are uncommitted C/C++ changes against HEAD first.
status_res = subprocess.run(
['git', 'diff', '--name-only', 'HEAD'],
cwd=REPO_ROOT,
capture_output=True,
text=True,
)
has_uncommitted_cpp = any(
os.path.splitext(p.strip())[1] in CPP_EXTENSIONS
for p in status_res.stdout.splitlines()
if p.strip()
)
if has_uncommitted_cpp:
diff_cmd = ['git', 'diff', '-U0', '--no-color', 'HEAD']
desc = 'uncommitted changes vs HEAD'
else:
# Fall back to merge-base with upstream or origin/main.
upstream = '@{upstream}'
mb_res = subprocess.run(
['git', 'merge-base', upstream, 'HEAD'],
cwd=REPO_ROOT,
capture_output=True,
text=True,
)
if mb_res.returncode != 0:
upstream = 'origin/main'
mb_res = subprocess.run(
['git', 'merge-base', upstream, 'HEAD'],
cwd=REPO_ROOT,
capture_output=True,
text=True,
)
if mb_res.returncode == 0 and mb_res.stdout.strip():
merge_base = mb_res.stdout.strip()
diff_cmd = ['git', 'diff', '-U0', '--no-color', merge_base]
desc = f'branch changes vs {upstream} ({merge_base[:10]})'
else:
diff_cmd = ['git', 'diff', '-U0', '--no-color', 'HEAD~1']
desc = 'changes vs HEAD~1'
res = subprocess.run(diff_cmd, cwd=REPO_ROOT, capture_output=True, text=True)
if res.returncode != 0:
print(f'Error running git diff:\n{res.stderr}', file=sys.stderr)
sys.exit(2)
return desc, res.stdout
def get_diff_lines(
vcs: str,
base_ref: str | None,
revisions: str | None,
staged_only: bool,
file_patterns: list[str] | None = None,
) -> tuple[str, dict[str, tuple[list[int], list[tuple[int, int]]]]]:
"""Return (diff_description, {rel_path: (sorted_lines, [(start, count), ...])})."""
if vcs == 'jj':
desc, stdout = get_jj_diff_output(base_ref, revisions, staged_only)
else:
desc, stdout = get_git_diff_output(base_ref, revisions, staged_only)
if file_patterns:
desc = f"{desc} [files: {', '.join(file_patterns)}]"
return desc, parse_unified_diff(stdout, file_patterns)
def compute_line_stats(raw_segments: list[list]) -> dict[int, LineStat]:
"""Replicate llvm::coverage::LineCoverageIterator & LineCoverageStats."""
if not raw_segments:
return {}
segments = [
Segment(
line=int(s[0]),
col=int(s[1]),
count=int(s[2]),
has_count=bool(s[3]),
is_region_entry=bool(s[4]),
is_gap_region=bool(s[5]),
)
for s in raw_segments
]
max_line = segments[-1].line
seg_idx = 0
num_segs = len(segments)
wrapped_seg: Segment | None = None
line_segs: list[Segment] = []
stats: dict[int, LineStat] = {}
def is_start_of_region(seg: Segment) -> bool:
return not seg.is_gap_region and seg.has_count and seg.is_region_entry
for line in range(1, max_line + 1):
if seg_idx >= num_segs:
break
if line_segs:
wrapped_seg = line_segs[-1]
line_segs = []
while seg_idx < num_segs and segments[seg_idx].line == line:
line_segs.append(segments[seg_idx])
seg_idx += 1
min_region_count = sum(1 for s in line_segs if is_start_of_region(s))
start_of_skipped = bool(line_segs) and not line_segs[0].has_count and line_segs[0].is_region_entry
mapped = not start_of_skipped and (
(wrapped_seg is not None and wrapped_seg.has_count) or (min_region_count > 0)
)
if not mapped:
mapped = any(s.is_region_entry and s.has_count for s in line_segs)
if not mapped:
stats[line] = LineStat(line=line, mapped=False, count=0)
continue
exec_count = wrapped_seg.count if wrapped_seg is not None else 0
if min_region_count > 0:
for s in line_segs:
if is_start_of_region(s):
exec_count = max(exec_count, s.count)
stats[line] = LineStat(line=line, mapped=True, count=exec_count)
return stats
def demangle_names(names: list[str]) -> dict[str, str]:
if not names:
return {}
demangler = find_llvm_tool('llvm-cxxfilt') or shutil.which('c++filt')
if not demangler:
return {n: n for n in names}
# Strip leading source filename prefix if present (e.g. "File.cpp:_Z...")
stripped = [n.split(':', 1)[-1] if ':' in n and n.split(':', 1)[0].endswith(('.cpp', '.c', '.h')) else n for n in names]
res = subprocess.run(
[demangler],
input='\n'.join(stripped),
capture_output=True,
text=True,
)
if res.returncode != 0:
return {n: n for n in names}
out_lines = res.stdout.splitlines()
if len(out_lines) != len(names):
return {n: n for n in names}
return dict(zip(names, [line.strip() for line in out_lines], strict=True))
def find_unreachable_lines(source_lines: list[str]) -> set[int]:
"""Identify 1-indexed lines with WASM_UNREACHABLE or exclusion markers."""
unreachable: set[int] = set()
for idx, text in enumerate(source_lines):
line_no = idx + 1
if UNREACHABLE_RE.search(text):
unreachable.add(line_no)
# Also exclude an immediately preceding standalone `default:` label.
if idx > 0 and DEFAULT_LABEL_RE.match(source_lines[idx - 1]):
unreachable.add(line_no - 1)
# Also exclude an immediately following standalone closing `}`.
if idx + 1 < len(source_lines) and CLOSING_BRACE_RE.match(source_lines[idx + 1]):
unreachable.add(line_no + 1)
return unreachable
def extract_snippet(source_lines: list[str], line_start: int, col_start: int, line_end: int, col_end: int) -> str:
if line_start < 1 or line_start > len(source_lines):
return ''
if line_start == line_end:
line_text = source_lines[line_start - 1]
c_start = max(0, col_start - 1)
c_end = max(c_start, col_end - 1) if col_end > col_start else len(line_text)
snippet = line_text[c_start:c_end].strip()
return snippet or line_text.strip()
first_part = source_lines[line_start - 1][max(0, col_start - 1):].strip()
return f'{first_part} ...'
def group_line_ranges(lines: list[int]) -> list[tuple[int, int]]:
if not lines:
return []
sorted_lines = sorted(lines)
ranges: list[tuple[int, int]] = []
start = prev = sorted_lines[0]
for ln in sorted_lines[1:]:
if ln == prev + 1:
prev = ln
else:
ranges.append((start, prev))
start = prev = ln
ranges.append((start, prev))
return ranges
def populate_regions_and_functions(
file_cov: FileDiffCoverage,
func_entries: list[dict],
demangled_map: dict[str, str],
active_diff_lines: list[int],
active_diff_set: set[int],
unreachable_lines: set[int],
include_system_macros: bool,
):
region_counts: dict[tuple[int, int, int, int], int] = {}
func_spans: dict[tuple[str, int, int], FunctionSpan] = {}
executable_diff_lines = [
ln for ln in active_diff_lines if file_cov.line_stats.get(ln) and file_cov.line_stats[ln].mapped
]
for fn in func_entries:
fn_regions = fn.get('regions', [])
if not fn_regions:
continue
outer = fn_regions[0]
f_start, f_end = int(outer[0]), int(outer[2])
if any(f_start <= ln <= f_end for ln in executable_diff_lines):
raw_name = fn['name']
demangled = demangled_map.get(raw_name, raw_name)
key = (demangled, f_start, f_end)
if key not in func_spans:
func_spans[key] = FunctionSpan(
name=raw_name,
demangled_name=demangled,
line_start=f_start,
line_end=f_end,
count=int(fn.get('count', 0)),
)
else:
func_spans[key].count += int(fn.get('count', 0))
for reg in fn_regions:
r_lstart, r_cstart, r_lend, r_cend = int(reg[0]), int(reg[1]), int(reg[2]), int(reg[3])
r_count, r_file_id, r_kind = int(reg[4]), int(reg[5]), int(reg[7])
if r_kind != REGION_KIND_CODE or r_file_id != 0:
continue
if not include_system_macros and r_lstart == r_lend and r_cstart == r_cend:
continue
if r_lstart in unreachable_lines or (
r_lstart < r_lend
and all(ln in unreachable_lines for ln in range(r_lstart + 1, r_lend + 1))
):
continue
r_key = (r_lstart, r_cstart, r_lend, r_cend)
region_counts[r_key] = region_counts.get(r_key, 0) + r_count
for span in sorted(func_spans.values(), key=lambda s: (s.line_start, s.demangled_name)):
file_cov.functions.add(1 if span.count > 0 else 0, 1)
if span.count == 0:
file_cov.uncovered_functions.append(span)
selected_regions: dict[tuple[int, int, int, int], int] = {}
lines_with_starting_region: set[int] = set()
for r_key, r_count in region_counts.items():
if r_key[0] in active_diff_set:
selected_regions[r_key] = r_count
lines_with_starting_region.add(r_key[0])
for ln in active_diff_lines:
st = file_cov.line_stats.get(ln)
if not (st and st.mapped) or ln in lines_with_starting_region:
continue
enclosing = [
(r_key, r_count)
for r_key, r_count in region_counts.items()
if r_key[0] <= ln <= r_key[2]
]
if enclosing:
innermost_key, innermost_count = min(
enclosing,
key=lambda item: (item[0][2] - item[0][0], item[0][3] - item[0][1]),
)
selected_regions[innermost_key] = innermost_count
uncovered_line_set = set(file_cov.uncovered_lines)
for r_key in sorted(selected_regions):
r_count = selected_regions[r_key]
file_cov.regions.add(1 if r_count > 0 else 0, 1)
r_lstart, r_cstart, r_lend, r_cend = r_key
if r_count == 0 and r_lstart in active_diff_set and r_lstart not in uncovered_line_set:
# Skip multi-line block regions (e.g. `{` at the end of `if (...) {`)
# whose interior lines are already reported in uncovered_lines.
if r_lstart < r_lend and any(
ln in uncovered_line_set for ln in range(r_lstart + 1, r_lend + 1)
):
continue
file_cov.uncovered_subregions.append(
CodeRegion(r_lstart, r_cstart, r_lend, r_cend, r_count),
)
def format_span_loc(line_start: int, col_start: int, line_end: int, col_end: int) -> str:
if line_start == line_end:
return f'Line {line_start}:{col_start}-{col_end}'
return f'Lines {line_start}:{col_start}-{line_end}:{col_end}'
def populate_branches_and_mcdc(
file_cov: FileDiffCoverage,
file_entry: dict,
active_diff_set: set[int],
unreachable_lines: set[int],
include_system_macros: bool,
):
branch_map: dict[tuple[int, int, int, int], list[int]] = {}
for br in file_entry.get('branches', []):
b_lstart, b_cstart, b_lend, b_cend = int(br[0]), int(br[1]), int(br[2]), int(br[3])
b_true, b_false, b_file_id = int(br[4]), int(br[5]), int(br[6])
if b_file_id != 0:
continue
if not include_system_macros and b_lstart == b_lend and b_cstart == b_cend:
continue
if b_lstart in unreachable_lines:
continue
if not any(ln in active_diff_set for ln in range(b_lstart, b_lend + 1)):
continue
b_key = (b_lstart, b_cstart, b_lend, b_cend)
if b_key not in branch_map:
branch_map[b_key] = [b_true, b_false]
else:
branch_map[b_key][0] += b_true
branch_map[b_key][1] += b_false
for b_key in sorted(branch_map):
b_true, b_false = branch_map[b_key]
cov_outcomes = (1 if b_true > 0 else 0) + (1 if b_false > 0 else 0)
file_cov.branches.add(cov_outcomes, 2)
if cov_outcomes < 2:
file_cov.uncovered_branches.append(
BranchRecord(b_key[0], b_key[1], b_key[2], b_key[3], b_true, b_false),
)
for mcdc in file_entry.get('mcdc_records', []):
m_lstart, m_cstart, m_lend, m_cend = int(mcdc[0]), int(mcdc[1]), int(mcdc[2]), int(mcdc[3])
m_file_id = int(mcdc[6])
conditions = [bool(c) for c in mcdc[9]] if len(mcdc) > 9 and isinstance(mcdc[9], list) else []
if m_file_id != 0 or not conditions or m_lstart in unreachable_lines:
continue
if not any(ln in active_diff_set for ln in range(m_lstart, m_lend + 1)):
continue
cov_conds = sum(1 for c in conditions if c)
file_cov.mcdc.add(cov_conds, len(conditions))
if cov_conds < len(conditions):
file_cov.uncovered_mcdc.append(
MCDCRecord(m_lstart, m_cstart, m_lend, m_cend, conditions),
)
def analyze_coverage(
cov_json: dict,
file_diffs: dict[str, tuple[list[int], list[tuple[int, int]]]],
include_unreachable: bool,
include_system_macros: bool,
) -> list[FileDiffCoverage]:
"""Intersect llvm-cov export JSON with diff line sets across all metrics."""
files_by_realpath: dict[str, dict] = {}
functions_by_realpath: dict[str, list[dict]] = {}
for datum in cov_json.get('data', []):
for file_entry in datum.get('files', []):
rp = os.path.realpath(file_entry['filename'])
files_by_realpath[rp] = file_entry
for func_entry in datum.get('functions', []):
filenames = func_entry.get('filenames', [])
if filenames:
rp = os.path.realpath(filenames[0])
functions_by_realpath.setdefault(rp, []).append(func_entry)
all_func_names = [fn['name'] for funcs in functions_by_realpath.values() for fn in funcs]
demangled_map = demangle_names(list(set(all_func_names)))
results: list[FileDiffCoverage] = []
for rel_path, (diff_lines, hunks) in file_diffs.items():
abs_path = os.path.realpath(os.path.join(REPO_ROOT, rel_path))
try:
with open(abs_path, errors='replace') as f:
source_lines = [line.rstrip('\r\n') for line in f]
except OSError:
source_lines = []
file_cov = FileDiffCoverage(
rel_path=rel_path,
abs_path=abs_path,
diff_lines=diff_lines,
hunks=hunks,
)
file_entry = files_by_realpath.get(abs_path)
if not file_entry:
file_cov.in_coverage_map = False
results.append(file_cov)
continue
unreachable_lines = set() if include_unreachable else find_unreachable_lines(source_lines)
diff_line_set = set(diff_lines)
active_diff_lines = sorted(diff_line_set - unreachable_lines)
file_cov.excluded_unreachable_lines = len(diff_line_set & unreachable_lines)
active_diff_set = set(active_diff_lines)
line_stats = compute_line_stats(file_entry.get('segments', []))
file_cov.line_stats = line_stats
for ln in active_diff_lines:
st = line_stats.get(ln)
if st and st.mapped:
file_cov.lines.add(1 if st.count > 0 else 0, 1)
if st.count == 0:
file_cov.uncovered_lines.append(ln)
else:
file_cov.non_executable_diff_lines += 1
populate_regions_and_functions(
file_cov=file_cov,
func_entries=functions_by_realpath.get(abs_path, []),
demangled_map=demangled_map,
active_diff_lines=active_diff_lines,
active_diff_set=active_diff_set,
unreachable_lines=unreachable_lines,
include_system_macros=include_system_macros,
)
populate_branches_and_mcdc(
file_cov=file_cov,
file_entry=file_entry,
active_diff_set=active_diff_set,
unreachable_lines=unreachable_lines,
include_system_macros=include_system_macros,
)
results.append(file_cov)
return results
def render_text_report(
diff_desc: str,
file_results: list[FileDiffCoverage],
annotate: bool,
context_lines: int,
):
totals = {
'lines': MetricCounts(),
'regions': MetricCounts(),
'branches': MetricCounts(),
'mcdc': MetricCounts(),
'functions': MetricCounts(),
}
total_excluded_unreachable = 0
for fc in file_results:
if not fc.in_coverage_map:
continue
totals['lines'].add(fc.lines.covered, fc.lines.total)
totals['regions'].add(fc.regions.covered, fc.regions.total)
totals['branches'].add(fc.branches.covered, fc.branches.total)
totals['mcdc'].add(fc.mcdc.covered, fc.mcdc.total)
totals['functions'].add(fc.functions.covered, fc.functions.total)
total_excluded_unreachable += fc.excluded_unreachable_lines
print(f'=== Diff Coverage Report ({diff_desc}) ===\n')
print(f" {'Metric':<12} {'Covered / Total':<22} {'Details'}")
print(f" {'-' * 12} {'-' * 22} {'-' * 34}")
print(f" {'Lines':<12} {totals['lines'].format_summary():<22} Executable lines in diff")
print(f" {'Regions':<12} {totals['regions'].format_summary():<22} Sub-line / block AST regions in diff")
print(f" {'Branches':<12} {totals['branches'].format_summary():<22} True/False branch outcomes in diff")
if totals['mcdc'].total > 0:
print(f" {'MC/DC':<12} {totals['mcdc'].format_summary():<22} Condition independence pairs in diff")
print(f" {'Functions':<12} {totals['functions'].format_summary():<22} Modified functions called >= 1x")
if total_excluded_unreachable > 0:
print(f'\n (Note: Excluded {total_excluded_unreachable} WASM_UNREACHABLE line(s); pass --include-unreachable to include)')
if len(file_results) > 1:
print('\n--- Per-File Summary ---')
for fc in file_results:
if not fc.in_coverage_map:
print(f' {fc.rel_path}: (not in coverage mapping - header/uncompiled)')
continue
if fc.lines.total == 0:
print(f' {fc.rel_path}: (only comments / non-executable lines changed in diff)')
continue
parts = [
f'Lines: {fc.lines.format_summary()}',
f'Regions: {fc.regions.format_summary()}',
f'Branches: {fc.branches.format_summary()}',
]
if fc.mcdc.total > 0:
parts.append(f'MC/DC: {fc.mcdc.format_summary()}')
print(f" {fc.rel_path}\n {', '.join(parts)}")
# Print actionable gaps for each file
any_gaps = False
for fc in file_results:
if not fc.in_coverage_map:
continue
has_file_gaps = bool(
fc.uncovered_functions
or fc.uncovered_lines
or fc.uncovered_subregions
or fc.uncovered_branches
or fc.uncovered_mcdc,
)
if not has_file_gaps:
continue
if not any_gaps:
print('\n=== Uncovered / Partially Covered Code in Diff ===')
any_gaps = True
try:
with open(fc.abs_path, errors='replace') as f:
source_lines = [line.rstrip('\r\n') for line in f]
except OSError:
source_lines = []
print(f'\nFile: {fc.rel_path}')
if fc.uncovered_functions:
print(' Uncovered Functions (0 calls):')
for fn in fc.uncovered_functions:
print(f' - Line {fn.line_start}-{fn.line_end}: {fn.demangled_name}')
if fc.uncovered_lines:
print(' Uncovered Executable Lines (0x):')
for start, end in group_line_ranges(fc.uncovered_lines):
loc = f'Line {start}' if start == end else f'Lines {start}-{end}'
print(f' - {loc}:')
for ln in range(start, end + 1):
src = source_lines[ln - 1] if 1 <= ln <= len(source_lines) else ''
print(f' {ln:5d} | {src}')
if fc.uncovered_subregions:
print(' Partially Covered Lines (Uncovered Sub-line Regions on Executed Lines):')
for reg in fc.uncovered_subregions:
snippet = extract_snippet(
source_lines, reg.line_start, reg.col_start, reg.line_end, reg.col_end,
)
loc = format_span_loc(reg.line_start, reg.col_start, reg.line_end, reg.col_end)
full_line = source_lines[reg.line_start - 1].strip() if 1 <= reg.line_start <= len(source_lines) else ''
context_suffix = f' (in `{full_line}`)' if reg.line_start == reg.line_end and full_line != snippet else ''
print(
f' - {loc} (0x): `{snippet}`{context_suffix}',
)
if fc.uncovered_branches:
print(' Uncovered / One-Sided Branches:')
for br in fc.uncovered_branches:
snippet = extract_snippet(
source_lines, br.line_start, br.col_start, br.line_end, br.col_end,
)
loc = format_span_loc(br.line_start, br.col_start, br.line_end, br.col_end)
if br.true_count == 0 and br.false_count == 0:
status = 'never reached (True: 0x, False: 0x)'
elif br.true_count == 0:
status = f'missing TRUE branch (True: 0x, False: {br.false_count}x)'
else:
status = f'missing FALSE branch (True: {br.true_count}x, False: 0x)'
print(
f' - {loc} `{snippet}` -> {status}',
)
if fc.uncovered_mcdc:
print(' Uncovered MC/DC Condition Independence Pairs:')
for rec in fc.uncovered_mcdc:
snippet = extract_snippet(
source_lines, rec.line_start, rec.col_start, rec.line_end, rec.col_end,
)
loc = format_span_loc(rec.line_start, rec.col_start, rec.line_end, rec.col_end)
missing_conds = [f'C{i + 1}' for i, cov in enumerate(rec.conditions) if not cov]
print(
f" - {loc} `{snippet}` -> "
f"missing independence pair for {', '.join(missing_conds)}",
)
if not any_gaps and totals['lines'].total > 0:
print('\nAll executable lines, regions, and branches in the diff are 100% covered!')
if annotate:
print('\n=== Annotated Diff Hunks ===')
for fc in file_results:
if not fc.in_coverage_map:
continue
try:
with open(fc.abs_path, errors='replace') as f:
source_lines = [line.rstrip('\r\n') for line in f]
except OSError:
continue
print(f'\n--- {fc.rel_path} ---')
diff_set = set(fc.diff_lines)
branches_by_line: dict[int, list[BranchRecord]] = {}
for br in fc.uncovered_branches:
branches_by_line.setdefault(br.line_start, []).append(br)
for start, count in fc.hunks:
lo = max(1, start - context_lines)
hi = min(len(source_lines), start + count - 1 + context_lines)
print(f'@@ lines {lo}-{hi} @@')
for ln in range(lo, hi + 1):
marker = '+' if ln in diff_set else ' '
st = fc.line_stats.get(ln)
if st and st.mapped:
cnt_str = f'{st.count:6d}x' if st.count > 0 else ' #####'
else:
cnt_str = ' '
print(f'{marker} {ln:5d} | {cnt_str} | {source_lines[ln - 1]}')
for br in branches_by_line.get(ln, []):
print(
f' | | ^-- Branch ({br.line_start}:{br.col_start}): '
f'[True: {br.true_count}x, False: {br.false_count}x]',
)
def render_markdown_report(
diff_desc: str,
file_results: list[FileDiffCoverage],
annotate: bool,
context_lines: int,
) -> str:
lines: list[str] = [f'## Diff Coverage Report ({diff_desc})', '']
if not file_results:
lines.append(f'No modified C/C++ source files found in {diff_desc}.')
return '\n'.join(lines)
totals = {
'lines': MetricCounts(),
'regions': MetricCounts(),
'branches': MetricCounts(),
'mcdc': MetricCounts(),
'functions': MetricCounts(),
}
total_excluded_unreachable = 0
for fc in file_results:
if not fc.in_coverage_map:
continue
totals['lines'].add(fc.lines.covered, fc.lines.total)
totals['regions'].add(fc.regions.covered, fc.regions.total)
totals['branches'].add(fc.branches.covered, fc.branches.total)
totals['mcdc'].add(fc.mcdc.covered, fc.mcdc.total)
totals['functions'].add(fc.functions.covered, fc.functions.total)
total_excluded_unreachable += fc.excluded_unreachable_lines
lines.extend([
'| Metric | Covered / Total | Details |',
'| :--- | :--- | :--- |',
f"| **Lines** | {totals['lines'].format_summary()} | Executable lines in diff |",
f"| **Regions** | {totals['regions'].format_summary()} | Sub-line / block AST regions in diff |",
f"| **Branches** | {totals['branches'].format_summary()} | True/False branch outcomes in diff |",
])
if totals['mcdc'].total > 0:
lines.append(
f"| **MC/DC** | {totals['mcdc'].format_summary()} | Condition independence pairs in diff |",
)
lines.append(
f"| **Functions** | {totals['functions'].format_summary()} | Modified functions called >= 1x |",
)
if total_excluded_unreachable > 0:
lines.extend([
'',
f'*(Note: Excluded {total_excluded_unreachable} `WASM_UNREACHABLE` line(s); '
'pass `--include-unreachable` to include)*',
])
if len(file_results) > 1:
has_mcdc = any(fc.in_coverage_map and fc.mcdc.total > 0 for fc in file_results)
lines.extend(['', '### Per-File Summary', ''])
if has_mcdc:
lines.extend([
'| File | Lines | Regions | Branches | MC/DC |',
'| :--- | :--- | :--- | :--- | :--- |',
])
else:
lines.extend([
'| File | Lines | Regions | Branches |',
'| :--- | :--- | :--- | :--- |',
])
for fc in file_results:
if not fc.in_coverage_map:
cols = ['*(not in coverage mapping - header/uncompiled)*', '-', '-']
if has_mcdc:
cols.append('-')
elif fc.lines.total == 0:
cols = ['*(only comments / non-executable lines changed)*', '-', '-']
if has_mcdc:
cols.append('-')
else:
cols = [
fc.lines.format_summary(),
fc.regions.format_summary(),
fc.branches.format_summary(),
]
if has_mcdc:
cols.append(fc.mcdc.format_summary() if fc.mcdc.total > 0 else '-')
lines.append(f"| `{fc.rel_path}` | {' | '.join(cols)} |")
any_gaps = False
for fc in file_results:
if not fc.in_coverage_map:
continue
has_file_gaps = bool(
fc.uncovered_functions
or fc.uncovered_lines
or fc.uncovered_subregions
or fc.uncovered_branches
or fc.uncovered_mcdc,
)
if not has_file_gaps:
continue
if not any_gaps:
lines.extend(['', '### Uncovered / Partially Covered Code in Diff'])
any_gaps = True
try:
with open(fc.abs_path, errors='replace') as f:
source_lines = [line.rstrip('\r\n') for line in f]
except OSError:
source_lines = []
lines.extend(['', f'#### `{fc.rel_path}`'])
if fc.uncovered_functions:
lines.append('- **Uncovered Functions (0 calls):**')
for fn in fc.uncovered_functions:
lines.append(f' - Line {fn.line_start}-{fn.line_end}: `{fn.demangled_name}`')
if fc.uncovered_lines:
lines.append('- **Uncovered Executable Lines (0x):**')
for start, end in group_line_ranges(fc.uncovered_lines):
loc = f'Line {start}' if start == end else f'Lines {start}-{end}'
lines.append(f' - {loc}:')
lines.append(' ```cpp')
for ln in range(start, end + 1):
src = source_lines[ln - 1] if 1 <= ln <= len(source_lines) else ''
lines.append(f' {ln:5d} | {src}')
lines.append(' ```')
if fc.uncovered_subregions:
lines.append('- **Partially Covered Lines (Uncovered Sub-line Regions on Executed Lines):**')
for reg in fc.uncovered_subregions:
snippet = extract_snippet(
source_lines, reg.line_start, reg.col_start, reg.line_end, reg.col_end,
)
loc = format_span_loc(reg.line_start, reg.col_start, reg.line_end, reg.col_end)
full_line = source_lines[reg.line_start - 1].strip() if 1 <= reg.line_start <= len(source_lines) else ''
context_suffix = f' (in `{full_line}`)' if reg.line_start == reg.line_end and full_line != snippet else ''
lines.append(f' - {loc} (0x): `{snippet}`{context_suffix}')
if fc.uncovered_branches:
lines.append('- **Uncovered / One-Sided Branches:**')
for br in fc.uncovered_branches:
snippet = extract_snippet(
source_lines, br.line_start, br.col_start, br.line_end, br.col_end,
)
loc = format_span_loc(br.line_start, br.col_start, br.line_end, br.col_end)
if br.true_count == 0 and br.false_count == 0:
status = 'never reached (True: 0x, False: 0x)'
elif br.true_count == 0:
status = f'missing TRUE branch (True: 0x, False: {br.false_count}x)'
else:
status = f'missing FALSE branch (True: {br.true_count}x, False: 0x)'
lines.append(f' - {loc} `{snippet}` -> {status}')
if fc.uncovered_mcdc:
lines.append('- **Uncovered MC/DC Condition Independence Pairs:**')
for rec in fc.uncovered_mcdc:
snippet = extract_snippet(
source_lines, rec.line_start, rec.col_start, rec.line_end, rec.col_end,
)
loc = format_span_loc(rec.line_start, rec.col_start, rec.line_end, rec.col_end)
missing_conds = [f'C{i + 1}' for i, cov in enumerate(rec.conditions) if not cov]
lines.append(
f" - {loc} `{snippet}` -> "
f"missing independence pair for {', '.join(missing_conds)}",
)
if not any_gaps and totals['lines'].total > 0:
lines.extend([
'',
'All executable lines, regions, and branches in the diff are 100% covered!',
])
if annotate and any(fc.in_coverage_map for fc in file_results):
lines.extend([
'',
'<details>',
'<summary>Annotated Diff Hunks</summary>',
'',
])
for fc in file_results:
if not fc.in_coverage_map:
continue
try:
with open(fc.abs_path, errors='replace') as f:
source_lines = [line.rstrip('\r\n') for line in f]
except OSError:
continue
lines.extend([f'#### `{fc.rel_path}`', '', '```text'])
diff_set = set(fc.diff_lines)
branches_by_line: dict[int, list[BranchRecord]] = {}
for br in fc.uncovered_branches:
branches_by_line.setdefault(br.line_start, []).append(br)
for start, count in fc.hunks:
lo = max(1, start - context_lines)
hi = min(len(source_lines), start + count - 1 + context_lines)
lines.append(f'@@ lines {lo}-{hi} @@')
for ln in range(lo, hi + 1):
marker = '+' if ln in diff_set else ' '
st = fc.line_stats.get(ln)
if st and st.mapped:
cnt_str = f'{st.count:6d}x' if st.count > 0 else ' #####'
else:
cnt_str = ' '
lines.append(f'{marker} {ln:5d} | {cnt_str} | {source_lines[ln - 1]}')
for br in branches_by_line.get(ln, []):
lines.append(
f' | | ^-- Branch ({br.line_start}:{br.col_start}): '
f'[True: {br.true_count}x, False: {br.false_count}x]',
)
lines.extend(['```', ''])
lines.append('</details>')
return '\n'.join(lines)
def escape_github_annotation(msg: str) -> str:
return msg.replace('%', '%25').replace('\r', '%0D').replace('\n', '%0A')
def emit_github_annotations(file_results: list[FileDiffCoverage]):
for fc in file_results:
if not fc.in_coverage_map:
continue
try:
with open(fc.abs_path, errors='replace') as f:
source_lines = [line.rstrip('\r\n') for line in f]
except OSError:
source_lines = []
uncovered_line_set = set(fc.uncovered_lines)
for start, end in group_line_ranges(fc.uncovered_lines):
loc = f'Line {start}' if start == end else f'Lines {start}-{end}'
msg = escape_github_annotation(f'{loc} not executed by tests (0x)')
print(
f'::warning file={fc.rel_path},line={start},endLine={end},'
f'title=Uncovered diff lines::{msg}',
)
for reg in fc.uncovered_subregions:
snippet = extract_snippet(
source_lines, reg.line_start, reg.col_start, reg.line_end, reg.col_end,
)
msg = escape_github_annotation(
f'Sub-region `{snippet}` on executed line was never reached (0x)',
)
print(
f'::warning file={fc.rel_path},line={reg.line_start},endLine={reg.line_end},'
f'col={reg.col_start},endColumn={reg.col_end},'
f'title=Uncovered sub-region::{msg}',
)
for br in fc.uncovered_branches:
if br.line_start in uncovered_line_set:
continue
snippet = extract_snippet(
source_lines, br.line_start, br.col_start, br.line_end, br.col_end,
)
if br.true_count == 0 and br.false_count == 0:
status = 'never reached (True: 0x, False: 0x)'
elif br.true_count == 0:
status = f'missing TRUE branch (True: 0x, False: {br.false_count}x)'
else:
status = f'missing FALSE branch (True: {br.true_count}x, False: 0x)'
msg = escape_github_annotation(f'Branch `{snippet}` -> {status}')
print(
f'::warning file={fc.rel_path},line={br.line_start},endLine={br.line_end},'
f'col={br.col_start},endColumn={br.col_end},'
f'title=One-sided branch in diff::{msg}',
)
for rec in fc.uncovered_mcdc:
if rec.line_start in uncovered_line_set:
continue
snippet = extract_snippet(
source_lines, rec.line_start, rec.col_start, rec.line_end, rec.col_end,
)
missing_conds = [f'C{i + 1}' for i, cov in enumerate(rec.conditions) if not cov]
msg = escape_github_annotation(
f"Decision `{snippet}` -> missing MC/DC independence pair for {', '.join(missing_conds)}",
)
print(
f'::warning file={fc.rel_path},line={rec.line_start},endLine={rec.line_end},'
f'col={rec.col_start},endColumn={rec.col_end},'
f'title=Uncovered MC/DC condition::{msg}',
)
def render_json_report(diff_desc: str, file_results: list[FileDiffCoverage]) -> str:
totals = {
'lines': MetricCounts(),
'regions': MetricCounts(),
'branches': MetricCounts(),
'mcdc': MetricCounts(),
'functions': MetricCounts(),
}
for fc in file_results:
if not fc.in_coverage_map:
continue
totals['lines'].add(fc.lines.covered, fc.lines.total)
totals['regions'].add(fc.regions.covered, fc.regions.total)
totals['branches'].add(fc.branches.covered, fc.branches.total)
totals['mcdc'].add(fc.mcdc.covered, fc.mcdc.total)
totals['functions'].add(fc.functions.covered, fc.functions.total)
payload = {
'diff_target': diff_desc,
'totals': {
k: {'covered': v.covered, 'total': v.total, 'percent': v.percent}
for k, v in totals.items()
},
'files': [
{
'file': fc.rel_path,
'in_coverage_map': fc.in_coverage_map,
'metrics': {
'lines': {'covered': fc.lines.covered, 'total': fc.lines.total, 'percent': fc.lines.percent},
'regions': {'covered': fc.regions.covered, 'total': fc.regions.total, 'percent': fc.regions.percent},
'branches': {'covered': fc.branches.covered, 'total': fc.branches.total, 'percent': fc.branches.percent},
'mcdc': {'covered': fc.mcdc.covered, 'total': fc.mcdc.total, 'percent': fc.mcdc.percent},
'functions': {'covered': fc.functions.covered, 'total': fc.functions.total, 'percent': fc.functions.percent},
},
'uncovered_functions': [
{'name': fn.demangled_name, 'line_start': fn.line_start, 'line_end': fn.line_end}
for fn in fc.uncovered_functions
],
'uncovered_lines': fc.uncovered_lines,
'uncovered_subregions': [
{
'line_start': r.line_start,
'col_start': r.col_start,
'line_end': r.line_end,
'col_end': r.col_end,
}
for r in fc.uncovered_subregions
],
'uncovered_branches': [
{
'line_start': b.line_start,
'col_start': b.col_start,
'line_end': b.line_end,
'col_end': b.col_end,
'true_count': b.true_count,
'false_count': b.false_count,
}
for b in fc.uncovered_branches
],
}
for fc in file_results
],
}
return json.dumps(payload, indent=2)
def main():
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
'base',
nargs='?',
default=None,
help='Git ref or Jujutsu revset to diff against (default: branch changes vs trunk/upstream)',
)
parser.add_argument(
'--base',
dest='base_flag',
default=None,
help='Git ref or Jujutsu revset to diff against (overrides positional base)',
)
parser.add_argument(
'-r',
'--revisions',
default=None,
metavar='REVSET',
help='Revision(s) to analyze (e.g. Jujutsu `-r @-`, `-r "trunk()..@"`, or Git commit/range)',
)
parser.add_argument(
'--vcs',
choices=['auto', 'git', 'jj'],
default='auto',
help='Version control system to query for diff hunks (default: auto)',
)
parser.add_argument(
'--files',
'--file',
nargs='+',
default=None,
metavar='PATTERN',
help='Restrict report to modified C/C++ files matching repo-relative path, basename, or glob',
)
parser.add_argument(
'--staged',
action='store_true',
help='Analyze only staged changes (git diff --cached)',
)
parser.add_argument(
'-B',
'--build-dir',
default=None,
help='CMake build directory configured with coverage flags (default: auto-detect out/cov, build/cov, .)',
)
parser.add_argument(
'--lit',
nargs='+',
metavar='TEST',
help='Run binaryen-lit on the specified test file(s)/directories to collect coverage',
)
parser.add_argument(
'--gtest',
nargs='?',
const='*',
default=None,
metavar='FILTER',
help='Run binaryen-unittests (optionally with --gtest_filter=FILTER) to collect coverage',
)
parser.add_argument(
'--run',
action='append',
metavar='CMD',
help='Run custom command(s) with LLVM_PROFILE_FILE set to collect coverage',
)
parser.add_argument(
'--append',
action='store_true',
help='Keep existing .profraw files when running tests via --lit/--gtest/--run',
)
parser.add_argument(
'--clear',
action='store_true',
help='Clear all collected .profraw and .profdata files in the build directory',
)
parser.add_argument(
'--profdata',
default=None,
help='Explicit path to a merged .profdata file',
)
parser.add_argument(
'--object',
action='append',
dest='objects',
metavar='BIN',
help='Explicit instrumented library/executable path(s) for llvm-cov (auto-detected by default)',
)
parser.add_argument(
'-a',
'--annotate',
action='store_true',
help='Print annotated diff hunks with per-line execution counts and branch counts',
)
parser.add_argument(
'-C',
'--context',
type=int,
default=2,
help='Number of context lines around annotated diff hunks (default: 2)',
)
parser.add_argument(
'--json',
action='store_true',
help='Emit coverage report as JSON',
)
parser.add_argument(
'--markdown',
action='store_true',
help='Emit coverage report as GitHub-flavored Markdown (e.g. for $GITHUB_STEP_SUMMARY)',
)
parser.add_argument(
'--github-annotations',
action='store_true',
help='Emit GitHub Actions workflow warning annotations for uncovered diff code',
)
parser.add_argument(
'--include-unreachable',
action='store_true',
help='Include WASM_UNREACHABLE lines in coverage totals',
)
parser.add_argument(
'--include-system-macros',
action='store_true',
help='Include 0-width system macro expansions (such as assert() internal branches)',
)
parser.add_argument(
'--fail-under-lines',
type=float,
default=None,
metavar='PCT',
help='Exit with code 1 if diff line coverage percentage is below PCT',
)
parser.add_argument(
'--fail-under-regions',
type=float,
default=None,
metavar='PCT',
help='Exit with code 1 if diff region coverage percentage is below PCT',
)
parser.add_argument(
'--fail-under-branches',
type=float,
default=None,
metavar='PCT',
help='Exit with code 1 if diff branch coverage percentage is below PCT',
)
parser.add_argument(
'--llvm-cov',
default=None,
help='Path to llvm-cov binary',
)
parser.add_argument(
'--llvm-profdata',
default=None,
help='Path to llvm-profdata binary',
)
argv = sys.argv[2:] if len(sys.argv) > 2 and sys.argv[1] == '--' and sys.argv[2].startswith('-') else sys.argv[1:]
args = parser.parse_args(argv)
if args.base in {'-h', '--help'}:
parser.print_help()
sys.exit(0)
base_ref = args.base_flag or args.base
if args.revisions and base_ref:
print('Error: Cannot combine -r/--revisions with a base ref.', file=sys.stderr)
sys.exit(2)
if args.revisions and args.staged:
print('Error: Cannot combine -r/--revisions with --staged.', file=sys.stderr)
sys.exit(2)
vcs = detect_vcs(args.vcs)
# 1. Verify llvm-cov and llvm-profdata are installed.
llvm_cov = find_llvm_tool('llvm-cov', args.llvm_cov)
llvm_profdata = find_llvm_tool('llvm-profdata', args.llvm_profdata)
if not llvm_cov or not llvm_profdata:
missing = [name for name, path in [('llvm-cov', llvm_cov), ('llvm-profdata', llvm_profdata)] if not path]
print(
f"Error: Required LLVM coverage tool(s) not found in PATH: {', '.join(missing)}\n"
'Install the llvm package (e.g. `sudo apt-get install llvm`) or pass --llvm-cov / --llvm-profdata.',
file=sys.stderr,
)
sys.exit(2)
# 2. Discover & validate the CMake build directory.
build_dir, is_valid_config, reason, _cache = discover_build_dir(args.build_dir)
if not is_valid_config:
print_setup_instructions(reason, args.build_dir or 'out/cov')
sys.exit(2)
if args.clear:
removed = clear_coverage_profiles(build_dir)
rel_build = os.path.relpath(build_dir, REPO_ROOT)
print(f"Cleared {removed} coverage profile file(s) from '{rel_build}'.")
if not (args.lit or args.gtest is not None or args.run):
sys.exit(0)
# 3. Collect diff hunks first so we know which files/binaries are involved.
diff_desc, file_diffs = get_diff_lines(
vcs=vcs,
base_ref=base_ref,
revisions=args.revisions,
staged_only=args.staged,
file_patterns=args.files,
)
if not file_diffs:
if args.json:
print(render_json_report(diff_desc, []))
elif args.markdown:
print(render_markdown_report(diff_desc, [], args.annotate, args.context))
else:
tip = (
f'(Tip: pass `-r @-` or a base revset such as `{SCRIPT_PATH} main` to compare other revisions.)'
if vcs == 'jj'
else f'(Tip: pass a base ref such as `{SCRIPT_PATH} origin/main` or `HEAD~1` to compare committed changes.)'
)
print(f'No modified C/C++ source files found in {diff_desc}.\n{tip}')
sys.exit(0)
# 4. Run tests if --lit, --gtest, or --run was specified.
profraw_files: list[str] | None = None
if args.lit or args.gtest is not None or args.run:
_profraw_dir, profraw_files = run_test_commands(
build_dir=build_dir,
lit_tests=args.lit,
gtest_filter=args.gtest,
custom_cmds=args.run,
append_profiles=args.append,
)
# 5. Locate instrumented binary/library objects.
objects = find_coverage_objects(
build_dir=build_dir,
explicit_objects=args.objects,
diff_rel_paths=list(file_diffs.keys()),
used_gtest=args.gtest is not None,
)
if not objects:
rel_build = os.path.relpath(build_dir, REPO_ROOT)
print(
f"Error: No built libraries or executables found in '{rel_build}'.\n"
f'Build your targets first, e.g.:\n ninja -C {rel_build} wasm-opt binaryen-unittests',
file=sys.stderr,
)
sys.exit(2)
# 6. Locate or merge .profdata (invalidating any profiles older than rebuilt objects).
profdata_path = resolve_profdata(
build_dir=build_dir,
llvm_profdata=llvm_profdata,
explicit_profdata=args.profdata,
profraw_files=profraw_files,
objects=objects,
)
if not profdata_path:
print_missing_profile_instructions(build_dir)
sys.exit(2)
# 7. Run `llvm-cov export` restricted to the modified files in the diff.
source_paths = [os.path.join(REPO_ROOT, rel) for rel in file_diffs]
cov_cmd = [
llvm_cov,
'export',
'--format=text',
'--skip-expansions',
f'--instr-profile={profdata_path}',
objects[0],
*[f'--object={obj}' for obj in objects[1:]],
'--',
*source_paths,
]
cov_res = subprocess.run(cov_cmd, cwd=REPO_ROOT, capture_output=True, text=True)
if cov_res.returncode != 0:
rel_build = os.path.relpath(build_dir, REPO_ROOT)
print(
f'Error running llvm-cov export:\n{cov_res.stderr.strip()}\n',
file=sys.stderr,
)
if 'No coverage data found' in cov_res.stderr or 'Failed to load coverage' in cov_res.stderr:
print_setup_instructions(
f"Binaries in '{rel_build}' do not contain LLVM coverage mapping sections.",
rel_build,
)
sys.exit(2)
cov_json = json.loads(cov_res.stdout)
file_results = analyze_coverage(
cov_json=cov_json,
file_diffs=file_diffs,
include_unreachable=args.include_unreachable,
include_system_macros=args.include_system_macros,
)
if args.json:
print(render_json_report(diff_desc, file_results))
elif args.markdown:
print(
render_markdown_report(
diff_desc=diff_desc,
file_results=file_results,
annotate=args.annotate,
context_lines=args.context,
),
)
else:
render_text_report(
diff_desc=diff_desc,
file_results=file_results,
annotate=args.annotate,
context_lines=args.context,
)
if args.github_annotations:
emit_github_annotations(file_results)
# 8. Check optional threshold gates.
failed_gate = False
for metric_name, threshold in [
('lines', args.fail_under_lines),
('regions', args.fail_under_regions),
('branches', args.fail_under_branches),
]:
if threshold is None:
continue
agg = MetricCounts()
for fc in file_results:
if fc.in_coverage_map:
m: MetricCounts = getattr(fc, metric_name)
agg.add(m.covered, m.total)
if agg.percent is not None and agg.percent < threshold:
print(
f'FAIL: Diff {metric_name} coverage ({agg.percent:.1f}%) is below threshold ({threshold:.1f}%).',
file=sys.stderr,
)
failed_gate = True
if failed_gate:
sys.exit(1)
if __name__ == '__main__':
main()