universalisos/tools/uos-cover/uos_cins.py

690 lines
25 KiB
Python

#!/usr/bin/env python3
"""
uos_cins.py — C source instrumenter for UniversalisOS coverage toolchain.
Regex-based source-to-source instrumenter that parses C source and inserts
coverage instrumentation macros (UOS_I, UOS_E, UOS_T, UOS_F, UOS_CM, UOS_DM).
Input: C source file + profile
Output: Instrumented C source + .uxsc structural record
Profiles: COV_STATEMENTS, COV_DECISIONS, COV_MCDC, COV_CALLS, COV_FUNCTIONS
"""
from __future__ import annotations
import re
import sys
import argparse
from pathlib import Path
_SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(_SCRIPT_DIR))
from libuosipoint import (
FileRecord, Function, Ipoint, Condition, write_uxsc, file_md5,
KIND_STMT, KIND_DECISION, KIND_CALL, KIND_FUNCTION,
)
# ─── Comment stripping ──────────────────────────────────────────────────────
def strip_comments(src: str) -> str:
"""Remove C/C++ comments from source, preserving line structure."""
out: list[str] = []
i = 0
in_str = False
str_ch = ""
while i < len(src):
c = src[i]
if in_str:
out.append(c)
if c == "\\" and i + 1 < len(src):
i += 1
out.append(src[i])
elif c == str_ch:
in_str = False
i += 1
elif c in ('"', "'"):
in_str = True
str_ch = c
out.append(c)
i += 1
elif c == "/" and i + 1 < len(src):
nc = src[i + 1]
if nc == "/":
while i < len(src) and src[i] != "\n":
i += 1
elif nc == "*":
i += 2
while i < len(src) - 1:
if src[i] == "*" and src[i + 1] == "/":
i += 2
break
i += 1
else:
out.append(c)
i += 1
else:
out.append(c)
i += 1
return "".join(out)
# ─── Keywords ────────────────────────────────────────────────────────────────
_KEYWORDS = frozenset({
"if", "else", "for", "while", "do", "switch", "case", "default",
"return", "break", "continue", "goto", "sizeof", "typedef",
"struct", "enum", "union", "const", "volatile", "static", "extern",
"inline", "register", "auto",
"void", "char", "short", "int", "long", "float", "double",
"_Bool", "_Complex", "bool", "true", "false", "NULL",
"unsigned", "signed",
"int8_t", "uint8_t", "int16_t", "uint16_t", "int32_t", "uint32_t",
"int64_t", "uint64_t", "size_t", "ssize_t", "ptrdiff_t",
"UOS_I", "UOS_E", "UOS_T", "UOS_F", "UOS_CM", "UOS_DM", "UOS_DML",
"UOS_Init", "UOS_Output", "UOS_Build_Id", "UOS_Begin_Test",
})
_TYPE_STARTS = frozenset({
"unsigned", "signed", "const", "volatile", "static", "extern",
"inline", "struct", "enum", "union", "long", "short",
})
# ─── Paren extraction ────────────────────────────────────────────────────────
def extract_paren(line: str, start: int) -> tuple[str, int, int] | None:
"""Extract parenthesised expression. Returns (inner, open_pos, close_pos) or None."""
pos = line.find("(", start)
if pos < 0:
return None
depth = 0
for ci in range(pos, len(line)):
ch = line[ci]
if ch == "(":
depth += 1
elif ch == ")":
depth -= 1
if depth == 0:
return line[pos + 1 : ci], pos, ci
return None
def split_top_level(expr: str, sep: str) -> list[str]:
"""Split *expr* by a two-char separator at the top paren level."""
parts: list[str] = []
depth = 0
cur: list[str] = []
i = 0
while i < len(expr):
c = expr[i]
if c == "(":
depth += 1
elif c == ")":
depth -= 1
if depth == 0 and i + 1 < len(expr) and expr[i : i + 2] == sep:
parts.append("".join(cur).strip())
cur = []
i += 1
else:
cur.append(c)
i += 1
rem = "".join(cur).strip()
if rem:
parts.append(rem)
return [p for p in parts if p]
def split_semicolons(expr: str) -> list[str]:
"""Split by top-level semicolons (for for-loop init;cond;incr)."""
parts: list[str] = []
depth = 0
cur: list[str] = []
for c in expr:
if c == "(":
depth += 1
elif c == ")":
depth -= 1
if c == ";" and depth == 0:
parts.append("".join(cur).strip())
cur = []
else:
cur.append(c)
parts.append("".join(cur).strip())
return parts
# ─── Function detection ─────────────────────────────────────────────────────
def find_functions(clean_source: str) -> list[tuple[str, int, int]]:
"""Find function definitions. Returns [(name, start_line_0, end_line_0)]."""
lines = clean_source.split("\n")
functions: list[tuple[str, int, int]] = []
i = 0
while i < len(lines):
line = lines[i].strip()
if not line or line.startswith("#") or line in ("{", "}"):
i += 1
continue
# Heuristic: line contains an identifier followed by '(' and no ';' before '('
m = re.match(
r"^(?:(?:static|inline|extern|__attribute__\s*\(\([^)]*\)\))\s+)*"
r"(?:(?:unsigned|signed|const|volatile|long|short|struct|enum|union)\s+)*"
r"(?:\w+\s*\**\s+)*"
r"(\w+)\s*\(",
line,
)
if not m:
i += 1
continue
paren_pos = line.find("(", m.start())
if paren_pos < 0:
i += 1
continue
before = line[:paren_pos].rstrip()
if ";" in before:
i += 1
continue
name_m = re.search(r"(\w+)\s*$", before)
if not name_m:
i += 1
continue
name = name_m.group(1)
if name in _KEYWORDS or name in _TYPE_STARTS:
i += 1
continue
# Find opening '{'
j = i
found = False
while j < min(i + 12, len(lines)):
if "{" in lines[j]:
found = True
break
if ";" in lines[j]:
break
j += 1
if not found:
i += 1
continue
func_start = j
brace_depth = 0
k = j
while k < len(lines):
for ch in lines[k]:
if ch == "{":
brace_depth += 1
elif ch == "}":
brace_depth -= 1
if brace_depth == 0:
functions.append((name, func_start, k))
break
if brace_depth == 0:
break
k += 1
i = k + 1 if brace_depth == 0 else j + 1
return functions
# ─── Instrumenter ────────────────────────────────────────────────────────────
class CInstrumenter:
"""Regex-based C source instrumenter."""
def __init__(self, source: str, filepath: str, profiles: set[str]):
self.source = source
self.filepath = filepath
self.profiles = profiles
self._next_id = 1
self.functions: list[Function] = []
self._cur_fn: Function | None = None
self._else_stack: list[int] = [] # pending false-branch ipoint IDs
# ── id allocation ────────────────────────────────────────────────────
def _id(self) -> int:
v = self._next_id
self._next_id += 1
return v
# ── main entry ───────────────────────────────────────────────────────
def instrument(self) -> tuple[str, FileRecord]:
clean = strip_comments(self.source)
func_ranges = find_functions(clean)
func_by_line: dict[int, Function] = {}
for name, start, end in func_ranges:
fn = Function(name=name, line=start + 1, end_line=end + 1)
self.functions.append(fn)
for ln in range(start, end + 1):
func_by_line[ln] = fn
lines = self.source.split("\n")
clean_lines = clean.split("\n")
output = list(lines)
in_func = False
brace_depth = 0
for ln in range(len(output)):
line = output[ln]
cl = clean_lines[ln] if ln < len(clean_lines) else ""
stripped = cl.strip()
if ln in func_by_line:
self._cur_fn = func_by_line[ln]
in_func = True
if not in_func or self._cur_fn is None:
continue
if stripped.startswith("#"):
continue
result = line
# Decisions / MC/DC
if self.profiles & {"COV_DECISIONS", "COV_MCDC"}:
result = self._do_decision(result, cl, ln)
# Statements
if "COV_STATEMENTS" in self.profiles:
result = self._do_stmt(result, cl, ln)
# Calls
if "COV_CALLS" in self.profiles:
result = self._do_calls(result, cl, ln)
output[ln] = result
# Track brace depth AFTER instrumentation so single-line
# functions are fully instrumented before } sets in_func=False.
for ch in cl:
if ch == "{":
brace_depth += 1
elif ch == "}":
brace_depth -= 1
if brace_depth == 0:
in_func = False
self._cur_fn = None
self._else_stack.clear()
record = FileRecord(
file=self.filepath,
md5=file_md5(self.filepath) if Path(self.filepath).exists() else "",
functions=self.functions,
)
return "\n".join(output), record
# ── decision instrumentation ─────────────────────────────────────────
def _do_decision(self, line: str, clean: str, ln: int) -> str:
s = clean.strip()
# else if
if re.search(r"\belse\s+if\s*\(", s):
return self._inst_else_if(line, clean, ln)
# plain else
if re.search(r"\belse\b", s) and not re.search(r"\belse\s+if\b", s):
return self._inst_else(line, clean, ln)
# if
if re.search(r"\bif\s*\(", s):
return self._inst_if(line, clean, ln)
# for (before while, because for contains ;)
if re.search(r"\bfor\s*\(", s):
return self._inst_for(line, clean, ln)
# do-while (} while)
if re.search(r"\}\s*while\s*\(", s):
return self._inst_do_while(line, clean, ln)
# plain while (not do-while)
if re.search(r"\bwhile\s*\(", s) and not re.search(r"\}\s*while", s):
return self._inst_while(line, clean, ln)
# switch
if re.search(r"\bswitch\s*\(", s):
return self._inst_switch(line, clean, ln)
return line
# ── if ───────────────────────────────────────────────────────────────
def _inst_if(self, line: str, clean: str, ln: int) -> str:
m = re.search(r"\bif\s*\(", clean)
if not m:
return line
r = extract_paren(clean, m.start())
if not r:
return line
cond, ps, pe = r
dec_id = self._id()
# MC/DC path
if "COV_MCDC" in self.profiles:
parts = split_top_level(cond, "&&") + split_top_level(cond, "||")
if len(parts) > 1:
line = self._replace_cond(line, clean, ps, pe, cond, dec_id, ln)
self._else_stack.append(dec_id)
return line
# Regular decision
ip = Ipoint(
id=dec_id, kind=KIND_DECISION, line=ln + 1, n_cond=1,
conditions=[Condition(idx=0, text=cond.strip())],
)
self._cur_fn.ipoints.append(ip)
line = self._replace_cond(line, clean, ps, pe, cond, dec_id, ln)
self._else_stack.append(dec_id)
return line
# ── else if ──────────────────────────────────────────────────────────
def _inst_else_if(self, line: str, clean: str, ln: int) -> str:
# Pop previous decision — its false branch is implicit here
if self._else_stack:
prev_id = self._else_stack.pop()
m = re.search(r"\belse\s+if\s*\(", clean)
if not m:
return line
r = extract_paren(clean, m.start())
if not r:
return line
cond, ps, pe = r
dec_id = self._id()
if "COV_MCDC" in self.profiles:
parts = split_top_level(cond, "&&") + split_top_level(cond, "||")
if len(parts) > 1:
line = self._replace_cond(line, clean, ps, pe, cond, dec_id, ln)
self._else_stack.append(dec_id)
return line
ip = Ipoint(
id=dec_id, kind=KIND_DECISION, line=ln + 1, n_cond=1,
conditions=[Condition(idx=0, text=cond.strip())],
)
self._cur_fn.ipoints.append(ip)
line = self._replace_cond(line, clean, ps, pe, cond, dec_id, ln)
self._else_stack.append(dec_id)
return line
# ── else ─────────────────────────────────────────────────────────────
def _inst_else(self, line: str, clean: str, ln: int) -> str:
if not self._else_stack:
return line
false_id = self._else_stack.pop()
# Insert UOS_F after 'else {'
m = re.search(r"\belse\s*\{", clean)
if m:
# find '{' position in original line
else_end_in_clean = m.end()
brace_in_clean = clean.find("{", m.start())
if brace_in_clean >= 0:
# Map brace position to original line
# Find the { in the original line around the same area
orig_brace = line.find("{", m.start())
if orig_brace >= 0:
return line[: orig_brace + 1] + f" UOS_F({false_id});" + line[orig_brace + 1 :]
# Fallback: insert after 'else'
m2 = re.search(r"\belse\b", clean)
if m2:
orig_else_end = line.find("else", m2.start()) + 4
return line[:orig_else_end] + f" UOS_F({false_id});" + line[orig_else_end:]
return line
# ── while ────────────────────────────────────────────────────────────
def _inst_while(self, line: str, clean: str, ln: int) -> str:
m = re.search(r"\bwhile\s*\(", clean)
if not m:
return line
r = extract_paren(clean, m.start())
if not r:
return line
cond, ps, pe = r
dec_id = self._id()
if "COV_MCDC" in self.profiles:
parts = split_top_level(cond, "&&") + split_top_level(cond, "||")
if len(parts) > 1:
return self._replace_cond(line, clean, ps, pe, cond, dec_id, ln)
ip = Ipoint(
id=dec_id, kind=KIND_DECISION, line=ln + 1, n_cond=1,
conditions=[Condition(idx=0, text=cond.strip())],
)
self._cur_fn.ipoints.append(ip)
return self._replace_cond(line, clean, ps, pe, cond, dec_id, ln)
# ── for ──────────────────────────────────────────────────────────────
def _inst_for(self, line: str, clean: str, ln: int) -> str:
m = re.search(r"\bfor\s*\(", clean)
if not m:
return line
r = extract_paren(clean, m.start())
if not r:
return line
full, ps, pe = r
parts = split_semicolons(full)
if len(parts) != 3 or not parts[1]:
return line
cond = parts[1]
dec_id = self._id()
ip = Ipoint(
id=dec_id, kind=KIND_DECISION, line=ln + 1, n_cond=1,
conditions=[Condition(idx=0, text=cond)],
)
self._cur_fn.ipoints.append(ip)
# Build new for expression: init; UOS_E(id,(cond)); incr
new_for = parts[0] + "; " + f"UOS_E({dec_id}, ({cond}))" + "; " + parts[2]
return line[: ps + 1] + new_for + line[pe :]
# ── do-while ─────────────────────────────────────────────────────────
def _inst_do_while(self, line: str, clean: str, ln: int) -> str:
m = re.search(r"while\s*\(", clean)
if not m:
return line
r = extract_paren(clean, m.start())
if not r:
return line
cond, ps, pe = r
dec_id = self._id()
ip = Ipoint(
id=dec_id, kind=KIND_DECISION, line=ln + 1, n_cond=1,
conditions=[Condition(idx=0, text=cond.strip())],
)
self._cur_fn.ipoints.append(ip)
return self._replace_cond(line, clean, ps, pe, cond, dec_id, ln)
# ── switch ───────────────────────────────────────────────────────────
def _inst_switch(self, line: str, clean: str, ln: int) -> str:
m = re.search(r"\bswitch\s*\(", clean)
if not m:
return line
r = extract_paren(clean, m.start())
if not r:
return line
cond, ps, pe = r
dec_id = self._id()
ip = Ipoint(
id=dec_id, kind=KIND_DECISION, line=ln + 1, n_cond=1,
conditions=[Condition(idx=0, text=cond.strip())],
)
self._cur_fn.ipoints.append(ip)
return self._replace_cond(line, clean, ps, pe, cond, dec_id, ln)
# ── MC/DC helper ─────────────────────────────────────────────────────
def _apply_mcdc(self, line: str, clean: str, ps: int, pe: int,
cond: str, dec_id: int, ln: int) -> str:
ands = split_top_level(cond, "&&")
ors = split_top_level(cond, "||")
# Use && parts first; if only one, try || parts
parts = ands if len(ands) > 1 else ors if len(ors) > 1 else [cond]
n = len(parts)
indent = line[: len(line) - len(line.lstrip())]
key_decl = f"{indent}unsigned int uos_k{dec_id} = 0;"
cm_parts = [
f"UOS_CM(&uos_k{dec_id}, {i}, ({p.strip()}))" for i, p in enumerate(parts)
]
mcdc_expr = " && ".join(cm_parts)
ip = Ipoint(
id=dec_id, kind=KIND_DECISION, line=ln + 1, n_cond=n,
conditions=[Condition(idx=i, text=p.strip()) for i, p in enumerate(parts)],
map_bits=(n + 7) // 8,
)
self._cur_fn.ipoints.append(ip)
wrapped = f"UOS_DM({dec_id}, &uos_k{dec_id}, {mcdc_expr})"
new_line = line[: ps + 1] + wrapped + line[pe :]
return f"{key_decl}\n{new_line}"
# ── condition replacement helper ─────────────────────────────────────
def _replace_cond(self, line: str, clean: str, ps: int, pe: int,
cond: str, dec_id: int, ln: int) -> str:
"""Replace condition text with UOS_E(dec_id, (cond)) or MC/DC variant."""
if "COV_MCDC" in self.profiles:
ands = split_top_level(cond, "&&")
ors = split_top_level(cond, "||")
parts = ands if len(ands) > 1 else ors if len(ors) > 1 else []
if len(parts) > 1:
return self._apply_mcdc(line, clean, ps, pe, cond, dec_id, ln)
ip = Ipoint(
id=dec_id, kind=KIND_DECISION, line=ln + 1, n_cond=1,
conditions=[Condition(idx=0, text=cond.strip())],
)
self._cur_fn.ipoints.append(ip)
return line[: ps + 1] + f"UOS_E({dec_id}, ({cond}))" + line[pe :]
# ── statement instrumentation ────────────────────────────────────────
def _do_stmt(self, line: str, clean: str, ln: int) -> str:
s = clean.strip()
if not s:
return line
if s.startswith("#"):
return line
if not s.endswith(";"):
return line
if s in ("}", "};", "{", "){"):
return line
if s.startswith("}"):
return line
# Skip declarations (type keyword at start)
if re.match(
r"^(?:static|extern|const|volatile|register|unsigned|signed|"
r"long|short|struct|enum|union|void|char|int|float|double|"
r"_Bool|_Complex|size_t|ssize_t|ptrdiff_t|uint\d+_t|int\d+_t)\s",
s,
):
return line
# Skip labels
if re.match(r"^\w+\s*:", s):
return line
# Skip already-instrumented lines
if "UOS_I(" in line or "UOS_E(" in line:
return line
sid = self._id()
ip = Ipoint(id=sid, kind=KIND_STMT, line=ln + 1)
self._cur_fn.ipoints.append(ip)
indent_m = re.match(r"^(\s*)", line)
indent = indent_m.group(1) if indent_m else ""
return f"{indent}UOS_I({sid}); {line.lstrip()}"
# ── call instrumentation ─────────────────────────────────────────────
def _do_calls(self, line: str, clean: str, ln: int) -> str:
s = clean.strip()
if not s or s.startswith("#"):
return line
if "UOS_I(" in line:
return line
# Skip function definition lines (e.g. "int add(int a, int b) {")
if s.endswith("{") and "(" in s:
return line
calls = []
for cm in re.finditer(r"\b(\w+)\s*\(", clean):
name = cm.group(1)
if name not in _KEYWORDS and not name.startswith("UOS_"):
calls.append((name, cm.start()))
if not calls:
return line
result = line
for name, _start in reversed(calls):
cid = self._id()
ip = Ipoint(id=cid, kind=KIND_CALL, line=ln + 1)
self._cur_fn.ipoints.append(ip)
# Find the call in the (possibly already modified) line
pos = result.find(name + "(")
if pos >= 0:
indent_m = re.match(r"^(\s*)", result)
indent = indent_m.group(1) if indent_m else ""
result = result[:pos] + f"UOS_I({cid}); " + result[pos:]
return result
# ─── Standalone API ──────────────────────────────────────────────────────────
def instrument(source: str, filepath: str, profile_str: str) -> tuple[str, FileRecord]:
"""Instrument C source. *profile_str* may be compound (e.g. 'A+B')."""
profiles = set(profile_str.split("+")) if isinstance(profile_str, str) else profile_str
inst = CInstrumenter(source, filepath, profiles)
return inst.instrument()
# ─── CLI ─────────────────────────────────────────────────────────────────────
def main() -> None:
p = argparse.ArgumentParser(description="uos-cins: C source coverage instrumenter")
p.add_argument("source", help="Input C source file")
p.add_argument(
"--profile",
required=True,
choices=[
"COV_STATEMENTS", "COV_DECISIONS", "COV_MCDC",
"COV_CALLS", "COV_FUNCTIONS",
],
help="Coverage profile",
)
p.add_argument("-o", "--output", required=True, help="Output instrumented C file")
p.add_argument("--uxsc", required=True, help="Output .uxsc structural record")
args = p.parse_args()
src = Path(args.source).read_text()
output, record = instrument(src, args.source, args.profile)
Path(args.output).write_text(output)
write_uxsc(record, args.uxsc)
n_ipoints = sum(len(fn.ipoints) for fn in record.functions)
print(f"uos-cins: {len(record.functions)} functions, {n_ipoints} ipoints")
print(f" source -> {args.output}")
print(f" uxsc -> {args.uxsc}")
if __name__ == "__main__":
main()