diff options
| author | hathach <[email protected]> | 2026-09-03 17:14:39 +0700 |
|---|---|---|
| committer | hathach <[email protected]> | 2026-09-04 04:21:57 +0700 |
| commit | 9e681141bc24ff71081dfce683ae4371e7520124 (patch) | |
| tree | 582dfcaad9b9f23be348143f12f1f9d0520e2052 | |
| parent | dcb3fcfd30ac03b810d6de82674c008d5163400f (diff) | |
hil: sysview CI - rig capture subcommands and report generator
sysview_ci.py builds, flashes and captures the sysview-flagged boards
(stm32f407disco and raspberry_pi_pico in tinyusb.json) on the rig, taking
each board's flock via hil_lock and restoring park firmware afterwards.
Hard-won correctness baked in from the dogfood runs: firmware paths carry
the flasher's extension via FLASHER_SUFFIX (JLinkExe and OpenOCD infer the
image format from it, and the RTT control block still comes from the ELF,
which is not always the flashed image); WCH boards are detected by their
wch-riscv target config (is_wch_board(), the plain "openocd" flasher name
since #3804 folded openocd_wch/openocd_adi away); build and flash failures
are excerpted from BOTH ends (cmake prints the diagnosis first and the call
stack last, ninja the reverse) and flash failures include the flasher's own
output so rc=124 can distinguish a wedged probe from a dead target; a bad
ELF no longer aborts the whole multi-board run; the enumeration wait sits
after the capture session's own reset; each capture picks a free RTT server
port. session_resets()/sysview.attach_only skips the in-session reset for
boards that never come back from it (metro_m4_express: both campaign
captures died with 'No control block found' while attach-without-reset
streamed immediately) - the flasher's post-flash reset already supplied the
fresh boot.
The suites join the pre-commit hil-test hook and master's test/hil/test/
carve-out pin.
| -rw-r--r-- | .pre-commit-config.yaml | 8 | ||||
| -rw-r--r-- | test/hil/sysview_ci.py | 668 | ||||
| -rw-r--r-- | test/hil/test/test_ci_select.py | 3 | ||||
| -rw-r--r-- | test/hil/test/test_sysview_ci.py | 448 | ||||
| -rw-r--r-- | test/hil/test/test_sysview_record.py | 33 | ||||
| -rw-r--r-- | test/hil/test/test_sysview_report.py | 419 | ||||
| -rw-r--r-- | test/hil/tinyusb.json | 13 | ||||
| -rwxr-xr-x | tools/ci_select.py | 6 |
8 files changed, 1594 insertions, 4 deletions
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 537ed3bc4..7c9afcaa5 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -66,11 +66,13 @@ repos: # pre-commit CI. # No types_or: the rig rosters (*.json) are inputs too. # examples/device/mtp/src is in scope: test_hil_bounded parses README_TXT_CONTENT - # and md5-checks the logo header from there as its MTP fixtures. + # and md5-checks the logo header from there as its MTP fixtures. The test_[hs]* pattern + # is test_hil*.py plus the test_sysview*.py suites; .claude/skills/sysview/scripts is in + # scope because test_sysview_report exercises sysview_report.py, which lives there. - id: hil-test name: hil-test - files: ^(test/hil/|examples/device/mtp/src/|tools/rtt\.py$) - entry: python3 -m unittest discover -s test/hil/test -p 'test_hil*.py' + files: ^(test/hil/|examples/device/mtp/src/|tools/rtt\.py$|\.claude/skills/sysview/scripts/) + entry: python3 -m unittest discover -s test/hil/test -p 'test_[hs]*.py' pass_filenames: false language: system # hil-validate.js decides which boards ship. Its result join has been wrong three times -- diff --git a/test/hil/sysview_ci.py b/test/hil/sysview_ci.py new file mode 100644 index 000000000..5a7ac6582 --- /dev/null +++ b/test/hil/sysview_ci.py @@ -0,0 +1,668 @@ +#!/usr/bin/env python3 +"""SystemView CI: per-board capture on the HIL rig + PR performance report. + +report: pure -- two directories of sysview-<board>.json in, markdown out. +capture (Task 5): flash SYSVIEW build, drive workload, RTT-capture, decode. +Spec: docs/superpowers/specs/2026-07-29-sysview-hil-report-design.md +""" +import argparse, glob, json, os, re, sys + +GATE_MIN_N = 50 +# N13: wave 3 changed pct() (sysview_report.py) to the conventional nearest-rank index +# int((n-1)*p/100) -- with that formula the p99 index is already strictly below the max index +# for every n>=2 (0.99*(n-1) < n-1 whenever n-1>=1), so P99_MIN_N is no longer about avoiding a +# collision with the max sample (the old "distinct from the max sample" reasoning, true only of +# the pre-wave-3 int(n*p/100) formula). It is kept at 100 for a different, still-valid reason: +# a tail percentile needs roughly n>=1/(1-p) samples to be a meaningful estimate rather than +# effectively picking one extreme sample -- 1/(1-0.99)=100 for p99. +P99_MIN_N = 100 +CHART_MAX_BARS = 8 +HEADER = "## ⚡ SystemView performance — HIL" +LEGEND = ("<sub>**Legend** — **p50/p99**: median / 99th-percentile duration over all calls " + "in the capture window (µs; p50 = typical cost, p99 = tail latency). **Δ**: " + "change vs base branch; **−** is faster/better; **+∞%** marks a jump from an " + "(effectively) zero base -- not literally infinite, just too big to express as a " + "normal percentage; **gone** marks a row base had that PR does not (work that used to " + "run no longer does, or was renamed -- investigate before merging). **pt**: " + "percentage points. **CPU load, workload win**: context's busy share summed from the raw scheduling events over the workload window (the span of `tud_cdc_read` activity -- the throughput test's own footprint); falls back to SystemView's whole-recording share, labelled as such. " + "**stack high-water**: peak bytes of stack used. Function rows/bars are ordered by " + "CPU occupancy (calls × p50) in the PR capture, hottest first; base-only rows sort " + "after them. **– ⚠︎**: metric withheld — RTT ring overflowed (`overflow N`), no " + "timestamp frequency (`no time base`), or too few samples (`n<50` for p50; p99 needs " + "`n>100` for the tail estimate to be statistically meaningful, so it can show `n=<n>` " + "even on a row whose p50 is shown); withheld beats wrong. Durations read " + "**p50 / p99 / max**, all as captured — spliced CALL/RET pairs are discarded " + "upstream and counted in `dropped_pairs`. Capture: OpenOCD RTT " + "@1 ms poll — p50/p99 match the J-Link recorder within ~1%.</sub>") + +def load_set(d): + out = {} + for p in glob.glob(os.path.join(d, "sysview-*.json")): + try: + with open(p) as f: + j = json.load(f) + out[j["board"]] = j + except (OSError, ValueError, KeyError): + continue + return out + +def gate(side, name_metrics): + """None if usable, else the withhold reason string.""" + if side is None: + return "new" + if not side.get("metrics"): + # Malformed/older-schema capture JSON: no "metrics" key at all (missing, not just + # None) used to raise KeyError here on the very next line -- reachable even when the + # OTHER side is fine, e.g. via the chart's gate(base_j, None) -- and abort the whole + # report the same way F4's board_section bug did before it was guarded there. + return "failed" + if (side.get("capture") or {}).get("workload_ok") is False: + return "workload" + if side["metrics"].get("live_window_s") is None: + return "no time base" # F3: report.py couldn't establish a timestamp frequency + if side["metrics"].get("overflow", 0) > 0: + return f"overflow {side['metrics']['overflow']}" + if name_metrics is not None and name_metrics.get("n", 0) < GATE_MIN_N: + return f"n={name_metrics.get('n', 0)}" + return None + +def gate_p99(side, name_metrics): + """Like gate(), but withholds p99 whenever n<=P99_MIN_N even though p50 (GATE_MIN_N=50) is + shown (F1). N13: sysview_report.py's pct() now picks the conventional nearest-rank index + sorted_vals[int((n-1)*p/100)] (wave 3), which is already strictly below the max index for + every n>=2 -- so this is no longer about avoiding a collision with the max sample. It stays + a stricter bar than p50's because a 99th-percentile estimate needs roughly n>=1/(1-p)=100 + samples to mean anything -- below that it is effectively just picking one extreme sample and + calling it "p99". See P99_MIN_N's own comment.""" + reason = gate(side, name_metrics) + if reason is not None: + return reason + n = name_metrics.get("n", 0) if name_metrics else 0 + if n <= P99_MIN_N: + return f"n={n}" + return None + +def by_name(mlist): + return {m["name"]: m for m in (mlist or [])} + +def fmt_delta(b, p): + if b is None and p is None: + return "—" + if b is None: + return "new" + if p is None: + # N2: base had this row, PR does not -- a real disappearance, not "no change" (the + # bare "—" a normal near-zero delta would show). + return "**gone**" + if b == 0: + # N3: num_us() rounds to 1 decimal, so a base under 0.05 (µs, or any other unit this is + # used for) serialises as exactly 0 -- a plain ratio would divide by zero, and the old + # blanket "b == 0 -> '—'" hid a real jump (base 0.0, PR 5.0) behind a cell that reads as + # "no change". p == 0 too is a genuine no-op (both sides inactive); only a nonzero PR + # against a zero base is the regression worth flagging. + return "—" if p == 0 else "**+∞%**" + d = (p - b) / b * 100 + if abs(d) < 1.0: + return "—" + mark = " ✅" if d < 0 else "" + bold = ("**", "**") if abs(d) >= 5 else ("", "") + return f"{bold[0]}{d:+.1f}%{bold[1]}{mark}" + +def metric_cell(side_json, m, fmt): + reason = gate(side_json, m) + if reason in (None,): + return fmt(m), None + if reason == "new": + return None, "new" + if reason == "failed": + return None, "failed" + return f"– ⚠︎ {reason}", reason + +def row(label, bmap, pmap, name, base_j, pr_j, key="p50_us", unit=" µs"): + bm, pm = bmap.get(name), pmap.get(name) + if pm is None and bm is None: + return None + fmt = lambda m: f"{m[key]:.1f}{unit}" if m else "—" + bcell, bgate = metric_cell(base_j, bm, fmt) if bm else ("—", None) + pcell, pgate = metric_cell(pr_j, pm, fmt) if pm else ("—", None) + if bgate or pgate: + delta = "—" + bcell, pcell = bcell or "—", pcell or "—" + else: + delta = fmt_delta(bm and bm[key], pm and pm[key]) + return f"| {label} | {bcell} | {pcell} | {delta} |" + +def gated_pair(base_j, pr_j, b, p, fmt, delta_fn): + """Render (base_cell, pr_cell, delta) for a base/PR value pair that has no per-sample "n" + of its own (contexts' cpu_pct, stack's bytes_used) -- gated through gate() the same as every + duration metric (F2b), passing name_metrics=None so gate()'s n<GATE_MIN_N check no-ops. + Mirrors row()/metric_cell()'s Δ precedence: a gate reason on either side wins over "new", + which wins over an actual delta. + + p is None (N2): this name is base-only (union iteration in board_section() found no PR-side + entry at all) -- gate(pr_j, None) can't tell that apart from "PR side is fine, this metric + just wasn't asked for", so it must be special-cased here rather than left to fall through.""" + if p is None: + breason = gate(base_j, None) + if breason is None: + return fmt(b), "—", "**gone**" + return f"– ⚠︎ {breason}", "—", "—" # base itself unusable -- can't confirm "gone" either + preason = gate(pr_j, None) + breason = "new" if b is None else gate(base_j, None) + pcell = fmt(p) if preason is None else ("—" if preason == "new" else f"– ⚠︎ {preason}") + if breason is None: + bcell = fmt(b) + elif breason == "new": + bcell = "—" + else: + bcell = f"– ⚠︎ {breason}" + if preason or (breason not in (None, "new")): + delta = "—" + elif breason == "new": + delta = "new" + else: + delta = delta_fn(b, p) + return bcell, pcell, delta + +def board_section(name, base_j, pr_j): + live_window_s = (pr_j.get("capture") or {}).get("live_window_s") + heading = f"### {name}" + (f" — live {live_window_s:.1f} s" if live_window_s is not None else "") + lines = [heading, ""] + if pr_j.get("error"): + return "\n".join(lines + [f"capture failed: {pr_j['error']}", ""]) + pm_all = pr_j.get("metrics") or {} + if not pm_all: + # F4: a malformed/older-schema capture JSON (missing "metrics", no "error" set either) + # used to raise KeyError here and abort the WHOLE report -- no comment for any board. + return "\n".join(lines + ["capture failed: missing metrics", ""]) + bm_all = (base_j or {}).get("metrics") or {} + lines += ["| metric | base | PR | Δ |", "|---|---:|---:|---:|"] + bisr, pisr = by_name(bm_all.get("isr")), by_name(pm_all.get("isr")) + # N2: union of both sides, not just pisr -- an ISR/function/context/stack row present only + # in base (work that used to run and no longer does) must still render, not silently vanish + # because nothing ever iterates bisr/bfn/bctx/bst on their own. sorted() for a deterministic + # order (by_name()'s dicts are already insertion-ordered from an alphabetically-sorted JSON + # list, so this is a no-op for the common both-sides-present case). + for iname in sorted(set(pisr) | set(bisr)): + pm, bm = pisr.get(iname), bisr.get(iname) + # F1: p50 and p99 are gated independently -- p99 needs n>P99_MIN_N (see its comment, + # N13) for the tail estimate to be meaningful, a stricter bar than p50's GATE_MIN_N, so + # p99 can be withheld on a row whose p50 still shows. "gone"/"new" (pm/bm respectively + # absent) are per-half reasons here too, same precedence as an ordinary gate() withhold. + preason50 = "gone" if pm is None else gate(pr_j, pm) + preason99 = "gone" if pm is None else gate_p99(pr_j, pm) + breason50 = "new" if bm is None else gate(base_j, bm) + breason99 = "new" if bm is None else gate_p99(base_j, bm) + def _half(m, reason, key): + if reason is None: + return f"{m[key]:.1f}" + return "—" if reason in ("new", "gone") else f"– ⚠︎ {reason}" + def _cell(m, r50, r99): + # max is shown, not hidden: it is a real measured duration. The exporter's spliced + # CALL/RET pairs (the reason it used to be suppressed) are discarded in + # sysview_report.py and counted as dropped_pairs, so what survives is measured. + return (f"{_half(m, r50, 'p50_us')} / {_half(m, r99, 'p99_us')} / " + f"{_half(m, r50, 'max_us')} µs") + def _one_delta(preason, breason, bm_, pm_, key): + if preason == "gone" and breason is None: + return "**gone**" + if preason or (breason not in (None, "new")): + return "—" + if breason == "new": + return "new" + return fmt_delta(bm_[key], pm_[key]) + def _delta(): + d50 = _one_delta(preason50, breason50, bm, pm, 'p50_us') + d99 = _one_delta(preason99, breason99, bm, pm, 'p99_us') + return f"{d50} / {d99}" + lines.append(f"| {iname} p50 / p99 | {_cell(bm, breason50, breason99)} | " + f"{_cell(pm, preason50, preason99)} | {_delta()} |") + bfn, pfn = by_name(bm_all.get("functions")), by_name(pm_all.get("functions")) + # PR-side rows keep their existing CPU-occupancy order; base-only rows (N2) have no PR + # occupancy to rank by, so they sort alphabetically after every PR-ranked row. + order = sorted(pfn, key=lambda k: pfn[k]["n"] * pfn[k]["p50_us"], reverse=True) + order += sorted(set(bfn) - set(pfn)) + for fname in order: + r = row(f"`{fname}` p50", bfn, pfn, fname, base_j, pr_j) + if r: lines.append(r) + # CPU load rows prefer cpu_pct_workload -- busy time summed from the raw scheduling + # events over the workload window (span of tud_cdc_read activity), the number that + # answers "load while the throughput test ran". Falls back to SystemView's + # whole-recording cpu_pct, labelled as such, when no workload window was found. + bwl, pwl = by_name(bm_all.get("contexts_workload")), by_name(pm_all.get("contexts_workload")) + use_wl = bool(pwl) + bctx, pctx = (bwl, pwl) if use_wl else \ + (by_name(bm_all.get("contexts")), by_name(pm_all.get("contexts"))) + key = "cpu_pct_workload" if use_wl else "cpu_pct" + label = "CPU load, workload win" if use_wl else "CPU load, whole recording" + for cname in sorted(set(pctx) | set(bctx)): + if cname.lower() in ("usbd", "usbh", "isr") or cname.lower().startswith("isr "): + b, c = bctx.get(cname), pctx.get(cname) + bcell, pcell, d = gated_pair( + base_j, pr_j, b, c, lambda m: f"{m[key]:.1f} %", + lambda b_, p_: (f"{p_[key]-b_[key]:+.1f} pt" + if abs(p_[key] - b_[key]) >= 0.1 else "—")) + lines.append(f"| {label} ({cname}) | {bcell} | {pcell} | {d} |") + bst, pst = by_name(bm_all.get("stack")), by_name(pm_all.get("stack")) + for sname in sorted(set(pst) | set(bst)): + b, sm = bst.get(sname), pst.get(sname) + bcell, pcell, d = gated_pair( + base_j, pr_j, b, sm, lambda m: f"{m['bytes_used']} B", + lambda b_, p_: ("—" if b_['bytes_used'] == p_['bytes_used'] else + f"{p_['bytes_used']-b_['bytes_used']:+d} B")) + lines.append(f"| `{sname}` stack high-water | {bcell} | {pcell} | {d} |") + # chart: occupancy order, PR-gate-eligible only, capped; base series only if + # every charted function is ungated on the base side too (no holes in a mermaid series). + # `c in pfn` excludes base-only rows (N2): the chart is explicitly "hot functions ... in the + # PR capture" (see LEGEND), and a base-only row has no PR bar to plot. + eligible = [c for c in order if c in pfn and gate(pr_j, pfn[c]) is None][:CHART_MAX_BARS] + if eligible: + base_ok = (gate(base_j, None) is None and + all(c in bfn and gate(base_j, bfn[c]) is None for c in eligible)) + short = [re.sub(r'^(tud_|tuh_|dcd_|hcd_)', '', c) for c in eligible] + pv = [f"{pfn[c]['p50_us']:.1f}" for c in eligible] + bars = [f" bar [{', '.join(pv)}]"] + values = list(pv) + if base_ok: + bv = [f"{bfn[c]['p50_us']:.1f}" for c in eligible] + bars = [f" bar [{', '.join(bv)}]"] + bars + values += bv + peak = max(float(v) for v in values) + # N7: every charted p50 rounding to 0.0 gives mermaid a "0 --> 0" axis, which fails to + # render and leaves a raw, broken code fence in the PR comment -- skip the chart + # entirely rather than publish that (the table above already shows every value). + if peak > 0: + ymax = max(peak * 1.3, 0.1) + lines += ["", "```mermaid", "xychart-beta", + ' title "hot functions p50 µs (base vs PR)"', + f" x-axis [{', '.join(short)}]", + f' y-axis "µs" 0 --> {ymax:.1f}'] + bars + ["```"] + return "\n".join(lines) + "\n" + +def report(base_dir, pr_dir): + pr = load_set(pr_dir) + if not pr: + return "" + base = load_set(base_dir) + if base and not (set(base) & set(pr)): + return "" + boards = sorted(pr) + any_pr = next(iter(pr.values())) + # .get throughout: one malformed capture JSON must not abort the whole report -- the same + # rule gate()/board_section() already follow one layer down. + base_commit = next(iter(base.values())).get("commit", "?") if base else "(none)" + head = (f"*{any_pr.get('example', '?')} `SYSVIEW=4`, workload `{any_pr.get('workload', '?')}` " + f"{any_pr.get('duration_s', '?')} s, OpenOCD rtt @1 ms · " + f"base `{base_commit}` → PR `{any_pr.get('commit', '?')}`*") + parts = [HEADER, "", head, ""] + for b in boards: + parts.append(board_section(b, base.get(b), pr[b])) + parts.append(LEGEND) + return "\n".join(parts) + "\n" + +# ---------------------------------------------------------------- capture +def select_boards(cfg, board_args): + picked = [b for b in cfg["boards"] if "sysview" in b] + if board_args: + picked = [b for b in picked if b["name"] in set(board_args)] + return picked + +def capture_ocd_args(board): + # Only the plain "openocd" flasher name exists since upstream #3804 folded + # openocd_wch away (openocd_adi likewise); the rig binary is the unified fork. + args = board["sysview"].get("ocd_args") or ( + board["flasher"]["args"] if board["flasher"]["name"] == "openocd" else None) + if not args: + raise ValueError(f"{board.get('name')}: non-openocd flasher and no sysview.ocd_args") + return [a.strip('"') for a in re.findall(r'"[^"]*"|\S+', args)] + +def is_wch_board(board): + """WCH parts are detected by their target config -- the only spelling left after + #3804 folded the openocd_wch flasher name into openocd.""" + return "wch-riscv" in (board["flasher"].get("args") or "") + +def session_resets(board): + """Whether the capture session issues its own `reset run`. Two board classes must + attach without one: WCH -- under SDI the target never comes back and USB never + re-enumerates -- and boards flagged sysview.attach_only (metro_m4_express: after + openocd's reset on atsame5x under an attached debugger the core stays held, the + SAMD5x DSU reset extension is the prime suspect, so the app never reboots and the + RTT control block never appears; measured 2026-08-11: attach-without-reset streams + immediately while both reset-run captures died with 'No control block found'). + Either way the flasher's own post-flash reset already supplied the fresh boot.""" + return not (is_wch_board(board) or board["sysview"].get("attach_only")) + +def board_result(board, commit, metrics=None, capture_info=None, error=None): + sv = board["sysview"] + return {"board": board["name"], "commit": commit, + "example": sv["example"], "workload": sv["workload"], + "duration_s": sv["duration_s"], + "capture": capture_info or {}, "metrics": metrics, "error": error} + +def _workload_cdc_burst(node, duration_s): + """Returns True if traffic ran for the full window, False if the serial + link died early (device dropped off the bus) -- never raises.""" + import serial, time + # write_timeout: without it, a device that stops draining its OUT endpoint blocks s.write() + # forever, wedging PHASE 2 while the board flock is held -- the same failure mode hil_test.py + # guards against in open_serial_dev() (SERIAL_WRITE_TIMEOUT, default 10s, sized for + # whole-firmware-image transfers there). Hardcoded here rather than imported from hil_test: + # this workload only ever writes 64-byte micro-bursts, so a much shorter deadline catches a + # wedge faster, and importing hil_test would pull in its pymtp dependency for one constant. + s = serial.Serial(node, 115200, timeout=0.02, write_timeout=2) + end = time.monotonic() + duration_s + while time.monotonic() < end: + t = time.monotonic() + while time.monotonic() - t < 0.30 and time.monotonic() < end: + try: + s.write(b"x" * 64); s.read(64) + except Exception: + s.close() + return False + time.sleep(min(1.0, max(0, end - time.monotonic()))) + s.close() + return True + +WORKLOADS = {"cdc_burst": _workload_cdc_burst, + "idle": lambda node, duration_s: __import__("time").sleep(duration_s) or True} + +def _drain_stderr(pipe, buf): + """Continuously read a subprocess's stderr into a bounded deque, run on a daemon + thread. OpenOCD logs for the whole capture window; with stderr=PIPE and nothing + reading it, the pipe's OS buffer fills and OpenOCD blocks writing to it, wedging + the capture. Only the last ~200 lines are kept, for error reporting.""" + for line in iter(pipe.readline, ""): + buf.append(line) + pipe.close() + +def _err_excerpt(text, head=500, tail=500): + """Excerpt a long error from BOTH ends. cmake prints the diagnosis first and the + call stack last, ninja the reverse -- keeping only one end drops the useful half + (a tail-only slice reported 'eCache.txt' plus a bare call stack for a missing + hw/mcu dependency).""" + t = (text or "").strip() + if len(t) <= head + tail: + return t + return f"{t[:head]}\n[...{len(t) - head - tail} chars omitted...]\n{t[-tail:]}" + +def build_one(board, repo_root): + """PHASE 1 (no board lock): cmake configure + build. Returns (fw_path, error) -- + fw_path INCLUDES the flasher's extension. Never raises.""" + import subprocess, hil_flash + sv, name = board["sysview"], board["name"] + try: + exdir = os.path.join(repo_root, "examples", sv["example"]) + bdir = os.path.join(repo_root, "examples", f"cmake-build-sysview-{name}") + buf = [f"-DSYSVIEW_BUFFER_SIZE={sv['buffer']}"] if "buffer" in sv else [] + for cmd in ([ "cmake", "-B", bdir, f"-DBOARD={name}", "-G", "Ninja", + "-DCMAKE_BUILD_TYPE=MinSizeRel", "-DSYSVIEW=4", *buf, "." ], + [ "cmake", "--build", bdir ]): + r = subprocess.run(cmd, cwd=exdir, capture_output=True, text=True, timeout=1800) + if r.returncode: + return None, f"build failed: {_err_excerpt(r.stderr)}" + # Both JLinkExe's `loadfile` and OpenOCD's `program` infer the image format from + # the extension, so the flasher needs the real filename: an extension-less path + # dies with "File is of unknown / unsupported format" after taking the board lock. + base = os.path.basename(sv["example"]) + ext = hil_flash.FLASHER_SUFFIX.get(board["flasher"]["name"].lower(), ".elf") + fw = os.path.join(bdir, base + ext) + if not os.path.exists(fw): + return None, f"build produced no {ext}: {fw}" + return fw, None + except Exception as e: + return None, f"{type(e).__name__}: {e}" + +def flash_and_capture_one(board, fw, out_dir, repo_root): + """PHASE 2 (under the board lock): flash, wait for enumeration, RTT-capture raw + SVDat bytes while the workload runs, then always reflash pristine firmware before + returning. Returns (capture_dict, error); never raises.""" + import subprocess, time, socket, signal, threading + from collections import deque + sv, name = board["sysview"], board["name"] + try: + scripts = os.path.join(repo_root, ".claude", "skills", "sysview", "scripts") + sys.path.insert(0, scripts) + from sysview_record import rtt_cb_from_elf + import hil_flash + from helper import hil_util + ocd_args = capture_ocd_args(board) + flash = getattr(hil_flash, f"flash_{board['flasher']['name']}") + r = flash(board, fw) + if r.returncode: + # run_cmd merges stderr into stdout and fills it even on timeout (rc=124); + # without the flasher's own output 'rc=124' cannot distinguish a wedged + # probe from a dead target. + out = _err_excerpt(hil_util.cmd_stdout_text(r.stdout)) + return None, f"flash failed: rc={r.returncode}" + (f"\n{out}" if out else "") + node = hil_util.get_serial_dev(board['uid'], 'TinyUSB', 'TinyUSB_Device', 0) + try: + # fw carries the flasher's extension, which is not always .elf (lm4flash and + # esptool take .bin); the RTT control block always comes from the ELF's symbols. + cb = int(rtt_cb_from_elf(os.path.splitext(fw)[0] + ".elf"), 16) + except SystemExit as e: + # rtt_cb_from_elf() sys.exit()s on a bad/non-SYSVIEW ELF -- SystemExit derives from + # BaseException, so it would otherwise sail past the `except Exception` below and + # abort the whole multi-board capture loop in main() before boards already captured + # got a chance to be written out. Convert it to this board's own error instead. + return None, f"no RTT symbol: {e}" + resets = session_resets(board) # False: WCH (SDI) and attach_only boards + # Ask the kernel for a free port instead of a fixed one: a leftover openocd + # still holding the port turns every later capture into a misleading + # "rtt server never listened". + with socket.socket() as s: + s.bind(("localhost", 0)) + port = s.getsockname()[1] + ocd = ["openocd", "-c", "tcl_port disabled", "-c", "gdb_port disabled", + "-c", "telnet_port disabled", + "-c", f"adapter serial {board['flasher']['uid']}"] + ocd_args + \ + ["-c", "init"] + (["-c", "reset run", "-c", "sleep 2000"] if resets else []) + \ + ["-c", f'rtt setup {cb} 0x1000 "SEGGER RTT"', + "-c", "rtt polling_interval 1", "-c", "rtt start", + "-c", f"rtt server start {port} 1"] + raw = os.path.join(out_dir, f"{name}-capture.SVDat") + p = subprocess.Popen(ocd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True) + stderr_tail = deque(maxlen=200) + threading.Thread(target=_drain_stderr, args=(p.stderr, stderr_tail), daemon=True).start() + workload_ok, wl_err, wl_start = True, None, None + try: + # Readiness comes from OpenOCD's own log line, never from a throwaway probe + # connection: a connection registers an RTT sink, and its data is only read while + # a sink exists -- if the 1 ms poll fires between accept and seeing the probe's + # close, channel 1's one-shot Init record (sync + timestamp frequency) is drained + # into the discarded socket and the recording that follows decodes without a time + # base. The line also proves the listener on `port` is THIS openocd, not another + # process that won the bind-0/close/rebind race above. + listening = f"Listening on port {port} for rtt connections" + for _ in range(80): + time.sleep(0.2) + if p.poll() is not None: + return None, f"openocd: {''.join(stderr_tail)[-200:]}" + if any(listening in line for line in list(stderr_tail)): + break + else: + return None, "rtt server never listened" + # Wait for enumeration AFTER the capture session's own "reset run" above (baked + # into `ocd` when session_resets(), i.e. neither WCH nor attach_only) has actually + # happened, not before it: waiting earlier (right after flash()) validated a boot + # this reset immediately throws away, and the workload below needs THIS boot's + # node, which can take real time to reappear. For no-reset boards the flash's own + # reset supplied the boot, so the node is usually already present here. + for _ in range(25): + if os.path.exists(node): break + time.sleep(1) + else: + return None, "device never enumerated after flash" + with open(raw, "wb") as f: + nc = subprocess.Popen(["nc", "localhost", str(port)], stdout=f) + wl_start = time.monotonic() + try: + try: + workload_ok = bool(WORKLOADS[sv["workload"]](node, sv["duration_s"])) + except Exception as e: + workload_ok, wl_err = False, e + finally: + nc.send_signal(signal.SIGINT) + try: nc.wait(5) + except subprocess.TimeoutExpired: nc.kill() + finally: + p.send_signal(signal.SIGINT) + try: p.wait(8) + except subprocess.TimeoutExpired: p.kill() + # A serial death in the first half of the window means the whole capture is + # mostly idle bus -- fail it outright. A late death still leaves a mostly-live + # capture; keep it, but flag workload_ok=false so report() gates the metrics. + if not workload_ok and wl_start is not None and \ + (time.monotonic() - wl_start) < sv["duration_s"] / 2: + return None, f"workload died: {wl_err or 'stopped early'}" + return {"raw": raw, "workload_ok": workload_ok}, None + except Exception as e: + return None, f"{type(e).__name__}: {e}" + finally: + _reflash_pristine(board, repo_root) + +def decode_one(board, commit, out_dir, capture, repo_root): + """PHASE 3 (no board lock): decode the raw SVDat capture + report. Never raises.""" + import subprocess + sv, name = board["sysview"], board["name"] + scripts = os.path.join(repo_root, ".claude", "skills", "sysview", "scripts") + try: + dec = os.path.join(out_dir, f"{name}-decoded") + r = subprocess.run([sys.executable, os.path.join(scripts, "sysview_record.py"), + "--from-raw", capture["raw"], "--out", dec], + capture_output=True, text=True, timeout=900) + if r.returncode: + return board_result(board, commit, error=f"record decode failed: {r.stderr[-200:]}") + r = subprocess.run([sys.executable, os.path.join(scripts, "sysview_report.py"), + dec, "--json"], capture_output=True, text=True, timeout=600) + if r.returncode: + return board_result(board, commit, error=f"decode failed: {r.stderr[-200:]}") + m = json.loads(r.stdout) + info = {"route": "openocd-rtt", "poll_ms": 1, "live_window_s": m.get("live_window_s"), + "workload_ok": capture["workload_ok"]} + return board_result(board, commit, metrics=m, capture_info=info) + except Exception as e: + return board_result(board, commit, error=f"{type(e).__name__}: {e}") + +def _reflash_pristine(board, repo_root): + import hil_flash, subprocess + name = board["name"] + try: + # Reflash device/board_test, not the sysview example: it's the same park image + # hil_test.py flashes as its own end-of-board teardown (disables the board's USB) -- + # leaving the SYSVIEW-instrumented example running instead would mean a board the + # rig's contract calls "parked" is actually still enumerating a live CDC+MSC device. + # find_firmware's default root only covers ONE park-image layout at a time (its + # `build_dir` module global -- 'cmake-build' unless a caller like hil_test.main sets it + # from argparse; nothing sets it for us). The rig mirror builds park images under + # examples/cmake-build-<board>/, but the CI runner instead uses tools/build.py's + # cmake-build/cmake-build-<board>/ layout -- mutating the global to "examples" alone (as + # this used to do) finds the rig's images but makes the lookup silently miss, and the + # restore silently no-op, on the CI runner. Pass both roots explicitly instead. + fw = hil_flash.find_firmware(name, "device/board_test", roots=["examples", "cmake-build"]) + if not fw: + print(f"warning: {name}: no park (device/board_test) firmware found for " + f"post-capture reflash (board left running the SYSVIEW-instrumented build)", + file=sys.stderr) + return + r = getattr(hil_flash, f"flash_{board['flasher']['name']}")(board, str(fw)) + if r.returncode: + print(f"warning: {name}: pristine reflash failed rc={r.returncode}", + file=sys.stderr) + except Exception as e: + # pristine reflash is best-effort -- never raise out of here -- but a + # silent swallow left a board's post-capture firmware state unknowable + # from the CI log, so at least report what happened. + print(f"warning: {name}: pristine reflash raised {type(e).__name__}: {e}", + file=sys.stderr) + +def _git_commit(repo_root): + import subprocess + try: + r = subprocess.run(["git", "rev-parse", "--short", "HEAD"], cwd=repo_root, + capture_output=True, text=True, timeout=10) + return r.stdout.strip() if r.returncode == 0 else "unknown" + except Exception: + return "unknown" + +def main(): + ap = argparse.ArgumentParser() + sub = ap.add_subparsers(dest="cmd", required=True) + rp = sub.add_parser("report") + rp.add_argument("base_dir"); rp.add_argument("pr_dir") + rp.add_argument("-o", "--out", default="sysview_report.md") + cp = sub.add_parser("capture") + cp.add_argument("config") + cp.add_argument("-b", "--board", action="append", default=[], + help="Boards to capture, all sysview-flagged boards if not specified") + cp.add_argument("--out", default="sysview-out") + args = ap.parse_args() + if args.cmd == "report": + md = report(args.base_dir, args.pr_dir) + with open(args.out, "w") as f: + f.write(md) + print(f"{'empty (no captures)' if not md else args.out}") + elif args.cmd == "capture": + from helper import hil_lock + try: + with open(args.config) as f: + cfg = json.load(f) + except (OSError, ValueError) as e: + print(f"error: cannot load {args.config}: {e}", file=sys.stderr) + sys.exit(1) + boards = select_boards(cfg, args.board) + if not boards: + # Neither case is an error: a rig config with no sysview-flagged boards at all + # (e.g. hfp.json) and a -b filter that matched none of this rig's sysview boards + # (e.g. a board from the CI matrix that isn't on this particular rig) are both + # "nothing to capture here", not a failure. + if not any("sysview" in b for b in cfg["boards"]): + print(f"no sysview-flagged boards in {args.config}, nothing to capture") + else: + print("no sysview board selected, nothing to capture") + sys.exit(0) + os.makedirs(args.out, exist_ok=True) + repo_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + commit = _git_commit(repo_root) + # 3 phases so the board lock -- which blocks concurrent CI on that board -- is only + # held for the part that actually touches hardware (PHASE 2). Building and decoding + # (PHASES 1/3) are lock-free and, for builds, could run concurrently across boards; + # kept sequential here for simplicity, same as before this split. + # PHASE 1: build every selected board, no lock held. + builds = {board["name"]: build_one(board, repo_root) for board in boards} + # PHASE 2: per board, flash + RTT-capture under that board's lock only. + captures = {} + for board in boards: + name = board["name"] + fw, err = builds[name] + if err: + captures[name] = (None, err) + continue + try: + lock_fh = hil_lock.acquire_board_lock(name, reason="sysview capture") + except RuntimeError as e: + captures[name] = (None, f"board locked: {e}") + continue + try: + captures[name] = flash_and_capture_one(board, fw, args.out, repo_root) + finally: + # mirror hil_test.py's test_board(): clear our pid record before + # dropping the flock so a freed board never reads as still-locked + if lock_fh: + try: + lock_fh.truncate(0) + except OSError: + pass + lock_fh.close() + # PHASE 3: decode + report every board with a successful capture, no lock held. + for board in boards: + name = board["name"] + capture, err = captures[name] + result = board_result(board, commit, error=err) if err else \ + decode_one(board, commit, args.out, capture, repo_root) + with open(os.path.join(args.out, f"sysview-{name}.json"), "w") as f: + json.dump(result, f) + print(f"{name}: {'ok' if not result['error'] else result['error']}") + sys.exit(0) + +if __name__ == "__main__": + main() diff --git a/test/hil/test/test_ci_select.py b/test/hil/test/test_ci_select.py index 22fbde17b..802802896 100644 --- a/test/hil/test/test_ci_select.py +++ b/test/hil/test/test_ci_select.py @@ -993,6 +993,9 @@ class TestTheHarnessTestsAreNotTheHarness(unittest.TestCase): 'test/hil/test/test_hil_report.py', 'test/hil/test/test_hil_rtt.py', 'test/hil/test/test_hil_util.py', + 'test/hil/test/test_sysview_ci.py', + 'test/hil/test/test_sysview_record.py', + 'test/hil/test/test_sysview_report.py', ], 'test/hil/test/ gained or lost a file; it is carved out of rule 2, so confirm ' 'the rig still does not read anything in there before updating this list') diff --git a/test/hil/test/test_sysview_ci.py b/test/hil/test/test_sysview_ci.py new file mode 100644 index 000000000..30b85f439 --- /dev/null +++ b/test/hil/test/test_sysview_ci.py @@ -0,0 +1,448 @@ +#!/usr/bin/env python3 +import json, os, shutil, subprocess, sys, tempfile, unittest, unittest.mock +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import sysview_ci + +def metrics(overflow=0, funcs=None, isr=None, contexts=None, stack=None, live_window_s=14.2): + return {"contexts": contexts or [], "isr": isr or [], "ready_run": [], + "functions": funcs or [], "markers": [], "stack": stack or [], + "heap": None, "overflow": overflow, "overflow_total": overflow, + "dropped_pairs": 0, "live_window_s": live_window_s, + "warnings": []} + +def board_json(board="stm32f407disco", err=None, workload_ok=None, **mk): + cap = {"route": "openocd-rtt", "poll_ms": 1, "live_window_s": 14.2} + if workload_ok is not None: + cap["workload_ok"] = workload_ok + return {"board": board, "commit": "abc1234", "example": "device/cdc_msc", + "workload": "cdc_burst", "duration_s": 15, + "capture": cap, + "metrics": None if err else metrics(**mk), "error": err} + +def write_set(d, *objs): + os.makedirs(d, exist_ok=True) + for o in objs: + with open(os.path.join(d, f"sysview-{o['board']}.json"), 'w') as f: + json.dump(o, f) + +F = lambda name, n, p50: {"name": name, "n": n, "p50_us": p50, "p99_us": p50*2, "max_us": p50*99} +I = lambda name, n, p50: {"name": name, "n": n, "p50_us": p50, "p99_us": p50+3, "max_us": p50*9} + +class Report(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) + self.base, self.pr = os.path.join(self.tmp,'b'), os.path.join(self.tmp,'p') + + def go(self, base_objs, pr_objs): + write_set(self.base, *base_objs); write_set(self.pr, *pr_objs) + return sysview_ci.report(self.base, self.pr) + + def test_empty_intersection_returns_empty(self): + self.assertEqual(self.go([board_json(board="a")], [board_json(board="b")]), "") + self.assertEqual(sysview_ci.report(self.base, os.path.join(self.tmp,'nope')), "") + + def test_delta_and_occupancy_order(self): + b = board_json(funcs=[F("dcd_edpt_xfer",200,5.9), F("tud_task",22000,7.4)], + isr=[I("ISR 83",24000,6.7)]) + p = board_json(funcs=[F("dcd_edpt_xfer",200,6.1), F("tud_task",22000,7.4)], + isr=[I("ISR 83",24000,6.8)]) + md = self.go([b],[p]) + self.assertIn("## ⚡ SystemView performance — HIL", md) + self.assertIn("+3.4%", md) # dcd 5.9 -> 6.1 + self.assertLess(md.index("tud_task"), md.index("dcd_edpt_xfer")) # occupancy order + self.assertIn("```mermaid", md) + self.assertEqual(md.count("Legend"), 1) + self.assertNotIn("max", md.split("Legend")[0]) # max never rendered + + def test_overflow_gates_all_durations(self): + p = board_json(overflow=3, funcs=[F("tud_task",22000,7.4)]) + md = self.go([board_json(funcs=[F("tud_task",22000,7.4)])],[p]) + self.assertIn("⚠︎ overflow 3", md) + self.assertNotIn("+", md.split("tud_task")[1].split("\n")[0]) # no delta on gated row + + def test_low_n_gates_metric(self): + p = board_json(funcs=[F("mscd_xfer_cb",9,13.0)]) + md = self.go([board_json(funcs=[F("mscd_xfer_cb",9,13.0)])],[p]) + self.assertIn("⚠︎ n=9", md) + + def test_workload_died_gates_all_metrics(self): + p = board_json(workload_ok=False, funcs=[F("tud_task",22000,7.4)], + isr=[I("ISR 83",24000,6.7)]) + md = self.go([board_json(funcs=[F("tud_task",22000,7.4)], isr=[I("ISR 83",24000,6.7)])],[p]) + self.assertIn("⚠︎ workload", md) + self.assertNotIn("+", md.split("tud_task")[1].split("\n")[0]) # no delta on gated row + + def test_capture_failed_board(self): + md = self.go([board_json()],[board_json(err="flash failed: rc=1")]) + self.assertIn("capture failed: flash failed: rc=1", md) + + def test_missing_metrics_key_renders_capture_failed(self): + # F4: an older/malformed capture JSON with no "metrics" key at all (and no "error" + # set either) used to raise KeyError in board_section() and abort the WHOLE report -- + # no comment posted for ANY board, not just this one. + p = {"board": "stm32f407disco", "commit": "abc1234", "example": "device/cdc_msc", + "workload": "cdc_burst", "duration_s": 15, "capture": {}, "error": None} + md = self.go([board_json()], [p]) + self.assertIn("capture failed", md) + + def test_missing_base_metrics_key_does_not_crash_chart(self): + # Round-2 residual: gate()'s own `side["metrics"]` bracket access raised KeyError for a + # malformed BASE-side JSON with no "metrics" key and no "error" -- every OTHER gate() + # call site happens to short-circuit around it when the matching baseline entry is + # simply absent (bm/b is None), but the chart's `gate(base_j, None)` in base_ok calls + # straight into gate() with no such guard, so THIS was the one path that still crashed. + b = {"board": "stm32f407disco", "commit": "abc1234", "example": "device/cdc_msc", + "workload": "cdc_burst", "duration_s": 15, "capture": {}, "error": None} + p = board_json(funcs=[F("tud_task", 22000, 7.4)]) + md = self.go([b], [p]) # must not raise + self.assertIn("```mermaid", md) # PR-only chart still renders + self.assertEqual(md.count("bar ["), 1) # single series: base is gated "failed" + + def test_no_time_base_gates_all_metrics(self): + p = board_json(live_window_s=None, funcs=[F("tud_task",22000,7.4)], + isr=[I("ISR 83",24000,6.7)]) + md = self.go([board_json(funcs=[F("tud_task",22000,7.4)], isr=[I("ISR 83",24000,6.7)])],[p]) + self.assertIn("⚠︎ no time base", md) + self.assertNotIn("+", md.split("tud_task")[1].split("\n")[0]) # no delta on gated row + + def test_p99_withheld_at_low_n_p50_shown(self): + # F1: pct()'s p99 index lands on (or next to) the max sample at n<=100, so it must be + # withheld even though p50 (GATE_MIN_N=50) is well-supported and stays visible. + p = board_json(isr=[I("ISR 83", 60, 6.7)]) + md = self.go([board_json(isr=[I("ISR 83", 60, 6.7)])], [p]) + self.assertIn("6.7 / – ⚠︎ n=60", md) + + def test_p99_shown_at_high_n(self): + p = board_json(isr=[I("ISR 83", 500, 6.7)]) + md = self.go([board_json(isr=[I("ISR 83", 500, 6.7)])], [p]) + self.assertIn("6.7 / 9.7", md) # I() sets p99_us = p50_us + 3 + + def test_cpu_and_stack_rows_gated_on_overflow(self): + # F2b: CPU-load/stack rows used to bypass gate() entirely, publishing numbers on a + # board whose every other metric was voided by overflow. + p = board_json(overflow=3, contexts=[{"name": "usbd", "cpu_pct": 7.0}], + stack=[{"name": "cdc", "bytes_used": 300}]) + b = board_json(contexts=[{"name": "usbd", "cpu_pct": 6.5}], + stack=[{"name": "cdc", "bytes_used": 290}]) + md = self.go([b], [p]) + cpu_line = next(l for l in md.splitlines() if l.startswith("| CPU load")) + stack_line = next(l for l in md.splitlines() if "stack high-water" in l) + self.assertIn("⚠︎ overflow 3", cpu_line) + self.assertIn("⚠︎ overflow 3", stack_line) + self.assertIn("6.5 %", cpu_line) # base side (ungated) still shows its real value + self.assertIn("290", stack_line) + + def test_missing_baseline_absolute(self): + md = sysview_ci.report(os.path.join(self.tmp,'nobase'), self.pr) or \ + self.go([], [board_json(funcs=[F("tud_task",22000,7.4)])]) + self.assertIn("new", md) # Δ column shows new + self.assertIn("7.4", md) + + def test_chart_capped_at_8(self): + funcs=[F(f"fn{i}", 1000-i, 5.0+i) for i in range(10)] + md = self.go([board_json(funcs=funcs)],[board_json(funcs=funcs)]) + chart = md.split("```mermaid")[1].split("```")[0] + self.assertLessEqual(len(chart.split("x-axis")[1].split("]")[0].split(",")), 8) + + def test_chart_excludes_low_n_pr_function(self): + funcs = [F("tud_task",22000,7.4), F("mscd_xfer_cb",9,13.0)] + md = self.go([board_json(funcs=funcs)],[board_json(funcs=funcs)]) + chart = md.split("```mermaid")[1].split("```")[0] + xaxis = chart.split("x-axis")[1].split("]")[0] + self.assertIn("task", xaxis) + self.assertNotIn("xfer_cb", xaxis) # n<50 -> excluded from chart + self.assertIn("⚠︎ n=9", md) # table still shows the gated row + + def test_chart_single_series_on_base_overflow(self): + b = board_json(overflow=2, funcs=[F("tud_task",22000,7.4)]) + p = board_json(funcs=[F("tud_task",22000,7.4)]) + md = self.go([b],[p]) + chart = md.split("```mermaid")[1].split("```")[0] + self.assertEqual(chart.count("bar ["), 1) # single PR-only series + table = md.split("```mermaid")[0] + self.assertIn("⚠︎ overflow 2", table) # table still shows base gate cell + + def test_cpu_load_new_when_context_missing_from_present_baseline(self): + b = board_json(funcs=[F("tud_task",22000,7.4)]) # baseline exists, no contexts + p = board_json(funcs=[F("tud_task",22000,7.4)], + contexts=[{"name": "usbd", "cpu_pct": 7.0}]) + md = self.go([b],[p]) + self.assertIn("| CPU load, whole recording (usbd) | — | 7.0 % | new |", md) + + def test_heading_shows_live_window(self): + md = self.go([board_json()], [board_json()]) + self.assertIn("### stm32f407disco — live 14.2 s", md) + + # ---- N2: base-only rows (present in base, absent from PR) must render, not vanish ---- + + def test_base_only_function_row_flagged_gone(self): + b = board_json(funcs=[F("tud_task",22000,7.4), F("old_fn",22000,3.0)]) + p = board_json(funcs=[F("tud_task",22000,7.4)]) # old_fn dropped + md = self.go([b],[p]) + self.assertIn("| `old_fn` p50 | 3.0 µs | — | **gone** |", md) + + def test_base_only_isr_row_flagged_gone(self): + b = board_json(isr=[I("ISR 83", 24000, 6.7)]) + p = board_json() # ISR 83 dropped + md = self.go([b],[p]) + line = next(l for l in md.splitlines() if l.startswith("| ISR 83")) + self.assertIn("6.7 / 9.7 / 60.3 µs", line) # base half still shown (p50/p99/max) + self.assertIn("**gone** / **gone**", line) # delta flags disappearance + + def test_base_only_context_row_flagged_gone(self): + b = board_json(contexts=[{"name": "usbd", "cpu_pct": 6.5}]) + p = board_json() # usbd context dropped + md = self.go([b],[p]) + self.assertIn("| CPU load, whole recording (usbd) | 6.5 % | — | **gone** |", md) + + def test_base_only_stack_row_flagged_gone(self): + b = board_json(stack=[{"name": "cdc", "bytes_used": 300}]) + p = board_json() # cdc stack row dropped + md = self.go([b],[p]) + self.assertIn("| `cdc` stack high-water | 300 B | — | **gone** |", md) + + def test_gone_row_not_flagged_when_base_itself_gated(self): + # can't confidently call a row "gone" if the base side that would prove it existed is + # itself unusable (overflow here) -- must fall back to the ordinary withheld cell. + b = board_json(overflow=3, funcs=[F("old_fn",22000,3.0)]) + p = board_json() + md = self.go([b],[p]) + line = next(l for l in md.splitlines() if "old_fn" in l) + self.assertNotIn("gone", line) + self.assertIn("⚠︎ overflow 3", line) + + # ---- N3: a zero (or rounds-to-zero) base must not hide a real jump behind "—" ---- + + def test_zero_base_nonzero_pr_flagged(self): + b = board_json(funcs=[F("tud_task",22000,0.0)]) + p = board_json(funcs=[F("tud_task",22000,5.0)]) + md = self.go([b],[p]) + line = next(l for l in md.splitlines() if "tud_task" in l) + self.assertIn("**+∞%**", line) + + def test_zero_base_zero_pr_stays_dash(self): + b = board_json(funcs=[F("tud_task",22000,0.0)]) + p = board_json(funcs=[F("tud_task",22000,0.0)]) + md = self.go([b],[p]) + line = next(l for l in md.splitlines() if "tud_task" in l) + self.assertTrue(line.rstrip().endswith("| — |"), line) + + # ---- N7: an all-zero charted series must not emit a broken "0 --> 0" mermaid axis ---- + + def test_chart_skipped_when_every_charted_value_is_zero(self): + funcs = [F("tud_task", 1000, 0.0)] + md = self.go([board_json(funcs=funcs)],[board_json(funcs=funcs)]) + self.assertNotIn("```mermaid", md) + self.assertIn("tud_task", md) # table row still present + +class P99Threshold(unittest.TestCase): + """N13: pct()'s corrected nearest-rank index (sysview_report.py) makes p99 distinct from the + max sample at every n>=2, so P99_MIN_N is no longer justified as "avoid colliding with the + max" -- gate_p99's docstring and P99_MIN_N's own comment now cite tail-estimate reliability + instead. This locks the boundary itself so a silent future change notices here too.""" + def test_withheld_at_boundary_n(self): + side = {"metrics": {"live_window_s": 14.2, "overflow": 0}, "capture": {}} + self.assertEqual(sysview_ci.gate_p99(side, {"n": sysview_ci.P99_MIN_N}), + f"n={sysview_ci.P99_MIN_N}") + + def test_shown_just_above_boundary(self): + side = {"metrics": {"live_window_s": 14.2, "overflow": 0}, "capture": {}} + self.assertIsNone(sysview_ci.gate_p99(side, {"n": sysview_ci.P99_MIN_N + 1})) + +class CaptureSelection(unittest.TestCase): + CFG = {"boards": [ + {"name": "a", "uid": "U1", "flasher": {"name": "openocd", "uid": "P1", + "args": "-f interface/x.cfg -f target/y.cfg"}, + "sysview": {"example": "device/cdc_msc", "workload": "cdc_burst", "duration_s": 15}}, + {"name": "b", "uid": "U2", "flasher": {"name": "jlink", "uid": "P2", "args": "-device X"}, + "sysview": {"example": "device/cdc_msc", "workload": "idle", "duration_s": 10, + "ocd_args": "-f interface/jlink.cfg -f target/z.cfg"}}, + {"name": "c", "uid": "U3", "flasher": {"name": "openocd", "uid": "P3", "args": ""}}]} + + def test_flagged_only(self): + self.assertEqual([b["name"] for b in sysview_ci.select_boards(self.CFG, [])], ["a", "b"]) + + def test_intersection_with_board_args(self): + self.assertEqual([b["name"] for b in sysview_ci.select_boards(self.CFG, ["b", "c"])], ["b"]) + + def test_ocd_args_resolution(self): + a, b = sysview_ci.select_boards(self.CFG, []) + self.assertIn("target/y.cfg", sysview_ci.capture_ocd_args(a)) # falls back to flasher + self.assertIn("target/z.cfg", sysview_ci.capture_ocd_args(b)) # explicit override + with self.assertRaises(ValueError): # jlink flasher, no override + sysview_ci.capture_ocd_args({"flasher": {"name": "jlink", "args": "-device X"}, + "sysview": {}}) + + def test_wrapper_error_shape(self): + j = sysview_ci.board_result(self.CFG["boards"][0], "abc1234", error="flash failed: rc=1") + self.assertEqual(j["error"], "flash failed: rc=1"); self.assertIsNone(j["metrics"]) + + +class WorkloadCpuRow(unittest.TestCase): + """When contexts_workload is present, the CPU-load row uses it -- busy share over the + throughput window -- labelled as such, with deltas computed on that number.""" + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) + self.base, self.pr = os.path.join(self.tmp, 'b'), os.path.join(self.tmp, 'p') + + def _wl(self, pct, busy): + j = board_json() + j["metrics"]["contexts_workload"] = [ + {"name": "usbd", "busy_ms": busy, "cpu_pct_workload": pct}] + j["metrics"]["workload_window_s"] = 14.0 + j["metrics"]["workload_anchor"] = "cdc-read-span" + return j + + def test_workload_number_preferred_and_labelled(self): + write_set(self.base, self._wl(25.0, 500.0)) + write_set(self.pr, self._wl(35.0, 700.0)) + md = sysview_ci.report(self.base, self.pr) + self.assertIn("| CPU load, workload win (usbd) | 25.0 % | 35.0 % | +10.0 pt |", md) + + +class SystemExitDoesNotAbortRun(unittest.TestCase): + """W1: sysview_record.rtt_cb_from_elf() sys.exit()s when the target ELF has no _SEGGER_RTT + symbol. SystemExit derives from BaseException, so flash_and_capture_one's blanket + `except Exception` did not catch it -- it used to escape all the way out of the per-board + loop in main()'s PHASE 2 and abort the whole multi-board run, so boards captured + successfully before the bad one never got their results written in PHASE 3. A board hitting + this must produce its own (None, error) result instead of raising, and any board after it + in iteration order must still be attempted.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) + # sysview_ci.py resolves the scripts/hil_flash modules relative to the real repo root, + # the same way main() does -- reuse that so this test exercises the genuine + # sysview_record.rtt_cb_from_elf, not a hand-rolled stand-in. + self.repo_root = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(sysview_ci.__file__)))) + scripts = os.path.join(self.repo_root, ".claude", "skills", "sysview", "scripts") + sys.path.insert(0, scripts) + import sysview_record + import hil_flash + from helper import hil_util + self.sysview_record, self.hil_flash, self.hil_util = sysview_record, hil_flash, hil_util + self._orig_rtt_cb = sysview_record.rtt_cb_from_elf + self._orig_flash_openocd = hil_flash.flash_openocd + self._orig_get_serial = hil_util.get_serial_dev + self.addCleanup(self._restore) + # a serial "node" that already exists, so the enumeration-wait loops in + # flash_and_capture_one don't spend real seconds sleeping + node = os.path.join(self.tmp, "node") + open(node, "w").close() + hil_util.get_serial_dev = lambda *a, **k: node + hil_flash.flash_openocd = lambda board, fw: subprocess.CompletedProcess([], 0) + + def _restore(self): + self.sysview_record.rtt_cb_from_elf = self._orig_rtt_cb + self.hil_flash.flash_openocd = self._orig_flash_openocd + self.hil_util.get_serial_dev = self._orig_get_serial + + def _board(self, name): + return {"name": name, "uid": "U", "flasher": {"name": "openocd", "uid": "P", + "args": "-f interface/x.cfg -f target/y.cfg"}, + "sysview": {"example": "device/cdc_msc", "workload": "idle", "duration_s": 1}} + + def test_bad_elf_errors_without_raising_and_next_board_is_still_attempted(self): + self.sysview_record.rtt_cb_from_elf = lambda elf: (_ for _ in ()).throw( + SystemExit(f"error: no _SEGGER_RTT symbol in {elf} -- not a SystemView build?")) + capture, err = sysview_ci.flash_and_capture_one( + self._board("bad"), os.path.join(self.tmp, "bad"), self.tmp, self.repo_root) + self.assertIsNone(capture) + self.assertIn("no RTT symbol", err) + + # A second, well-formed board must still be reached afterward -- the bug was that the + # SystemExit above aborted the whole process, not just this one board's capture. Whether + # THIS board's own capture ultimately succeeds doesn't matter here (no real probe/openocd + # is attached in this test environment); what matters is that rtt_cb_from_elf actually + # gets called for it, proving the loop moved on instead of dying on the first board. + called = [] + self.sysview_record.rtt_cb_from_elf = lambda elf: called.append(elf) or "0x0" + sysview_ci.flash_and_capture_one( + self._board("good"), os.path.join(self.tmp, "good"), self.tmp, self.repo_root) + self.assertEqual(called, [os.path.join(self.tmp, "good") + ".elf"]) + + +class SessionReset(unittest.TestCase): + """The capture session's own `reset run` must be skipped on two board classes: + WCH (under SDI the target never comes back -- detected by target config, the only + spelling left after #3804 folded the openocd_wch flasher name away), and boards + flagged sysview.attach_only (metro_m4_express: after openocd's reset on atsame5x + the core stays held and the RTT control block never appears -- the flasher's own + post-flash reset already supplied the fresh boot).""" + def test_wch_board_detected_by_target_config(self): + b = {"flasher": {"name": "openocd", "args": "-f target/wch-riscv.cfg"}} + self.assertTrue(sysview_ci.is_wch_board(b)) + self.assertFalse(sysview_ci.session_resets({**b, "sysview": {}})) + + def test_non_wch_openocd_board_resets(self): + b = {"flasher": {"name": "openocd", "args": "-f target/rp2040.cfg"}, "sysview": {}} + self.assertFalse(sysview_ci.is_wch_board(b)) + self.assertTrue(sysview_ci.session_resets(b)) + + def test_attach_only_board_does_not_reset(self): + b = {"flasher": {"name": "jlink", "args": "-device ATSAMD51J19"}, + "sysview": {"attach_only": True}} + self.assertFalse(sysview_ci.session_resets(b)) + + +class TestErrExcerpt(unittest.TestCase): + def test_short_error_kept_whole(self): + self.assertEqual(sysview_ci._err_excerpt(" boom "), "boom") + + def test_long_error_keeps_both_ends(self): + # cmake puts the diagnosis first and the call stack last; a tail-only slice + # dropped the "Cannot find source file" line that names the real problem. + err = "CMake Error: Cannot find source file: startup.s\n" + "x" * 4000 + "\nGenerate step failed." + got = sysview_ci._err_excerpt(err) + self.assertIn("Cannot find source file", got) + self.assertIn("Generate step failed.", got) + self.assertLess(len(got), len(err)) + + def test_none_error_is_empty(self): + self.assertEqual(sysview_ci._err_excerpt(None), "") + + +class TestBuildOneFirmwarePath(unittest.TestCase): + """build_one must return a path the flasher can actually open. JLinkExe's loadfile + and OpenOCD's program both infer the image format from the extension, so an + extension-less path dies with 'File is of unknown / unsupported format' -- and only + after phase 2 has taken the board lock. Observed on stm32f407disco + raspberry_pi_pico.""" + + def _board(self, flasher="jlink"): + return {"name": "stm32f407disco", "flasher": {"name": flasher}, + "sysview": {"example": "device/cdc_msc"}} + + def _repo(self, artifacts): + root = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, root, True) + bdir = os.path.join(root, "examples", "cmake-build-sysview-stm32f407disco") + os.makedirs(bdir) + for a in artifacts: + open(os.path.join(bdir, a), "w").close() + return root + + def test_returns_existing_file_with_extension(self): + root = self._repo(["cdc_msc.elf", "cdc_msc.bin", "cdc_msc.hex"]) + with unittest.mock.patch("subprocess.run", + return_value=subprocess.CompletedProcess([], 0, "", "")): + fw, err = sysview_ci.build_one(self._board(), root) + self.assertIsNone(err) + self.assertTrue(fw.endswith(".elf"), fw) + self.assertTrue(os.path.exists(fw), fw) + + def test_missing_binary_is_an_error_not_a_bad_path(self): + # a build that emitted nothing must fail in phase 1, before the board lock + root = self._repo([]) + with unittest.mock.patch("subprocess.run", + return_value=subprocess.CompletedProcess([], 0, "", "")): + fw, err = sysview_ci.build_one(self._board(), root) + self.assertIsNone(fw) + self.assertIn("no .elf", err) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/hil/test/test_sysview_record.py b/test/hil/test/test_sysview_record.py new file mode 100644 index 000000000..e3175d612 --- /dev/null +++ b/test/hil/test/test_sysview_record.py @@ -0,0 +1,33 @@ +#!/usr/bin/env python3 +"""Guards for sysview_record.py's host-side state handling.""" +import os, sys, tempfile, unittest +from pathlib import Path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', + '.claude', 'skills', 'sysview', 'scripts')) +import sysview_record as R + + +class StubDetection(unittest.TestCase): + """A SIGKILLed run leaves our stub in place with its backup already consumed. The next run + must not treat that stub as the user's config -- doing so lost a real 19 KB SystemView + config permanently on the rig.""" + + def test_our_stub_recognised(self): + with tempfile.TemporaryDirectory() as d: + p = Path(d) / "ini" + p.write_text("[Preferences]\nLoadProjectOnStart=false\nSaveProperties=false\n") + self.assertTrue(R._is_our_stub(p)) + + def test_real_config_not_mistaken_for_stub(self): + with tempfile.TemporaryDirectory() as d: + p = Path(d) / "ini" + p.write_text("[Preferences]\nLoadProjectOnStart=false\n" + "WindowState=@ByteArray(" + + "x" * 500 + ")\n") + self.assertFalse(R._is_our_stub(p)) + + def test_missing_file_is_not_a_stub(self): + self.assertFalse(R._is_our_stub(Path("/nonexistent/ini"))) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/hil/test/test_sysview_report.py b/test/hil/test/test_sysview_report.py new file mode 100644 index 000000000..95333f9fa --- /dev/null +++ b/test/hil/test/test_sysview_report.py @@ -0,0 +1,419 @@ +#!/usr/bin/env python3 +"""Tests for sysview_report.py's data-quality guards (spliced pairs, live window).""" +import importlib.util, json, os, re, statistics, subprocess, sys, tempfile, unittest + +REPORT = os.path.join(os.path.dirname(__file__), '..', '..', '..', + '.claude', 'skills', 'sysview', 'scripts', 'sysview_report.py') +TUSB_SYSVIEW_H = os.path.join(os.path.dirname(__file__), '..', '..', '..', + 'src', 'common', 'tusb_sysview.h') + +def _load_report_module(): + """Import sysview_report.py by path to unit-test its internals (pct()) directly, without + going through the subprocess/--json path run_report() below uses for the higher-level + behaviors. Module-level code only defines functions; main() is guarded by __name__.""" + spec = importlib.util.spec_from_file_location("sysview_report", REPORT) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + +sysview_report = _load_report_module() + +EV_HEADER = ("sequencenum,timestamp,context,event,detail,timestampint," + "contextinint,contextint,contextoutint,eventint,eventoffset,eventsize,eventdata\n") +CTX_HEADER = "Name,Type,Activations,CPU Load,Total Run Time,Total Blocked Time,Min Run Time,Avg Run Time,Max Run Time\n" + +def ev(seq, ts, event, detail=""): + return f'{seq},0.0,"ctx","{event}","{detail}",{ts},0x0,0x0,0x0,0,0,0,\n' + +INIT = ev(0, 0, "Init", "Cycle Freq.: 1000000, CPU Freq.: 48000000, ID Base: 0x20000000, ID Shift: 0") + +def run_report(events_rows, contexts_rows="", include_init=True): + with tempfile.TemporaryDirectory() as d: + with open(os.path.join(d, 'events.txt'), 'w') as f: + f.write(EV_HEADER + (INIT if include_init else "") + "".join(events_rows)) + with open(os.path.join(d, 'contexts.csv'), 'w') as f: + f.write(CTX_HEADER + contexts_rows) + r = subprocess.run([sys.executable, REPORT, d, '--json'], + capture_output=True, text=True) + assert r.returncode == 0, r.stderr + return json.loads(r.stdout) + +class SplicedPairs(unittest.TestCase): + def test_clean_pairs_kept(self): + rows = [] + t = 1000 + for i in range(60): # 60 clean call/ret pairs, 10 us each + rows.append(ev(len(rows)+1, t, "Function #512")) + rows.append(ev(len(rows)+1, t+10, "Function #512", "Returns after 10.000 us")) + t += 1000 + j = run_report(rows) + fn = {f['name']: f for f in j['functions']} + self.assertEqual(fn['tud_task']['n'], 60) + self.assertEqual(j['dropped_pairs'], 0) + + def test_ret_after_overflow_dropped(self): + rows = [ev(1, 1000, "Function #512")] # call + rows.append(ev(2, 1500, "*** Overflow ***")) # loss marker + rows.append(ev(3, 135000, "Function #512", "Returns after 134000.000 us")) # spliced ret + j = run_report(rows) + self.assertEqual(j.get('functions', []), []) # no bogus 134 ms sample + self.assertEqual(j['dropped_pairs'], 2) # invalidated call + orphan ret + + def test_double_call_drops_first(self): + rows = [ev(1, 1000, "Function #512"), # call, ret lost + ev(2, 2000, "Function #512"), # next call + ev(3, 2010, "Function #512", "Returns after 10.000 us")] + j = run_report(rows) + fn = {f['name']: f for f in j['functions']} + self.assertEqual(fn['tud_task']['n'], 1) # only the clean pair + self.assertEqual(j['dropped_pairs'], 1) + + def test_nested_call_ret_both_kept(self): + # CALL,CALL,RET,RET on the same id (e.g. a preempted task): LIFO pairing must match + # each RET to its innermost still-open CALL, keeping BOTH durations instead of + # dropping one and double-counting dropped_pairs (F5b). + rows = [ev(1, 1000, "Function #512"), # outer call + ev(2, 1010, "Function #512"), # inner call + ev(3, 1015, "Function #512", "Returns after 5.000 us"), # inner ret + ev(4, 1030, "Function #512", "Returns after 30.000 us")] # outer ret + j = run_report(rows) + fn = {f['name']: f for f in j['functions']} + self.assertEqual(fn['tud_task']['n'], 2) # both pairs kept, none dropped + self.assertEqual(fn['tud_task']['max_us'], 30.0) # outer (30 us), not spliced + self.assertEqual(j['dropped_pairs'], 0) + + def test_call_before_large_gap_ret_after_kept(self): + # F5a: a CALL must still enter open_calls even when a large gap in timestamps follows it, + # or its RET on the far side of the gap is wrongly discarded as data loss -- + # systematically losing the first invocation of every instrumented function. (There is no + # "window" boundary here to speak of -- pairing bookkeeping runs for every row regardless + # of the capture's live span; this is purely about surviving an arbitrarily large gap + # between a CALL and its RET.) + rows = [ev(1, 500, "Function #512")] # call + t = 99_000_000 # a large gap follows + rows.append(ev(2, t, "Function #512", "Returns after 5.000 us")) # ret, well after it + j = run_report(rows) + fn = {f['name']: f for f in j['functions']} + self.assertEqual(fn['tud_task']['n'], 1) # kept, not lost across the gap + self.assertEqual(j['dropped_pairs'], 0) + +class BareReturns(unittest.TestCase): + """SystemView annotates a return with its duration only rarely -- measured 0.5% of + returns on stm32f407disco and 0.8% on raspberry_pi_pico; the rest are a bare + 'Returns'. Keying CALL-vs-RET off the duration regex misread those as calls.""" + + def test_bare_returns_pair_and_are_not_dropped(self): + rows = [] + t = 1000 + for i in range(60): # 60 pairs, bare returns, 10 ticks = 10 us + rows.append(ev(len(rows)+1, t, "Function #512")) + rows.append(ev(len(rows)+1, t+10, "Function #512", "Returns")) + t += 1000 + j = run_report(rows) + fn = {f['name']: f for f in j['functions']} + self.assertEqual(fn['tud_task']['n'], 60) + self.assertEqual(j['dropped_pairs'], 0) + + def test_computed_duration_matches_annotated(self): + # INIT declares Cycle Freq. 1000000 -> 1 tick == 1 us. A bare return spanning + # 10 ticks must yield the same duration as an explicitly annotated 10.000 us one. + bare = [ev(1, 1000, "Function #512"), ev(2, 1010, "Function #512", "Returns")] + ann = [ev(1, 1000, "Function #512"), + ev(2, 1010, "Function #512", "Returns after 10.000 us")] + self.assertEqual(run_report(bare)['functions'][0]['p50_us'], + run_report(ann)['functions'][0]['p50_us']) + + def test_mixed_annotated_and_bare_all_counted(self): + rows = [ev(1, 1000, "Function #512"), + ev(2, 1010, "Function #512", "Returns after 10.000 us"), + ev(3, 2000, "Function #512"), + ev(4, 2010, "Function #512", "Returns")] + j = run_report(rows) + self.assertEqual(j['functions'][0]['n'], 2) + self.assertEqual(j['dropped_pairs'], 0) + + +class OverflowCount(unittest.TestCase): + """Overflow is the real count over the capture -- one number, no live/stale split. The + reporter no longer re-interprets which part of a capture 'counts'.""" + def test_every_overflow_marker_counted(self): + t = 99_000_000 + rows = [ev(1, 500, "*** Overflow ***"), + ev(2, t, "ISR Enter", "Runs for 5.000 us"), + ev(3, t + 50_000, "*** Overflow ***"), + ev(4, t + 100_000, "ISR Enter", "Runs for 5.000 us")] + j = run_report(rows) + self.assertEqual(j['overflow'], 2) + self.assertNotIn('overflow_total', j) # the second, "diagnostic" view is gone + +class NoTimeBase(unittest.TestCase): + """F3: a missing Init row means no trustworthy timestamp frequency -- withhold rather than + silently assume 1 MHz (wrong by up to 168x on a DWT core).""" + def test_missing_init_withholds_live_window_and_cpu_pct(self): + rows = [ev(1, 1000, "ISR Enter", "Runs for 5.000 us")] + ctx_rows = "usbd,Task,10,50.00%,1.000 000 s,0.000 000 s,0.000 000 s,0.000 000 s,0.000 000 s\n" + j = run_report(rows, ctx_rows, include_init=False) + self.assertIsNone(j['live_window_s']) + self.assertIsNone(j['contexts'][0]['cpu_pct']) + self.assertTrue(any('Init' in w for w in j['warnings'])) + +class NegativeLiveWindow(unittest.TestCase): + """N5: a decreasing timestamp (DWT/timer wrap) makes live_window_s negative -- the + a raw (negative) subtraction used to sail past both gate()'s `is None` check and the cpu_pct + rescale's `> 0` guard, publishing a 'live -N s' heading with CPU% mislabelled. A clock that + ran backwards is corrupt time, not capture data -- withholding is not re-interpreting it.""" + def test_decreasing_timestamps_withhold_as_no_time_base(self): + rows = [ev(1, 5_000_000, "ISR Enter", "Runs for 5.000 us"), + ev(2, 1_000_000, "ISR Enter", "Runs for 5.000 us")] # ts goes backwards + j = run_report(rows) + self.assertIsNone(j['live_window_s']) + self.assertTrue(any('not monotonic' in w for w in j['warnings']), j['warnings']) + + def test_zero_span_also_withheld(self): + rows = [ev(1, 0, "ISR Enter", "Runs for 5.000 us"), + ev(2, 0, "ISR Enter", "Runs for 5.000 us")] # all one ts -> 0 span + j = run_report(rows) + self.assertIsNone(j['live_window_s']) + +class CpuPctScope(unittest.TestCase): + """cpu_pct is SystemView's own CPU Load column, passed through as recorded. It used to be + recomputed against the window and clamped to 100 -- a derived number wearing the measured + one's name.""" + def test_cpu_pct_is_systemviews_value_unmodified(self): + rows = [ev(1, 1000, "ISR Enter", "Runs for 5.000 us"), + ev(2, 500_000, "ISR Enter", "Runs for 5.000 us")] + ctx_rows = "usbd,Task,10,50.00%,1.000 000 s,0.000 000 s,0.000 000 s,0.000 000 s,0.000 000 s\n" + j = run_report(rows, ctx_rows) + self.assertEqual(j['contexts'][0]['cpu_pct'], 50.0) # not rescaled, not clamped + +class PctFunction(unittest.TestCase): + """V5: pct() must use the conventional nearest-rank index (int((n-1)*p/100)), not + int(n*p/100) -- the latter lands one slot too high, e.g. reporting the MAX as the p50 median + at n=2, and biasing every low-n p50 upward.""" + def test_p50_two_elements_is_not_max(self): + self.assertEqual(sysview_report.pct([10, 100], 50), 10) + + def test_p50_matches_nearest_rank_median(self): + # Nearest-rank median == statistics.median_low() for both odd and even n. + for vals in ([1, 2, 3, 4, 5], [1, 2, 3, 4], list(range(1, 51))): + with self.subTest(vals=vals): + self.assertEqual(sysview_report.pct(list(vals), 50), statistics.median_low(vals)) + +class LiveWindow(unittest.TestCase): + """The reported window is the capture itself: first event to last, nothing trimmed. An idle + stretch is a real observation about the target, so it stays in the numbers.""" + def _mixed_rows(self): + rows, t = [], 0 + for i in range(55): + rows.append(ev(len(rows) + 1, t, "ISR Enter", "Runs for 100.000 us")); t += 15 + t = 99_000_000 # a long quiet stretch, then more traffic + for i in range(60): + rows.append(ev(len(rows) + 1, t, "ISR Enter", "Runs for 5.000 us")); t += 100_000 + return rows + + def test_all_samples_kept(self): + j = run_report(self._mixed_rows()) + self.assertEqual(j['isr'][0]['n'], 115) # every recorded sample, not a chosen subset + + def test_window_spans_the_whole_capture(self): + j = run_report(self._mixed_rows()) + self.assertAlmostEqual(j['live_window_s'], 104.9, delta=0.2) + + def test_no_gap_means_full_span(self): + rows = [ev(1, 0, "ISR Enter", "Runs for 5.000 us"), + ev(2, 500_000, "ISR Enter", "Runs for 5.000 us")] + j = run_report(rows) + self.assertAlmostEqual(j['live_window_s'], 0.5, delta=0.01) + + +class NoTrimming(unittest.TestCase): + """The reporter reports the capture, it does not decide which part of it counts. + + A gap heuristic used to pick a 'live window' and silently drop everything before it. Every + variant was wrong for some real capture: anchoring on the last qualifying gap emptied + stm32f072disco's tables when its capture merely ended quiet (the OpenOCD route's own ~2.0 s + pre-stop settle measured 2.019 s), and anchoring on the largest instead discarded half the + events whenever a device idled mid-capture.""" + + def _burst(self, rows, seq, ts, count, step=10_000): + for _ in range(count): + rows.append(ev(seq, ts, "ISR Enter", "Runs for 10.000 us")); seq += 1; ts += step + return seq, ts + + def test_trailing_lull_keeps_everything(self): + rows, seq, ts = [], 1, 200_000 + seq, ts = self._burst(rows, seq, ts, 400) + ts += 2_019_000 # the f072 case + rows.append(ev(seq, ts, "ISR Enter", "Runs for 50.000 us")) + self.assertEqual(run_report(rows)['isr'][0]['n'], 401) + + def test_mid_capture_idle_keeps_everything(self): + rows, seq, ts = [], 1, 1_000_000 + seq, ts = self._burst(rows, seq, ts, 300) + ts += 30_000_000 # device quiet between bursts + seq, ts = self._burst(rows, seq, ts, 300) + self.assertEqual(run_report(rows)['isr'][0]['n'], 600) + + def test_long_leading_gap_keeps_everything(self): + rows = [ev(1, 0, "ISR Enter", "Runs for 5.000 us")] + self._burst(rows, 2, 90_000_000, 300) + j = run_report(rows) + self.assertEqual(j['isr'][0]['n'], 301) + self.assertAlmostEqual(j['live_window_s'], 92.99, delta=0.2) + + +class WorkloadWindowCpu(unittest.TestCase): + """cpu_pct_workload: per-context busy time summed from the raw scheduling events + (Task Run / System Idle / ISR Enter+Exit transitions), over a window DEFINED by the + workload's own footprint -- the span of tud_cdc_read events (Function #516) -- or an + explicit --window. A derived metric under its own name; SystemView's cpu_pct column is + never touched. Busy time inside a window cannot exceed the window, so no clamp exists.""" + + def _sched(self, seq, ts, event, ctx="ctx"): + return f'{seq},0.0,"{ctx}","{event}","",{ts},0x0,0x0,0x0,0,0,0,\n' + + def _capture(self): + """1s idle | 2s workload (usbd 50% / Idle 50%, CDC reads bracket it) | 1s idle.""" + rows, seq = [], 1 + rows.append(self._sched(seq, 0, "System Idle", "Idle")); seq += 1 + t = 1_000_000 # workload starts + rows.append(ev(seq, t, "Function #516")); seq += 1 # first tud_cdc_read CALL + # alternate 100ms usbd / 100ms Idle for 2s => usbd busy 50% of window + for i in range(10): + rows.append(self._sched(seq, t, "Task Run", "usbd")); seq += 1; t += 100_000 + rows.append(self._sched(seq, t, "System Idle", "Idle")); seq += 1; t += 100_000 + rows.append(ev(seq, t, "Function #516", "Returns after 5.000 us")); seq += 1 # last read RET + rows.append(self._sched(seq, t + 1_000_000, "System Idle", "Idle")); seq += 1 + return rows + + def test_workload_window_anchored_on_cdc_reads(self): + j = run_report(self._capture()) + self.assertAlmostEqual(j['workload_window_s'], 2.0, delta=0.01) + self.assertEqual(j['workload_anchor'], 'cdc-read-span') + + def test_busy_share_within_window(self): + j = run_report(self._capture()) + by = {c['name']: c for c in j['contexts_workload']} + self.assertAlmostEqual(by['usbd']['cpu_pct_workload'], 50.0, delta=1.0) + self.assertAlmostEqual(by['Idle']['cpu_pct_workload'], 50.0, delta=1.0) + + def test_isr_time_attributed_to_isr_not_task(self): + rows, seq = [], 1 + t = 0 + rows.append(ev(seq, t, "Function #516")); seq += 1 + rows.append(self._sched(seq, t, "Task Run", "usbd")); seq += 1 + # 1s window: usbd runs, but a 200ms ISR preempts in the middle + rows.append(self._sched(seq, t + 400_000, "ISR Enter", "ISR 83")); seq += 1 + rows.append(self._sched(seq, t + 600_000, "ISR Exit", "ISR 83")); seq += 1 + rows.append(ev(seq, t + 1_000_000, "Function #516", "Returns after 5.000 us")); seq += 1 + j = run_report(rows) + by = {c['name']: c for c in j['contexts_workload']} + self.assertAlmostEqual(by['usbd']['cpu_pct_workload'], 80.0, delta=1.0) + self.assertAlmostEqual(by['ISR 83']['cpu_pct_workload'], 20.0, delta=1.0) + + def test_no_cdc_activity_means_null(self): + rows = [ev(1, 0, "ISR Enter", "Runs for 5.000 us"), + ev(2, 1_000_000, "ISR Enter", "Runs for 5.000 us")] + j = run_report(rows) + self.assertIsNone(j['workload_window_s']) + self.assertEqual(j['contexts_workload'], []) + + def test_never_exceeds_100(self): + j = run_report(self._capture()) + for c in j['contexts_workload']: + self.assertLessEqual(c['cpu_pct_workload'], 100.0) + + +class ClockCorruptWithholdsWorkloadMetrics(unittest.TestCase): + """W5: the backwards-step check already withholds live_window_s when timestamps are + non-monotonic (counter wrap), but workload_window_s/workload_anchor/contexts_workload are + computed from that same valid_ts/freq and used to still get published -- repro'd as + cpu_pct_workload 100.0 showing up right alongside the "no time base" warning that says the + clock can't be trusted. All of it must be withheld together.""" + def test_wrapped_timestamps_withhold_workload_metrics_too(self): + rows = [ + ev(1, 5_000_000, "ISR Enter", "Runs for 5.000 us"), + ev(2, 1_000_000, "ISR Enter", "Runs for 5.000 us"), # ts goes backwards -> corrupt clock + ev(3, 2_000_000, "Function #516"), # tud_cdc_read CALL + ev(4, 2_000_000, "Task Run"), # "ctx" busy across the span + ev(5, 3_000_000, "Function #516", "Returns after 5.000 us"), # tud_cdc_read RET + ] + j = run_report(rows) + self.assertIsNone(j['live_window_s']) # sanity: the existing guard still fires + self.assertIsNone(j['workload_window_s']) + self.assertIsNone(j['workload_anchor']) + self.assertEqual(j['contexts_workload'], []) + + +class WindowCliValidation(unittest.TestCase): + """W6: --window's own CLI-anchor path had no last>first check -- both fallback anchors + (markers, cdc-read-span) already reject a non-positive span, so --window 5:2 was the one way + left to silently produce a negative workload_window_s.""" + def test_inverted_window_range_errors(self): + with tempfile.TemporaryDirectory() as d: + with open(os.path.join(d, 'events.txt'), 'w') as f: + f.write(EV_HEADER + INIT) + with open(os.path.join(d, 'contexts.csv'), 'w') as f: + f.write(CTX_HEADER) + r = subprocess.run([sys.executable, REPORT, d, '--json', '--window', '5:2'], + capture_output=True, text=True) + self.assertNotEqual(r.returncode, 0) + self.assertIn('--window end must be after start', r.stderr) + + +class FuncNameTableMatchesHeader(unittest.TestCase): + """W13: TU_SV_FUNC_NAMES (sysview_report.py) hand-mirrors the id->name table + src/common/tusb_sysview.h maintains for exactly this purpose -- its own comment says the + table is kept "so Task 5's host-side reporter (sysview_report.py) can map id -> name + itself". Nothing tied the two together before: inserting/reordering a tu_sysview_id_t member + would silently relabel every function after it with no test noticing. This parses the + header's id->name comment and its enum member count, and cross-checks both against + TU_SV_FUNC_NAMES / TU_SV_EVENT_BASE.""" + + def setUp(self): + with open(TUSB_SYSVIEW_H) as f: + self.text = f.read() + + def test_id_name_comment_matches_func_names_table(self): + m = re.search(r"map id -> name itself:\s*\n((?:\s*//.*\n)+?)\s*typedef enum", self.text) + self.assertIsNotNone(m, "couldn't find the id->name comment above tu_sysview_id_t") + pairs = re.findall(r"(\d+)\s+([A-Za-z_][A-Za-z0-9_]*)", m.group(1)) + self.assertTrue(pairs) + header_names = {int(i): name for i, name in pairs} + self.assertEqual(header_names, sysview_report.TU_SV_FUNC_NAMES) + + def test_enum_member_count_matches_table(self): + m = re.search(r"typedef enum\s*\{(.*?)\}\s*tu_sysview_id_t;", self.text, re.S) + self.assertIsNotNone(m, "couldn't find the tu_sysview_id_t enum body") + members = [ln.strip().split(",")[0].split("=")[0].strip() + for ln in m.group(1).splitlines() if ln.strip().startswith("TU_SV_ID_")] + members = [name for name in members if not name.endswith("_COUNT")] + self.assertEqual(len(members), len(sysview_report.TU_SV_FUNC_NAMES)) + + def test_event_base_matches_header(self): + m = re.search(r"#define\s+TU_SV_EVENT_BASE\s+(\d+)", self.text) + self.assertIsNotNone(m, "couldn't find #define TU_SV_EVENT_BASE in the header") + self.assertEqual(int(m.group(1)), sysview_report.TU_SV_EVENT_BASE) + + +class OverflowInExitReasons(unittest.TestCase): + """SystemView marks data loss in TWO ways: explicit "*** Overflow ***" event rows, and exit + reasons reading "Returns to *** OVERFLOW ***" -- the decoder lost the return context to ring + loss. A dual-role dogfood measured 96-99% of ISR exits carrying the second form while the + JSON overflow field read 0 or 1: the report was drastically understating real loss.""" + def test_lost_context_exits_counted(self): + rows = [ev(1, 1000, "ISR Enter", "Runs for 5.000 us"), + ev(2, 2000, "ISR Exit", "Returns to *** OVERFLOW ***"), + ev(3, 3000, "ISR Enter", "Runs for 5.000 us"), + ev(4, 4000, "ISR Exit", "Returns to Idle"), + ev(5, 5000, "*** Overflow ***")] + j = run_report(rows) + self.assertEqual(j['overflow'], 2) # 1 event row + 1 lost-context exit + def test_clean_capture_still_zero(self): + rows = [ev(1, 1000, "ISR Enter", "Runs for 5.000 us"), + ev(2, 2000, "ISR Exit", "Returns to Idle")] + self.assertEqual(run_report(rows)['overflow'], 0) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/hil/tinyusb.json b/test/hil/tinyusb.json index 8fd4683a4..3a4f9804a 100644 --- a/test/hil/tinyusb.json +++ b/test/hil/tinyusb.json @@ -261,6 +261,12 @@ "vid_pid": "0x2e8a 0x000c", "args": "-f interface/cmsis-dap.cfg -f target/rp2040.cfg -c \"adapter speed 5000\"", "verify": true + }, + "sysview": { + "example": "device/cdc_msc", + "workload": "cdc_burst", + "duration_s": 15, + "buffer": 65536 } }, { @@ -379,6 +385,13 @@ "name": "jlink", "uid": "000773661813", "args": "-device stm32f407vg" + }, + "sysview": { + "example": "device/cdc_msc", + "workload": "cdc_burst", + "duration_s": 15, + "buffer": 65536, + "ocd_args": "-f interface/jlink.cfg -c \"transport select swd\" -f target/stm32f4x.cfg" } }, { diff --git a/tools/ci_select.py b/tools/ci_select.py index 47f303d16..9512a0ad7 100755 --- a/tools/ci_select.py +++ b/tools/ci_select.py @@ -87,7 +87,11 @@ def _read(path: str) -> str: _NONCODE_RE = re.compile( # LICENSE is anchored and LICENSES/ named separately: a bare `LICENSE` alternative # also swallowed anything merely STARTING with it (a future LICENSE_extra.c), - # which is the silent-under-selection direction + # which is the silent-under-selection direction. + # .claude/skills/sysview/scripts/ stays here on purpose: the rig's sysview capture + # (test/hil/sysview_ci.py) runs those scripts, but only in a continue-on-error step + # whose output is an informational comment, so a break there cannot fail a run; + # the pre-commit hil-test hook runs their unit tests instead. r'^(docs/|\.claude/|.*\.(md|rst)$|LICENSE$|LICENSES/)') # Repo metadata and tooling that no CI build reads. Enumerated rather than left to # rule 17, which widens BOTH axes: a PR touching only .gitignore and a README was |
