aboutsummaryrefslogtreecommitdiffziptar.gz
path: root/scripts
diff options
context:
space:
mode:
Diffstat (limited to 'scripts')
-rwxr-xr-xscripts/alsoft_wave.sh18
-rwxr-xr-xscripts/audio_report.py101
-rwxr-xr-xscripts/digest_compare.py34
-rw-r--r--scripts/godosa_licenses.py145
-rwxr-xr-xscripts/mutate.sh52
-rw-r--r--scripts/test_audio_report.py44
-rw-r--r--scripts/test_digest_compare.py42
-rw-r--r--scripts/test_godosa_licenses.py116
-rw-r--r--scripts/test_thread_cpu.py24
-rw-r--r--scripts/test_thread_instr.py21
-rwxr-xr-xscripts/thread_cpu.py60
-rwxr-xr-xscripts/thread_instr.py42
12 files changed, 699 insertions, 0 deletions
diff --git a/scripts/alsoft_wave.sh b/scripts/alsoft_wave.sh
new file mode 100755
index 0000000..31644ec
--- /dev/null
+++ b/scripts/alsoft_wave.sh
@@ -0,0 +1,18 @@
+#!/usr/bin/env bash
+# Writes DIR/alsoft.conf so OpenAL Soft renders to DIR/audio.wav (wave backend, stereo
+# int16) instead of the speakers, and removes an old DIR/audio.wav. Run the game with
+# ALSOFT_CONF=DIR/alsoft.conf. Env: ALSOFT_EXTRA="key = value" lines added to [general].
+# Usage: alsoft_wave.sh DIR
+set -euo pipefail
+out=${1:?usage: alsoft_wave.sh DIR}
+mkdir -p "$out"
+rm -f "$out/audio.wav"
+cat >"$out/alsoft.conf" <<CONF
+[general]
+drivers = wave
+channels = stereo
+sample-type = int16
+${ALSOFT_EXTRA:-}
+[wave]
+file = $out/audio.wav
+CONF
diff --git a/scripts/audio_report.py b/scripts/audio_report.py
new file mode 100755
index 0000000..14af9ab
--- /dev/null
+++ b/scripts/audio_report.py
@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""Levels (and a spectrogram) of a game audio capture — smoke runs write DIR/audio.wav
+through OpenAL Soft's wave backend (audio spec §8). The game is killed at the end of a
+run, so the WAV header sizes may be unfinished: data is read to end of file.
+
+Usage: scripts/audio_report.py WAV [--png OUT.png] [--start S] [--end S]
+ prints one line per second: RMS and peak in dBFS (mono mix)
+ --png: spectrogram of [start, end] via ffmpeg (skipped if ffmpeg is missing)
+"""
+import argparse
+import array
+import math
+import shutil
+import struct
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+
+def read_wav(path):
+ """Returns (rate, channels, samples as list of ints, interleaved). 16-bit PCM only."""
+ data = Path(path).read_bytes()
+ if data[:4] != b"RIFF" or data[8:12] != b"WAVE":
+ raise ValueError("not a RIFF/WAVE file")
+ pos, rate, channels, bits = 12, None, None, None
+ while pos + 8 <= len(data):
+ cid, size = data[pos:pos + 4], struct.unpack("<I", data[pos + 4:pos + 8])[0]
+ body = pos + 8
+ if cid == b"fmt ":
+ _fmt, channels, rate = struct.unpack("<HHI", data[body:body + 8])
+ bits = struct.unpack("<H", data[body + 14:body + 16])[0]
+ elif cid == b"data":
+ if bits != 16:
+ raise ValueError(f"need 16-bit PCM, got {bits}")
+ raw = data[body:] # to EOF: the size may be a placeholder
+ raw = raw[: len(raw) - len(raw) % (2 * channels)]
+ samples = array.array("h")
+ samples.frombytes(raw)
+ if sys.byteorder == "big":
+ samples.byteswap()
+ return rate, channels, samples
+ pos = body + size + (size & 1)
+ raise ValueError("no data chunk")
+
+
+def db(x):
+ return -120.0 if x <= 0 else 20 * math.log10(x)
+
+
+def levels(rate, channels, samples):
+ """[(second, rms_dbfs, peak_dbfs)] of the mono mix, one per whole or partial second."""
+ frames = len(samples) // channels
+ out = []
+ for start in range(0, frames, rate):
+ end = min(frames, start + rate)
+ total, peak = 0.0, 0
+ for f in range(start, end):
+ v = sum(samples[f * channels + c] for c in range(channels)) / channels
+ total += v * v
+ peak = max(peak, abs(v))
+ rms = math.sqrt(total / (end - start)) / 32768
+ out.append((start // rate, db(rms), db(peak / 32768)))
+ return out
+
+
+def write_pcm16(path, rate, channels, samples):
+ raw = samples.tobytes() if sys.byteorder == "little" else array.array("h", samples).byteswap() or samples.tobytes()
+ header = struct.pack("<4sI4s4sIHHIIHH4sI", b"RIFF", 36 + len(raw), b"WAVE", b"fmt ", 16, 1, channels,
+ rate, rate * channels * 2, channels * 2, 16, b"data", len(raw))
+ Path(path).write_bytes(header + raw)
+
+
+def main():
+ ap = argparse.ArgumentParser()
+ ap.add_argument("wav")
+ ap.add_argument("--png")
+ ap.add_argument("--start", type=float, default=0.0)
+ ap.add_argument("--end", type=float)
+ args = ap.parse_args()
+ rate, channels, samples = read_wav(args.wav)
+ print(f"{rate} Hz, {channels} ch, {len(samples) / channels / rate:.1f} s")
+ for second, rms, peak in levels(rate, channels, samples):
+ bar = "#" * max(0, int((rms + 60) / 2))
+ print(f"{second:4d}s rms {rms:7.1f} dBFS peak {peak:7.1f} {bar}")
+ if args.png:
+ if not shutil.which("ffmpeg"):
+ print("ffmpeg missing: no spectrogram")
+ return
+ first = int(args.start * rate) * channels
+ last = len(samples) if args.end is None else int(args.end * rate) * channels
+ with tempfile.TemporaryDirectory() as tmp:
+ fixed = Path(tmp) / "fixed.wav"
+ write_pcm16(fixed, rate, channels, samples[first:last])
+ subprocess.run(["ffmpeg", "-y", "-v", "error", "-i", str(fixed), "-lavfi",
+ "showspectrumpic=s=1200x480:legend=1:scale=log:fscale=log", args.png], check=True)
+ print(f"spectrogram: {args.png}")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/digest_compare.py b/scripts/digest_compare.py
new file mode 100755
index 0000000..b4144fe
--- /dev/null
+++ b/scripts/digest_compare.py
@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+"""Compare the `digest <frame> <time> <hex>` lines of two run logs (deterministic runs).
+
+Usage: digest_compare.py A.log B.log → exit 0 when equal, 1 with the first difference.
+"""
+import sys
+from pathlib import Path
+
+
+def digests(path: str) -> list[str]:
+ return [line.strip() for line in Path(path).read_text(errors="replace").splitlines() if line.startswith("digest ")]
+
+
+def main(argv: list[str]) -> int:
+ if len(argv) != 3:
+ print(__doc__.strip())
+ return 2
+ a, b = digests(argv[1]), digests(argv[2])
+ if not a and not b:
+ print("FAIL: no digest lines")
+ return 1
+ for k, (x, y) in enumerate(zip(a, b), 1):
+ if x != y:
+ print(f"FAIL: digest {k} differs\n {x}\n {y}")
+ return 1
+ if len(a) != len(b):
+ print(f"FAIL: {len(a)} vs {len(b)} digests")
+ return 1
+ print(f"ok: {len(a)} digests equal, last {a[-1]}")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main(sys.argv))
diff --git a/scripts/godosa_licenses.py b/scripts/godosa_licenses.py
new file mode 100644
index 0000000..f28f312
--- /dev/null
+++ b/scripts/godosa_licenses.py
@@ -0,0 +1,145 @@
+#!/usr/bin/env python3
+"""Third-party licence list for a .NET game: generates licenses.md and checks it (Python stdlib only).
+
+A game keeps a small `scripts/licenses.py` that builds a `Config` and calls `main(argv, config)`:
+
+ python3 scripts/licenses.py check: fail if stale, a licence is unknown, or a copyleft
+ package ships (exit 1)
+ python3 scripts/licenses.py --write regenerate licenses.md
+
+Sources: every csproj in the solution: its obj/project.assets.json (needs `dotnet restore`/build first), the nuspec in
+the NuGet cache for licence + copyright. Lifted from an earlier project of the same author (NuGet part).
+"""
+import json
+import os
+import re
+import sys
+import xml.etree.ElementTree as ET
+from dataclasses import dataclass, field
+from pathlib import Path
+
+# Licence expressions that must not ship linked into the game's code. LGPL is allowed only as a separate shared
+# library, via Config.allow_shipped with the reason.
+COPYLEFT = re.compile(r"\b(A?GPL|LGPL|MPL|EPL|CDDL|EUPL|OSL|CC-BY-SA)", re.I)
+
+
+@dataclass
+class Config:
+ solution: str # the .slnx whose projects count
+ shipped_dirs: tuple[str, ...] # project dir prefixes whose output ships; everything else = dev/test only
+ header: str # markdown put above the tables
+ allow_shipped: dict[str, str] = field(default_factory=dict) # package id -> why a copyleft licence may ship
+ overrides: dict[str, str] = field(default_factory=dict) # package id -> licence (nuspec has no expression)
+ extra: list[tuple[str, str, str, str, str]] = field(default_factory=list) # (what, version, licence, copyright, where)
+ planned: list[tuple[str, str, str, str, str]] = field(default_factory=list) # (what, version, licence, ships as, checked in)
+ md_path: str = "licenses.md"
+ command: str = "python3 scripts/licenses.py"
+
+
+def nuget_root() -> Path:
+ return Path(os.environ.get("NUGET_PACKAGES", Path.home() / ".nuget" / "packages"))
+
+
+def nuspec_info(pid: str, ver: str, cache: Path) -> tuple[str, str]:
+ d = cache / pid.lower() / ver.lower()
+ specs = list(d.glob("*.nuspec")) if d.is_dir() else []
+ if not specs:
+ return "", ""
+ root = ET.parse(specs[0]).getroot()
+ lic = copy = ""
+ for el in root.iter():
+ tag = el.tag.rsplit("}", 1)[-1]
+ if tag == "license" and el.get("type") == "expression":
+ lic = (el.text or "").strip()
+ elif tag == "copyright":
+ copy = " ".join((el.text or "").split())
+ return lic, copy
+
+
+def has_runtime_assets(target: dict) -> bool:
+ """False for build-only packages (analyzers, MSBuild tasks): nothing lands in the output."""
+ for key in ("runtime", "native", "runtimeTargets"):
+ if any(not f.endswith("/_._") for f in target.get(key, {})):
+ return True
+ return False
+
+
+def collect_nuget(repo: Path, cache: Path, cfg: Config) -> tuple[dict, list[str]]:
+ """{(id, ver): {"users": set, "ship_users": set, "shipped": bool}}, errors."""
+ pkgs: dict = {}
+ errors = []
+ for rel in sorted(re.findall(r'Path="([^"]+\.csproj)"', (repo / cfg.solution).read_text())):
+ csproj = repo / rel
+ assets = csproj.parent / "obj" / "project.assets.json"
+ if not assets.exists():
+ errors.append(f"{rel}: no obj/project.assets.json (run dotnet restore)")
+ continue
+ data = json.loads(assets.read_text())
+ runtime = {}
+ for tgt in data.get("targets", {}).values():
+ for key, t in tgt.items():
+ if t.get("type") == "package" and has_runtime_assets(t):
+ runtime[key] = True
+ shipped = rel.startswith(cfg.shipped_dirs)
+ for key, lib in data.get("libraries", {}).items():
+ if lib.get("type") != "package":
+ continue
+ pid, ver = key.split("/", 1)
+ e = pkgs.setdefault((pid, ver), {"users": set(), "ship_users": set(), "shipped": False})
+ e["users"].add(csproj.stem)
+ if key in runtime and shipped:
+ e["ship_users"].add(csproj.stem)
+ e["shipped"] = True
+ return pkgs, errors
+
+
+def generate(repo: Path, cache: Path, cfg: Config) -> tuple[str, list[str]]:
+ pkgs, errors = collect_nuget(repo, cache, cfg)
+ rows = []
+ for (pid, ver), e in sorted(pkgs.items(), key=lambda kv: (kv[0][0].lower(), kv[0][1])):
+ lic, copy = nuspec_info(pid, ver, cache)
+ lic = cfg.overrides.get(pid, lic)
+ if not lic:
+ errors.append(f"NuGet {pid} {ver}: unknown licence (add to the overrides after checking)")
+ lic = "UNKNOWN"
+ if e["shipped"] and COPYLEFT.search(lic) and pid not in cfg.allow_shipped:
+ errors.append(f"NuGet {pid} {ver}: copyleft licence {lic} ships")
+ rows.append((pid, ver, lic, copy, e))
+
+ def cell(s: str) -> str:
+ return s.replace("|", "\\|")
+
+ md = [cfg.header]
+ md.append("## Shipped (NuGet)\n")
+ md.append("| Package | Version | Licence | Copyright | Used by | Notes |\n|---|---|---|---|---|---|")
+ for pid, ver, lic, copy, e in rows:
+ if e["shipped"]:
+ md.append(f"| {pid} | {ver} | {lic} | {cell(copy)} | {', '.join(sorted(e['ship_users']))} | {cfg.allow_shipped.get(pid, '')} |")
+ md.append("\n## Not a package\n")
+ md.append("| What | Version | Licence | Copyright | Where |\n|---|---|---|---|---|")
+ for what, ver, lic, copy, where in cfg.extra:
+ md.append(f"| {what} | {ver} | {lic} | {copy} | {where} |")
+ md.append("\n## Planned (audited, not yet used)\n")
+ md.append("| What | Version | Licence | Ships as | Checked in |\n|---|---|---|---|---|")
+ for what, ver, lic, how, src in cfg.planned:
+ md.append(f"| {what} | {ver} | {cell(lic)} | {how} | {src} |")
+ md.append("\n## Test and build-only (NuGet, nothing in the output)\n")
+ md.append("| Package | Version | Licence | Used by |\n|---|---|---|---|")
+ for pid, ver, lic, copy, e in rows:
+ if not e["shipped"]:
+ md.append(f"| {pid} | {ver} | {lic} | {', '.join(sorted(e['users']))} |")
+ return "\n".join(md) + "\n", errors
+
+
+def main(argv: list[str], cfg: Config, repo: Path, cache: Path | None = None) -> int:
+ cache = cache or nuget_root()
+ md, errors = generate(repo, cache, cfg)
+ if "--write" in argv:
+ (repo / cfg.md_path).write_text(md)
+ else:
+ f = repo / cfg.md_path
+ if not f.exists() or f.read_text() != md:
+ errors.append(f"{cfg.md_path} is stale: run {cfg.command} --write")
+ for e in errors:
+ print(e, file=sys.stderr)
+ return 1 if errors else 0
diff --git a/scripts/mutate.sh b/scripts/mutate.sh
new file mode 100755
index 0000000..e3132de
--- /dev/null
+++ b/scripts/mutate.sh
@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+# Mutation check: prove a test can fail. Replaces an exact snippet in one
+# source file (must match exactly once), runs the tests, and ALWAYS restores
+# the file. Exit 0 = mutant killed (some test failed, as it should); 1 =
+# mutant SURVIVED (tests too weak); 3 = mutant doesn't build (invalid —
+# replace a statement with `{ }` instead of deleting a loop body line).
+#
+# Usage (from the repo whose tests must catch it — a game root, or this lib's root):
+# <path>/mutate.sh <file> <old-snippet> <new-snippet>
+# e.g. scripts/mutate.sh src/MyGame.Core/Loop/InputLatch.cs \
+# '_jump |= sample.JumpPressed;' '_jump = sample.JumpPressed;'
+# <path>/mutate.sh <file> <old> <new> -- <test command...>
+# runs that command instead of the whole `dotnet test` (a filtered dotnet test, or any other runner: python
+# unittest, ...); a dotnet command keeps the build / [FAIL] parsing, any other is killed by a non-zero exit.
+set -uo pipefail
+
+file=$1 old=$2 new=$3
+shift 3
+cmd=(dotnet test --nologo)
+if [[ ${1:-} == -- ]]; then
+ shift
+ cmd=("$@")
+fi
+backup=$(mktemp)
+cp "$file" "$backup"
+trap 'cp "$backup" "$file"; rm -f "$backup"' EXIT
+
+python3 - "$file" "$old" "$new" <<'EOF' || exit 2
+import sys
+path, old, new = sys.argv[1:]
+text = open(path, encoding="utf-8").read()
+n = text.count(old)
+if n != 1:
+ sys.exit(f"snippet matches {n} times in {path} (need exactly 1)")
+open(path, "w", encoding="utf-8").write(text.replace(old, new))
+EOF
+
+result=$("${cmd[@]}" 2>&1)
+status=$?
+if [[ ${cmd[0]} != dotnet ]]; then
+ if (( status != 0 )); then
+ echo "KILLED (exit $status):"; tail -3 <<<"$result"; exit 0
+ fi
+ echo "SURVIVED: no test failed — strengthen the tests"; exit 1
+fi
+if grep -q "error CS" <<<"$result"; then
+ echo "INVALID: mutant doesn't build"; grep -m3 "error CS" <<<"$result"; exit 3
+fi
+if grep -q "Failed!" <<<"$result"; then
+ echo "KILLED by:"; grep "\[FAIL\]" <<<"$result" | sed -E 's/^\[xUnit[^]]*\] +//; s/ \[FAIL\]$//' | sort -u; exit 0
+fi
+echo "SURVIVED: no test failed — strengthen the tests"; exit 1
diff --git a/scripts/test_audio_report.py b/scripts/test_audio_report.py
new file mode 100644
index 0000000..87862ca
--- /dev/null
+++ b/scripts/test_audio_report.py
@@ -0,0 +1,44 @@
+import array
+import math
+import struct
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import audio_report as ar
+
+
+def wav_bytes(frames, rate=8000, channels=2, data_size=None):
+ raw = frames.tobytes()
+ size = len(raw) if data_size is None else data_size
+ return struct.pack("<4sI4s4sIHHIIHH4sI", b"RIFF", 36 + size, b"WAVE", b"fmt ", 16, 1, channels,
+ rate, rate * channels * 2, channels * 2, 16, b"data", size) + raw
+
+
+class AudioReportTests(unittest.TestCase):
+ def test_reads_to_eof_even_with_placeholder_data_size(self):
+ frames = array.array("h", [1000, -1000] * 8000) # 1 s stereo at 8 kHz
+ with tempfile.TemporaryDirectory() as tmp:
+ path = Path(tmp) / "a.wav"
+ path.write_bytes(wav_bytes(frames, data_size=0)) # killed writer: size never fixed
+ rate, channels, samples = ar.read_wav(path)
+ self.assertEqual((rate, channels, len(samples)), (8000, 2, 16000))
+
+ def test_levels_per_second_of_mono_mix(self):
+ # second 0: full-scale square on both channels (0 dBFS); second 1: silence
+ loud = [16384 if (i // 20) % 2 else -16384 for i in range(8000)]
+ frames = array.array("h", [v for s in loud for v in (s, s)] + [0, 0] * 8000)
+ lv = ar.levels(8000, 2, frames)
+ self.assertEqual(len(lv), 2)
+ self.assertAlmostEqual(lv[0][1], 20 * math.log10(0.5), places=3) # rms of ±0.5 square
+ self.assertEqual(lv[1][1], -120.0)
+
+ def test_opposite_channels_cancel_in_the_mix(self):
+ frames = array.array("h", [8000, -8000] * 8000)
+ self.assertEqual(ar.levels(8000, 2, frames)[0][1], -120.0)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/scripts/test_digest_compare.py b/scripts/test_digest_compare.py
new file mode 100644
index 0000000..400e057
--- /dev/null
+++ b/scripts/test_digest_compare.py
@@ -0,0 +1,42 @@
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+SCRIPT = Path(__file__).with_name("digest_compare.py")
+
+
+def run(a: str, b: str) -> subprocess.CompletedProcess:
+ with tempfile.TemporaryDirectory() as d:
+ pa, pb = Path(d, "a.log"), Path(d, "b.log")
+ pa.write_text(a)
+ pb.write_text(b)
+ return subprocess.run([sys.executable, str(SCRIPT), str(pa), str(pb)], capture_output=True, text=True)
+
+
+class DigestCompareTests(unittest.TestCase):
+ def test_equal(self):
+ r = run("x\ndigest 1 0 00000000000000aa\n", "y\ndigest 1 0 00000000000000aa\n")
+ self.assertEqual(0, r.returncode)
+ self.assertIn("1 digests equal", r.stdout)
+
+ def test_first_difference(self):
+ r = run("digest 1 0 00000000000000aa\ndigest 2 0 00000000000000bb\n",
+ "digest 1 0 00000000000000aa\ndigest 2 0 00000000000000cc\n")
+ self.assertEqual(1, r.returncode)
+ self.assertIn("digest 2 differs", r.stdout)
+
+ def test_count_mismatch(self):
+ r = run("digest 1 0 00000000000000aa\n", "digest 1 0 00000000000000aa\ndigest 2 0 00000000000000bb\n")
+ self.assertEqual(1, r.returncode)
+ self.assertIn("1 vs 2 digests", r.stdout)
+
+ def test_none(self):
+ r = run("x\n", "y\n")
+ self.assertEqual(1, r.returncode)
+ self.assertIn("no digest lines", r.stdout)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/scripts/test_godosa_licenses.py b/scripts/test_godosa_licenses.py
new file mode 100644
index 0000000..40e8387
--- /dev/null
+++ b/scripts/test_godosa_licenses.py
@@ -0,0 +1,116 @@
+"""python3 -m unittest discover -s scripts -p 'test_*.py'"""
+import contextlib
+import io
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+import godosa_licenses as gl
+
+
+def nuspec(cache: Path, pid: str, ver: str, lic: str | None, copy: str = "© Someone"):
+ d = cache / pid.lower() / ver
+ d.mkdir(parents=True)
+ licel = f'<license type="expression">{lic}</license>' if lic else '<license type="file">LICENSE</license>'
+ (d / f"{pid.lower()}.nuspec").write_text(
+ f'<?xml version="1.0"?><package xmlns="http://schemas.microsoft.com/packaging/2013/05/nuspec.xsd">'
+ f"<metadata><id>{pid}</id><version>{ver}</version>{licel}<copyright>{copy}</copyright></metadata></package>")
+
+
+def project(repo: Path, sln: str, rel: str, pkgs: dict[str, str]):
+ """pkgs: "Id/ver" -> "runtime" | "build" (analyzer, nothing in the output)."""
+ p = repo / rel
+ f = repo / sln
+ line = f'<Project Path="{rel}/{p.name}.csproj" />'
+ old = f.read_text() if f.exists() else ""
+ if line not in old:
+ f.write_text(old + line + "\n")
+ (p / "obj").mkdir(parents=True, exist_ok=True)
+ (p / f"{p.name}.csproj").write_text("<Project/>")
+ targets = {k: ({"type": "package", "runtime": {"lib/net10.0/X.dll": {}}} if kind == "runtime"
+ else {"type": "package", "runtime": {"lib/netstandard2.0/_._": {}}})
+ for k, kind in pkgs.items()}
+ (p / "obj" / "project.assets.json").write_text(json.dumps({
+ "targets": {"net10.0": targets},
+ "libraries": {k: {"type": "package"} for k in pkgs}}))
+
+
+class LicensesTest(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.TemporaryDirectory()
+ self.repo = Path(self.tmp.name) / "repo"
+ self.cache = Path(self.tmp.name) / "cache"
+ self.repo.mkdir()
+ self.cache.mkdir()
+ self.cfg = gl.Config("Game.slnx", ("src/Game/",), "# Licences\n",
+ extra=[("runtime", "10", "MIT", "© R", "all")],
+ planned=[("Later", "2.0", "BSD|0", "managed", "nuspec")],
+ command="tool")
+ (self.repo / "Game.slnx").write_text("")
+
+ def tearDown(self):
+ self.tmp.cleanup()
+
+ def run_main(self, *argv) -> tuple[int, str]:
+ err = io.StringIO()
+ with contextlib.redirect_stderr(err):
+ code = gl.main(list(argv), self.cfg, self.repo, self.cache)
+ return code, err.getvalue()
+
+ def test_write_then_check_is_clean_and_splits_shipped_from_test(self):
+ nuspec(self.cache, "Pack", "1.0.0", "MIT", "© P")
+ nuspec(self.cache, "xunit", "2.9.2", "Apache-2.0")
+ project(self.repo, "Game.slnx", "src/Game", {"Pack/1.0.0": "runtime"})
+ project(self.repo, "Game.slnx", "src/Game.Tests", {"xunit/2.9.2": "runtime", "Pack/1.0.0": "runtime"})
+ self.assertEqual(self.run_main("--write")[0], 0)
+ self.assertEqual(self.run_main(), (0, ""))
+ md = (self.repo / "licenses.md").read_text()
+ self.assertTrue(md.startswith("# Licences\n"))
+ shipped, rest = md.split("## Not a package")
+ self.assertIn("| Pack | 1.0.0 | MIT | © P | Game | |", shipped)
+ self.assertNotIn("xunit", shipped)
+ self.assertIn("| runtime | 10 | MIT | © R | all |", rest)
+ self.assertIn("| Later | 2.0 | BSD\\|0 | managed | nuspec |", rest)
+ self.assertIn("| xunit | 2.9.2 | Apache-2.0 | Game.Tests |", rest)
+
+ def test_stale_names_the_command(self):
+ self.assertEqual(self.run_main(), (1, "licenses.md is stale: run tool --write\n"))
+
+ def test_copyleft_ships_only_when_allowed(self):
+ nuspec(self.cache, "OpenAL.Native", "1.0.0", "LGPL-2.0-or-later")
+ project(self.repo, "Game.slnx", "src/Game.Tests", {"OpenAL.Native/1.0.0": "runtime"})
+ self.assertEqual(self.run_main("--write")[0], 0) # test project: not shipped
+ project(self.repo, "Game.slnx", "src/Game", {"OpenAL.Native/1.0.0": "runtime"})
+ self.assertEqual(self.run_main("--write"), (1, "NuGet OpenAL.Native 1.0.0: copyleft licence LGPL-2.0-or-later ships\n"))
+ self.cfg.allow_shipped["OpenAL.Native"] = "shared library"
+ self.assertEqual(self.run_main("--write")[0], 0)
+ self.assertIn("| shared library |", (self.repo / "licenses.md").read_text())
+
+ def test_build_only_package_is_not_shipped(self):
+ nuspec(self.cache, "Analyzer", "1.0.0", "GPL-3.0-only")
+ project(self.repo, "Game.slnx", "src/Game", {"Analyzer/1.0.0": "build"})
+ self.assertEqual(self.run_main("--write")[0], 0)
+
+ def test_unknown_licence_fails_until_overridden(self):
+ nuspec(self.cache, "FileLicensed", "1.0.0", None)
+ project(self.repo, "Game.slnx", "src/Game.Tests", {"FileLicensed/1.0.0": "runtime"})
+ self.assertEqual(self.run_main("--write")[0], 1)
+ self.cfg.overrides["FileLicensed"] = "MIT"
+ self.assertEqual(self.run_main("--write")[0], 0)
+
+ def test_projects_come_from_the_configured_solution(self):
+ nuspec(self.cache, "GplLib", "1.0.0", "GPL-3.0-only")
+ project(self.repo, "Other.slnx", "src/Game", {"GplLib/1.0.0": "runtime"})
+ self.assertEqual(self.run_main("--write")[0], 0)
+ self.cfg.solution = "Other.slnx"
+ self.assertEqual(self.run_main("--write")[0], 1)
+
+ def test_missing_assets_fails(self):
+ (self.repo / "src/Game").mkdir(parents=True)
+ (self.repo / "Game.slnx").write_text('<Project Path="src/Game/Game.csproj" />')
+ self.assertEqual(self.run_main("--write")[0], 1)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/scripts/test_thread_cpu.py b/scripts/test_thread_cpu.py
new file mode 100644
index 0000000..f60a2d0
--- /dev/null
+++ b/scripts/test_thread_cpu.py
@@ -0,0 +1,24 @@
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import thread_cpu as tc
+
+
+class ThreadCpuTests(unittest.TestCase):
+ def test_groups_by_name_and_counts_new_threads_from_zero(self):
+ before = {"1": ("main", 100), "2": ("worker", 10)}
+ after = {"1": ("main", 150), "2": ("worker", 20), "3": ("worker", 30)}
+ rows = tc.per_name(before, after, seconds=1.0, hz=100)
+ self.assertEqual(rows[0], ("main", 50.0, 1))
+ self.assertEqual(rows[1], ("worker", 40.0, 2)) # 10 + 30 ticks
+
+ def test_samples_this_process(self):
+ import os
+ threads = tc.sample(os.getpid())
+ self.assertIn(str(os.getpid()), threads)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/scripts/test_thread_instr.py b/scripts/test_thread_instr.py
new file mode 100644
index 0000000..d23e440
--- /dev/null
+++ b/scripts/test_thread_instr.py
@@ -0,0 +1,21 @@
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import thread_instr as ti
+
+
+class ThreadInstrTests(unittest.TestCase):
+ def test_sums_threads_by_name(self):
+ text = "\n".join([
+ "Game.Desk-100,5000,,instructions:u,100,100.00,,",
+ "audio-101,300,,instructions:u,100,100.00,,",
+ "audio-102,200,,instructions:u,100,100.00,,",
+ "gl0-103,<not counted>,,instructions:u,0,0.00,,",
+ ])
+ self.assertEqual(ti.parse(text), {"Game.Desk": 5000, "audio": 500})
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/scripts/thread_cpu.py b/scripts/thread_cpu.py
new file mode 100755
index 0000000..f22ad51
--- /dev/null
+++ b/scripts/thread_cpu.py
@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+"""Per-thread CPU of a process over an interval, grouped by thread name (Linux /proc).
+Shows where a smoke run's CPU goes: UI/render thread, audio thread, OpenAL mixer,
+compositor, GC... (smoke_run.sh step `threads:<s>`).
+
+Usage: scripts/thread_cpu.py PID SECONDS [--top N]
+"""
+import argparse
+import os
+import time
+from collections import defaultdict
+
+
+def sample(pid):
+ """{tid: (name, cpu_ticks)} for every live thread of pid."""
+ out = {}
+ base = f"/proc/{pid}/task"
+ for tid in os.listdir(base):
+ try:
+ with open(f"{base}/{tid}/stat") as f:
+ stat = f.read()
+ except OSError:
+ continue # thread ended
+ name = stat[stat.index("(") + 1: stat.rindex(")")]
+ fields = stat[stat.rindex(")") + 2:].split()
+ out[tid] = (name, int(fields[11]) + int(fields[12])) # utime + stime
+ return out
+
+
+def per_name(before, after, seconds, hz):
+ """[(name, percent of one core, thread count)] busiest first; threads that started
+ during the interval count from zero."""
+ total = defaultdict(float)
+ count = defaultdict(int)
+ for tid, (name, ticks) in after.items():
+ start = before.get(tid, (name, 0))[1]
+ total[name] += (ticks - start) * 100.0 / hz / seconds
+ count[name] += 1
+ return sorted(((n, total[n], count[n]) for n in total), key=lambda r: -r[1])
+
+
+def main():
+ ap = argparse.ArgumentParser()
+ ap.add_argument("pid", type=int)
+ ap.add_argument("seconds", type=float)
+ ap.add_argument("--top", type=int, default=10)
+ args = ap.parse_args()
+ hz = os.sysconf("SC_CLK_TCK")
+ a = sample(args.pid)
+ time.sleep(args.seconds)
+ b = sample(args.pid)
+ rows = per_name(a, b, args.seconds, hz)
+ print(f"threads: {sum(r[1] for r in rows):.1f}% total over {args.seconds:g}s")
+ for name, pct, n in rows[: args.top]:
+ if pct >= 0.05:
+ print(f" {pct:5.1f}% {name}" + (f" ×{n}" if n > 1 else ""))
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/thread_instr.py b/scripts/thread_instr.py
new file mode 100755
index 0000000..846d27f
--- /dev/null
+++ b/scripts/thread_instr.py
@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+"""User-space instructions per thread name over an interval (perf stat --per-thread) —
+a load-independent cost measure: CPU % swings with other work on the machine and
+clock speed, retired instructions don't (smoke_run.sh step `instr:<s>`).
+
+Usage: scripts/thread_instr.py PID SECONDS
+"""
+import re
+import subprocess
+import sys
+from collections import defaultdict
+
+
+def parse(text):
+ """{thread name: instructions} from `perf stat --per-thread -x,` output."""
+ out = defaultdict(int)
+ for line in text.splitlines():
+ parts = line.split(",")
+ if len(parts) < 4 or "instructions" not in parts[3]:
+ continue
+ try:
+ count = int(parts[1])
+ except ValueError:
+ continue # <not counted>
+ out[re.sub(r"-\d+$", "", parts[0].strip())] += count
+ return out
+
+
+def main():
+ pid, seconds = sys.argv[1], float(sys.argv[2])
+ res = subprocess.run(["perf", "stat", "--per-thread", "-x,", "-e", "instructions:u", "-p", pid,
+ "--", "sleep", str(seconds)], capture_output=True, text=True)
+ counts = parse(res.stderr)
+ total = sum(counts.values())
+ print(f"instr: {total / seconds / 1e6:.0f} M/s total over {seconds:g}s")
+ for name, n in sorted(counts.items(), key=lambda kv: -kv[1])[:8]:
+ if n / seconds >= 1e6:
+ print(f" {n / seconds / 1e6:7.0f} M/s {name}")
+
+
+if __name__ == "__main__":
+ main()