384 lines
12 KiB
Python
384 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
uos_covparse.py — Parse .umap coverage dumps and fold hits into .umdb datasets.
|
|
|
|
Reads a UMAP/1 hex dump produced by the on-target libuoscov runtime and
|
|
correlates each byte/bit with ipoint semantics from a .umdb structural
|
|
database. Writes hit records into a named dataset.
|
|
|
|
CLI: uos_covparse.py --map project.umdb run.umap --dataset test1 [--merge-results]
|
|
|
|
Uses libuosipoint for .umdb I/O and data model.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import re
|
|
import struct
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from libuosipoint import (
|
|
open_umdb,
|
|
create_umdb,
|
|
KIND_STMT,
|
|
KIND_DECISION,
|
|
)
|
|
|
|
|
|
# ─── .umap parser ─────────────────────────────────────────────────────────────
|
|
|
|
@dataclass
|
|
class UmapHeader:
|
|
"""Parsed .umap file header."""
|
|
format_version: str = ""
|
|
build_id: str = ""
|
|
map_crc: int = 0
|
|
|
|
|
|
@dataclass
|
|
class UmapSegment:
|
|
"""A single TEST segment in a .umap file."""
|
|
test_id: int = 0
|
|
description: str = ""
|
|
|
|
|
|
@dataclass
|
|
class UmapData:
|
|
"""Fully parsed .umap file."""
|
|
header: UmapHeader = None
|
|
segments: list[UmapSegment] = None
|
|
size: int = 0
|
|
map_bytes: bytes = b""
|
|
|
|
def __post_init__(self):
|
|
if self.header is None:
|
|
self.header = UmapHeader()
|
|
if self.segments is None:
|
|
self.segments = []
|
|
|
|
|
|
def parse_umap(text: str) -> UmapData:
|
|
"""Parse a UMAP/1 text dump into structured data.
|
|
|
|
Format:
|
|
UMAP/1
|
|
build_id: <id>
|
|
[map_crc: <hex>]
|
|
[TEST <id> <description>]
|
|
...
|
|
size: <n>
|
|
<hex bytes>
|
|
|
|
Returns the parsed UmapData.
|
|
Raises ValueError on parse errors.
|
|
"""
|
|
lines = text.strip().splitlines()
|
|
if not lines:
|
|
raise ValueError("Empty .umap file")
|
|
|
|
data = UmapData()
|
|
i = 0
|
|
|
|
# Parse UMAP/1 header
|
|
if not lines[0].startswith("UMAP/"):
|
|
raise ValueError(f"Missing UMAP header, got: {lines[0]!r}")
|
|
data.header.format_version = lines[0].strip()
|
|
i = 1
|
|
|
|
# Parse header fields (build_id, map_crc)
|
|
while i < len(lines):
|
|
line = lines[i].strip()
|
|
if line.lower().startswith("build_id:"):
|
|
data.header.build_id = line.split(":", 1)[1].strip()
|
|
i += 1
|
|
elif line.lower().startswith("map_crc:"):
|
|
crc_str = line.split(":", 1)[1].strip()
|
|
data.header.map_crc = int(crc_str, 16) if len(crc_str) <= 8 else int(crc_str, 10)
|
|
i += 1
|
|
elif line.upper().startswith("TEST "):
|
|
break
|
|
elif line.lower().startswith("size:"):
|
|
break
|
|
else:
|
|
i += 1
|
|
|
|
# Parse TEST segments
|
|
while i < len(lines):
|
|
line = lines[i].strip()
|
|
if line.upper().startswith("TEST "):
|
|
parts = line.split(None, 2)
|
|
test_id = int(parts[1]) if len(parts) > 1 else 0
|
|
desc = parts[2] if len(parts) > 2 else ""
|
|
data.segments.append(UmapSegment(test_id=test_id, description=desc))
|
|
i += 1
|
|
elif line.lower().startswith("size:"):
|
|
break
|
|
else:
|
|
i += 1
|
|
|
|
# Parse size line
|
|
if i >= len(lines):
|
|
raise ValueError("Missing size line")
|
|
line = lines[i].strip()
|
|
if not line.lower().startswith("size:"):
|
|
raise ValueError(f"Expected size line, got: {line!r}")
|
|
data.size = int(line.split(":", 1)[1].strip())
|
|
i += 1
|
|
|
|
# Parse hex bytes
|
|
hex_str = ""
|
|
while i < len(lines):
|
|
hex_str += lines[i].strip()
|
|
i += 1
|
|
|
|
# Remove whitespace from hex string
|
|
hex_str = re.sub(r'\s+', '', hex_str)
|
|
|
|
if not hex_str:
|
|
raise ValueError("Missing coverage map data")
|
|
|
|
try:
|
|
data.map_bytes = bytes.fromhex(hex_str)
|
|
except ValueError:
|
|
raise ValueError(f"Invalid hex data in coverage map")
|
|
|
|
if len(data.map_bytes) != data.size:
|
|
raise ValueError(
|
|
f"Map size mismatch: header says {data.size}, "
|
|
f"got {len(data.map_bytes)} bytes"
|
|
)
|
|
|
|
return data
|
|
|
|
|
|
def compute_map_crc(data: bytes) -> int:
|
|
"""Compute CRC32 of the coverage map data."""
|
|
return struct.unpack("!I", hashlib.md5(data).digest()[:4])[0] & 0xFFFFFFFF
|
|
|
|
|
|
# ─── Dataset folding ───────────────────────────────────────────────────────────
|
|
|
|
def fold_hits(
|
|
conn,
|
|
umap: UmapData,
|
|
dataset_name: str,
|
|
merge: bool = False,
|
|
) -> dict:
|
|
"""Fold .umap coverage data into the .umdb as a named dataset.
|
|
|
|
For each byte in the coverage map, look up the corresponding ipoint.
|
|
If the byte is non-zero, record a hit for that ipoint in the dataset.
|
|
For decision ipoints, the vector bytes are folded into mcdc_hits.
|
|
|
|
Args:
|
|
conn: SQLite connection to the .umdb.
|
|
umap: Parsed .umap data.
|
|
dataset_name: Name for the dataset.
|
|
merge: If True, merge with existing dataset of the same name.
|
|
|
|
Returns:
|
|
Summary dict with counts of hits recorded.
|
|
"""
|
|
cur = conn.cursor()
|
|
|
|
# Get all ipoints ordered by id
|
|
ipoints = {}
|
|
for row in cur.execute("SELECT id, function_id, kind, line, col, n_cond, map_byte, map_bits FROM ipoints"):
|
|
ipoints[row["id"]] = dict(row)
|
|
|
|
if not ipoints:
|
|
raise ValueError("No ipoints found in .umdb — run uos-xst first")
|
|
|
|
# Get the build_id from the .umdb if available
|
|
db_build_id = ""
|
|
try:
|
|
row = cur.execute("SELECT build_id FROM builds LIMIT 1").fetchone()
|
|
if row:
|
|
db_build_id = row["build_id"]
|
|
except Exception:
|
|
pass
|
|
|
|
# Validate build_id if present in both
|
|
if umap.header.build_id and db_build_id:
|
|
if umap.header.build_id != db_build_id:
|
|
print(
|
|
f"Warning: build_id mismatch: .umdb={db_build_id}, .umap={umap.header.build_id}",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
# Create or find the dataset
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
if merge:
|
|
row = cur.execute(
|
|
"SELECT id FROM datasets WHERE name = ?", (dataset_name,)
|
|
).fetchone()
|
|
if row:
|
|
dataset_id = row["id"]
|
|
else:
|
|
cur.execute(
|
|
"INSERT INTO datasets (name, build_id, created) VALUES (?, ?, ?)",
|
|
(dataset_name, umap.header.build_id, now),
|
|
)
|
|
dataset_id = cur.lastrowid
|
|
else:
|
|
# Check for existing dataset
|
|
row = cur.execute(
|
|
"SELECT id FROM datasets WHERE name = ?", (dataset_name,)
|
|
).fetchone()
|
|
if row:
|
|
print(
|
|
f"Dataset '{dataset_name}' already exists (id={row['id']}). "
|
|
f"Use --merge-results to merge.",
|
|
file=sys.stderr,
|
|
)
|
|
dataset_id = row["id"]
|
|
else:
|
|
cur.execute(
|
|
"INSERT INTO datasets (name, build_id, created) VALUES (?, ?, ?)",
|
|
(dataset_name, umap.header.build_id, now),
|
|
)
|
|
dataset_id = cur.lastrowid
|
|
|
|
# Fold hits from the coverage map
|
|
stmt_hits = 0
|
|
decision_hits = 0
|
|
mcdc_vectors = 0
|
|
total_ips = len(ipoints)
|
|
|
|
for ip_id, ip in sorted(ipoints.items()):
|
|
map_byte = ip["map_byte"]
|
|
kind = ip["kind"]
|
|
|
|
if map_byte >= len(umap.map_bytes):
|
|
continue
|
|
|
|
if kind == KIND_DECISION:
|
|
# Decision: byte at map_byte is hit flag
|
|
hit_flag = umap.map_bytes[map_byte]
|
|
if hit_flag:
|
|
decision_hits += 1
|
|
# Record the statement hit for the decision
|
|
cur.execute(
|
|
"INSERT OR REPLACE INTO hits (dataset_id, ipoint_id, count) "
|
|
"VALUES (?, ?, ?)",
|
|
(dataset_id, ip_id, hit_flag),
|
|
)
|
|
|
|
# Fold MC/DC vector bytes
|
|
n_cond = ip["n_cond"]
|
|
vec_bytes = (n_cond + 7) // 8
|
|
vector_bits = 0
|
|
for b in range(vec_bytes):
|
|
vb = map_byte + 1 + b
|
|
if vb < len(umap.map_bytes):
|
|
byte_val = umap.map_bytes[vb]
|
|
vector_bits |= byte_val << (8 * b)
|
|
|
|
if vector_bits:
|
|
mcdc_vectors += 1
|
|
cur.execute(
|
|
"INSERT INTO mcdc_hits (dataset_id, decision_id, vector_bits) "
|
|
"VALUES (?, ?, ?)",
|
|
(dataset_id, ip_id, vector_bits),
|
|
)
|
|
else:
|
|
# Statement/call/function: byte at map_byte = hit count
|
|
count = umap.map_bytes[map_byte]
|
|
if count:
|
|
stmt_hits += 1
|
|
cur.execute(
|
|
"INSERT OR REPLACE INTO hits (dataset_id, ipoint_id, count) "
|
|
"VALUES (?, ?, ?)",
|
|
(dataset_id, ip_id, count),
|
|
)
|
|
|
|
conn.commit()
|
|
|
|
return {
|
|
"dataset_id": dataset_id,
|
|
"dataset_name": dataset_name,
|
|
"total_ipoints": total_ips,
|
|
"stmt_hits": stmt_hits,
|
|
"decision_hits": decision_hits,
|
|
"mcdc_vectors": mcdc_vectors,
|
|
"map_size": umap.size,
|
|
"segments": len(umap.segments),
|
|
}
|
|
|
|
|
|
# ─── CLI ───────────────────────────────────────────────────────────────────────
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="uos-covparse: Parse .umap coverage dumps into .umdb datasets",
|
|
prog="uos_covparse.py",
|
|
)
|
|
parser.add_argument(
|
|
"--map",
|
|
required=True,
|
|
help="Path to the .umdb structural database",
|
|
)
|
|
parser.add_argument(
|
|
"umap",
|
|
help="Path to the .umap coverage dump",
|
|
)
|
|
parser.add_argument(
|
|
"--dataset",
|
|
required=True,
|
|
help="Name for the coverage dataset",
|
|
)
|
|
parser.add_argument(
|
|
"--merge-results",
|
|
action="store_true",
|
|
help="Merge results into existing dataset of the same name",
|
|
)
|
|
|
|
args = parser.parse_args(argv)
|
|
|
|
# Validate input files
|
|
umdb_path = Path(args.map)
|
|
if not umdb_path.exists():
|
|
print(f"Error: .umdb not found: {umdb_path}", file=sys.stderr)
|
|
return 1
|
|
|
|
umap_path = Path(args.umap)
|
|
if not umap_path.exists():
|
|
print(f"Error: .umap not found: {umap_path}", file=sys.stderr)
|
|
return 1
|
|
|
|
# Read the .umap file
|
|
try:
|
|
umap_text = umap_path.read_text()
|
|
umap = parse_umap(umap_text)
|
|
except Exception as e:
|
|
print(f"Error parsing .umap: {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"Parsed .umap: build_id={umap.header.build_id!r}, "
|
|
f"size={umap.size}, segments={len(umap.segments)}")
|
|
|
|
# Open the .umdb and fold hits
|
|
try:
|
|
conn = open_umdb(umdb_path)
|
|
result = fold_hits(conn, umap, args.dataset, args.merge_results)
|
|
conn.close()
|
|
except Exception as e:
|
|
print(f"Error folding hits: {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"Dataset '{result['dataset_name']}' (id={result['dataset_id']}): "
|
|
f"{result['stmt_hits']}/{result['total_ipoints']} stmt hits, "
|
|
f"{result['decision_hits']} decision hits, "
|
|
f"{result['mcdc_vectors']} MC/DC vectors")
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|