"""Tests for scripts/buglog.py (one Markdown record per bug). Run: python3 -m unittest discover -s scripts/tests -v """ import json import os import subprocess import sys import tempfile import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import buglog # noqa: E402 FULL_ENV = """\ - OS: Ubuntu 24.04 LTS; kernel Linux 6.18.33 x86_64 - libc: glibc 2.42 (from the binary's interpreter) - Toolchain: gcc 15.2.0, meson 1.9.1 - Build: asan (xce.build) - Code: scheduler-framework @ {sha} - Deployment: single host, xcecd_virtual""" FULL_BODY = """\ # {id}: xcecd_virtual leaks worker strings on every finished job ## Description **Observed:** ``` Direct leak of 12 byte(s) in 12 object(s) ``` **Expected:** no leak at exit **Impact:** grows with job count **Found by:** LeakSanitizer after a job burst ## Environment {env} ## Reproduction 1. `/xce-build worker` 2. run `finished_job_retention_test` - Frequency: deterministic - Evidence: the unit test ## Root cause {root} ## Fix {fix} ## Verification {verif} ## References - Related: none """ def git(root, *args): return subprocess.run(["git", "-C", str(root), *args], check=True, capture_output=True, text=True).stdout.strip() class BuglogTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) (self.root / ".wolf" / "bugs").mkdir(parents=True) git(self.root, "init", "-q") git(self.root, "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-q", "--allow-empty", "-m", "init") self.sha = git(self.root, "rev-parse", "--short=8", "HEAD") buglog.configure(self.root) def tearDown(self): self.tmp.cleanup() def write_record(self, bug_id="bug-100", status="open", legacy=False, root="TODO", fix="TODO", verif="TODO", commits=None, env=None, extra_meta=None): meta = {"id": bug_id, "status": status, "status_updated": "2026-09-27", "status_note": "" if status == "fixed" else "why", "component": "worker", "severity": "leak", "timestamp": "2026-09-25T15:42:02Z"} if legacy: meta["legacy"] = True if commits is not None: meta["fix_commits"] = commits meta.update(extra_meta or {}) body = FULL_BODY.format(id=bug_id, env=env if env is not None else FULL_ENV.format(sha=self.sha), root=root, fix=fix, verif=verif) text = buglog.render_meta(meta) + "\n" + body path = self.root / ".wolf" / "bugs" / f"{bug_id}.md" path.write_text(text, encoding="utf-8") return path class ParseRender(BuglogTestCase): def test_round_trip_keeps_every_field(self): path = self.write_record(extra_meta={"tags": ["worker", "asan"], "gitea_issue": 285}) rec = buglog.load("bug-100") self.assertEqual(rec.meta["tags"], ["worker", "asan"]) self.assertEqual(rec.meta["gitea_issue"], 285) self.assertEqual(rec.summary, "xcecd_virtual leaks worker strings on every finished job") self.assertIn("Direct leak", rec.sections["Description"]) self.assertEqual(buglog.render(rec), path.read_text(encoding="utf-8")) def test_section_header_inside_code_fence_is_not_a_section(self): self.write_record(root="See:\n```\n## Fix\nnot a header\n```") rec = buglog.load("bug-100") self.assertIn("not a header", rec.sections["Root cause"]) self.assertEqual(rec.sections["Fix"], "TODO") def test_escaped_section_header_round_trips(self): text = "a\n## Fix\nb" rec = buglog.Record({"id": "bug-7", "status": "open"}, "s", {"Description": buglog.escape_text(text)}) again = buglog.parse(buglog.render(rec)) self.assertEqual(buglog.unescape_text(again.sections["Description"]), text) self.assertNotIn("Fix", again.sections) def test_set_meta_leaves_body_bytes_unchanged(self): path = self.write_record() body_before = path.read_text(encoding="utf-8").split("\n---\n", 1)[1] buglog.set_meta("bug-100", {"gitea_issue": 9, "status_note": "x: y"}) text = path.read_text(encoding="utf-8") self.assertEqual(text.split("\n---\n", 1)[1], body_before) self.assertEqual(buglog.load("bug-100").meta["status_note"], "x: y") class Check(BuglogTestCase): def problems(self, bug_id="bug-100", target=None): return buglog.check(buglog.load(bug_id), target_status=target) def test_complete_open_record_passes(self): self.write_record() self.assertEqual(self.problems(), []) def test_environment_needs_every_item(self): env = FULL_ENV.format(sha=self.sha).replace( "- libc: glibc 2.42 (from the binary's interpreter)\n", "") self.write_record(env=env) self.assertTrue(any("libc" in p for p in self.problems())) def test_libc_needs_a_version(self): env = FULL_ENV.format(sha=self.sha).replace("glibc 2.42", "glibc") self.write_record(env=env) self.assertTrue(any("libc" in p for p in self.problems())) def test_code_needs_a_commit(self): env = FULL_ENV.format(sha=self.sha).replace(self.sha, "latest") self.write_record(env=env) self.assertTrue(any("Code" in p for p in self.problems())) def test_todo_counts_as_missing(self): env = FULL_ENV.format(sha=self.sha).replace( "single host, xcecd_virtual", "TODO: topology") self.write_record(env=env) self.assertTrue(any("Deployment" in p for p in self.problems())) def test_fixed_needs_root_cause_fix_verification_and_commit(self): self.write_record(status="fixed") probs = " | ".join(self.problems()) for word in ("Root cause", "Fix", "Verification", "fix_commits"): self.assertIn(word, probs) def test_fixed_with_everything_passes(self): self.write_record(status="fixed", root="remove_worker skips five strings", fix="free them", commits=[self.sha], verif="- Test: test_x: RED before, GREEN after\n" "- Suites: worker 12/12, asan") self.assertEqual(self.problems(), []) def test_unknown_commit_is_rejected(self): self.write_record(status="fixed", root="r", fix="f", commits=["deadbeef"], verif="- Test: t: RED before, GREEN after\n- Suites: s") self.assertTrue(any("deadbeef" in p for p in self.problems())) def test_open_status_needs_note(self): self.write_record(extra_meta={"status_note": ""}) self.assertTrue(any("status_note" in p for p in self.problems())) def test_legacy_record_is_exempt_until_it_changes_status(self): self.write_record(legacy=True, env=buglog.NOT_RECORDED) self.assertEqual(self.problems(), []) # marking it fixed now demands the fix-time sections probs = " | ".join(self.problems(target="fixed")) self.assertIn("Verification", probs) self.assertIn("fix_commits", probs) def test_migrated_placeholder_is_not_verification(self): self.write_record(legacy=True, root="r", fix="f", commits=[self.sha], verif=buglog.NOT_RECORDED) self.assertTrue(any("Verification" in p for p in self.problems(target="fixed"))) class Commands(BuglogTestCase): def run_cli(self, *argv): try: buglog.main(list(argv)) return 0 except SystemExit as e: return e.code def test_status_refuses_incomplete_fixed_and_leaves_file_alone(self): path = self.write_record() before = path.read_text(encoding="utf-8") rc = self.run_cli("status", "bug-100", "fixed", "--commit", self.sha) self.assertNotEqual(rc, 0) self.assertEqual(path.read_text(encoding="utf-8"), before) def test_status_fixed_records_commits(self): self.write_record(root="r", fix="f", verif="- Test: t: RED before, GREEN after\n- Suites: s") rc = self.run_cli("status", "bug-100", "fixed", "--commit", self.sha) self.assertIn(rc, (0, None)) meta = buglog.load("bug-100").meta self.assertEqual(meta["status"], "fixed") self.assertEqual(meta["fix_commits"], [self.sha]) def test_new_creates_record_with_environment_filled(self): self.run_cli("new", "--summary", "scheduler hangs on empty queue", "--component", "scheduler", "--severity", "hang") rec = buglog.load("bug-001") self.assertEqual(rec.meta["status"], "open") env = rec.sections["Environment"] self.assertRegex(env, r"- OS: .*Linux") self.assertRegex(env, r"- Code: .*" + self.sha) # the author still has to write the description self.assertTrue(buglog.check(rec)) def test_next_id_counts_archive(self): self.write_record(bug_id="bug-100") (self.root / ".wolf" / "buglog-archive.json").write_text( json.dumps({"bugs": [{"id": "bug-900"}]}), encoding="utf-8") self.assertEqual(buglog.next_id(), "bug-901") def test_index_keeps_openwolf_fields(self): self.write_record(extra_meta={"files": ["src/worker/a.c", "src/worker/b.c"], "tags": ["worker"]}, root="cause text") buglog.build_index() idx = json.loads((self.root / ".wolf" / "buglog.json").read_text()) b = idx["bugs"][0] self.assertEqual(b["id"], "bug-100") self.assertTrue(b["error_message"].startswith("xcecd_virtual leaks")) self.assertIn("Direct leak", b["error_message"]) self.assertEqual(b["root_cause"], "cause text") self.assertEqual(b["file"], "src/worker/a.c, src/worker/b.c") self.assertEqual(b["tags"], ["worker"]) self.assertEqual(b["status"], "open") def test_write_commands_refresh_the_index(self): self.write_record() self.run_cli("note", "bug-100", "seen again under load") idx = json.loads((self.root / ".wolf" / "buglog.json").read_text()) self.assertIn("seen again under load", idx["bugs"][0]["error_message"]) class Staged(BuglogTestCase): """`check --staged`, which the pre-commit hook runs.""" def commit(self, *paths): git(self.root, "add", *paths) git(self.root, "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-q", "-m", "x") def check_staged(self): try: buglog.main(["check", "--staged"]) return 0 except SystemExit as e: return e.code or 0 def test_complete_record_with_fresh_index_passes(self): self.write_record() buglog.build_index() git(self.root, "add", ".wolf") self.assertEqual(self.check_staged(), 0) def test_incomplete_record_is_refused(self): self.write_record(env="- OS: TODO") buglog.build_index() git(self.root, "add", ".wolf") self.assertNotEqual(self.check_staged(), 0) def test_stale_index_is_refused(self): self.write_record() git(self.root, "add", ".wolf") self.assertNotEqual(self.check_staged(), 0) def test_hand_edited_index_alone_is_refused(self): self.write_record() buglog.build_index() self.commit(".wolf") idx = self.root / ".wolf" / "buglog.json" idx.write_text(idx.read_text().replace("leaks", "LEAKS")) git(self.root, "add", ".wolf/buglog.json") self.assertNotEqual(self.check_staged(), 0) def test_legacy_record_marked_fixed_by_hand_needs_fix_sections(self): self.write_record(legacy=True, env=buglog.NOT_RECORDED) buglog.build_index() self.commit(".wolf") buglog.set_meta("bug-100", {"status": "fixed", "status_note": ""}) git(self.root, "add", ".wolf") self.assertNotEqual(self.check_staged(), 0) def test_migrated_record_matching_the_old_index_passes(self): # the migration commit: the record is new, but HEAD's buglog.json # already had this bug with the same status (self.root / ".wolf" / "buglog.json").write_text( json.dumps({"bugs": [{"id": "bug-100", "status": "fixed"}]})) self.commit(".wolf") self.write_record(legacy=True, status="fixed", env=buglog.NOT_RECORDED) buglog.build_index() git(self.root, "add", ".wolf") self.assertEqual(self.check_staged(), 0) def test_new_record_cannot_claim_legacy(self): self.write_record(legacy=True, status="fixed", env=buglog.NOT_RECORDED) buglog.build_index() git(self.root, "add", ".wolf") self.assertNotEqual(self.check_staged(), 0) def test_untouched_legacy_record_passes(self): self.write_record(legacy=True, env=buglog.NOT_RECORDED) buglog.build_index() self.commit(".wolf") buglog.set_meta("bug-100", {"gitea_issue": 5}) git(self.root, "add", ".wolf") self.assertEqual(self.check_staged(), 0) if __name__ == "__main__": unittest.main()