"""The Logbook store and its automatic commit log.""" import json import shutil import subprocess import unittest from helpers import HarborCase from harbor import logbook, logbook_auto, progress_map from harbor.queue import FunctionQueue from harbor.util import HarborError class LogbookStore(HarborCase): def setUp(self): super().setUp() self.h.project.logbook = {"dir": str(self.dir / "logbook")} def test_entries_stats_and_milestones(self): e = logbook.add_entry(self.h.project, "First", "Something happened.", kind="tooling", at="2026-10-05T10:00:00Z") self.assertEqual(e["id"], "20261005-1000-1") with self.assertRaises(HarborError): logbook.add_entry(self.h.project, "x", "y", kind="gossip") logbook.set_stats(self.h.project, {"functions": 10, "classified_percent": 50.0, "weighted_percent": 5.0}) logbook.upsert_milestone(self.h.project, {"id": "m1", "order": 1, "code": "M1", "title": "T", "items": []}) data = logbook.read_all(self.h.project) self.assertEqual((data["log"][0]["title"], data["stats"]["functions"], data["history"][0]["classified"], data["milestones"][0]["code"]), ("First", 10, 50.0, "M1")) class LogbookAuto(HarborCase): def setUp(self): super().setUp() self.repo = self.dir / "repo" self.repo.mkdir() git = lambda *a: subprocess.run(["git", "-C", str(self.repo), *a], check=True, capture_output=True) git("init", "-q") git("config", "user.name", "T") git("config", "user.email", "t@example.com") (self.repo / "a.txt").write_text("1", encoding="utf-8") git("add", "-A") git("commit", "-q", "-m", "first") self.first = subprocess.run(["git", "-C", str(self.repo), "rev-parse", "HEAD"], capture_output=True, text=True).stdout.strip() (self.repo / "annotations").mkdir() (self.repo / "annotations" / "names.csv").write_text("x", encoding="utf-8") git("add", "-A") git("commit", "-q", "-m", "Name things", "-m", "Body line one.\nCo-Authored-By: someone") self.h.project.logbook = {"dir": str(self.dir / "logbook"), "decomp": str(self.repo)} def test_new_commits_become_entries_once(self): logbook_auto._save_state(self.h.project, {"last_head": self.first}) out = logbook_auto.refresh(self.h.project) self.assertEqual(out["entries"], 1) entry = logbook.read_all(self.h.project)["log"][0] self.assertEqual((entry["title"], entry["kind"]), ("Name things", "naming")) self.assertIn("Body line one.", entry["detail"]) self.assertNotIn("Co-Authored-By", entry["detail"]) self.assertTrue(entry["at"].endswith("Z")) self.assertFalse(logbook_auto.refresh(self.h.project)["changed"]) def test_waits_for_the_lander_and_a_failed_stats_run_logs_nothing_twice(self): import os import time from unittest import mock logbook_auto._save_state(self.h.project, {"last_head": self.first}) lock = self.dir / "lander-busy.lock" lock.write_text("x", encoding="utf-8") self.h.project.logbook.update(lander_lock=str(lock), stats_repo=str(self.dir / "stats")) out = logbook_auto.refresh(self.h.project) self.assertFalse(out["changed"]) self.assertIn("lander", out["waiting"]) self.assertEqual(logbook.read_all(self.h.project)["log"], []) old = time.time() - 61 * 60 os.utime(lock, (old, old)) # stale: ignored with mock.patch.object(logbook_auto, "rebuild_stats", side_effect=RuntimeError("timed out")): out = logbook_auto.refresh(self.h.project) self.assertEqual((out["entries"], out["stats_error"]), (1, "timed out")) again = logbook_auto.refresh(self.h.project) # the statistics are retried ... self.assertIn("stats_error", again) self.assertNotIn("entries", again) # ... but the commit is not relogged self.assertEqual(len(logbook.read_all(self.h.project)["log"]), 1) class LanderGate(unittest.TestCase): def test_run_low_lowers_priority_and_ends_the_tree_on_timeout(self): import os import sys import time from harbor import lander_gate script = "import subprocess, sys, time; subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)']);" \ " print('ok', flush=True); time.sleep(60)" from unittest import mock seen = {} real = subprocess.Popen def popen(argv, **kw): seen.update(kw) return real([sys.executable, "-c", "pass"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) with mock.patch.object(lander_gate.subprocess, "Popen", popen): self.assertEqual(lander_gate.run_low(["x"], 60).returncode, 0) if os.name == "nt": self.assertTrue(seen["creationflags"] & subprocess.BELOW_NORMAL_PRIORITY_CLASS) start = time.monotonic() with self.assertRaises(subprocess.TimeoutExpired): lander_gate.run_low([sys.executable, "-c", script], 3) self.assertLess(time.monotonic() - start, 45) # the grandchild did not hold the pipes class ProgressMapTests(HarborCase): """GET /api/logbook/progress-map: units, tiers and the headline, by the Logbook's own definitions.""" def setUp(self): super().setUp() d = self.dir (d / "image.bin").write_bytes(b"\xc3" * 16) facts = [{"address": "0x401000", "size": 100, "unit": "", "data": ["0x500010", "0x500020"]}, {"address": "0x401100", "size": 50, "unit": ""}, {"address": "0x401200", "size": 30, "unit": ""}, {"address": "0x402000", "size": 40, "unit": "engine/NiFoo"}, {"address": "0x403000", "size": 20, "unit": ""}, {"address": "0x404000", "size": 10, "unit": ""}] (d / "facts.jsonl").write_text("\n".join(json.dumps(f) for f in facts), encoding="utf-8") (d / "sections.json").write_text(json.dumps([{"name": ".text", "start": 0x401000, "end": 0x500000}, {"name": ".data", "start": 0x500000, "end": 0x501000}]), encoding="utf-8") ann = d / "annotations" ann.mkdir() (ann / "names.csv").write_text("address,name,category,note\n0x00401000,First,game,x\n", encoding="utf-8") (ann / "globals.csv").write_text("address,name,type,comment\n0x00500010,g_one,int,x\n", encoding="utf-8") symbols = d / "stats" / "re" / "symbols" symbols.mkdir(parents=True) rows = [("0x401000", "First", "game", "decompiled"), ("0x401100", "Second", "game", "decompiled"), ("0x401200", "", "game", "classified"), ("0x402000", "NiFoo_Bar", "engine", "classified"), ("0x403000", "", "platform", "classified"), ("0x404000", "", "unsorted", "unknown")] (symbols / "pc102.csv").write_text("address,end,size,name,category,unit,status,xbox,notes\n" + "".join( f"{a},,0,{n},{c},,{s},,\n" for a, n, c, s in rows), encoding="utf-8") (d / "check.json").write_text(json.dumps([{"address": "0x401000", "status": "exact"}]), encoding="utf-8") (d / "diffrun.json").write_text(json.dumps([{"address": "0x401100", "verdict": "same"}]), encoding="utf-8") (d / "sources.csv").write_text("pc_address,pc_name,source_file,lines,xbox_address,confidence,kind,evidence\n" "0x401000,,a.cpp,,,high,,\n0x401100,,a.cpp,,,high,,\n", encoding="utf-8") self.h.project.reverse = {"image": str(d / "image.bin"), "facts": str(d / "facts.jsonl"), "annotations": str(ann), "match_results": str(d / "check.json"), "diffrun_results": str(d / "diffrun.json"), "source_files": str(d / "sources.csv")} self.h.project.logbook = {"dir": str(d / "logbook"), "stats_repo": str(d / "stats")} logbook.set_stats(self.h.project, {"functions": 6, "commit": "abc1234", "game_bytes_decompiled_percent": 83.33, "by_category": {"game": {"functions": 3, "bytes": 180}}}) self.old_ttl, progress_map.KEY_TTL = progress_map.KEY_TTL, 0 self.map = progress_map.ProgressMap(self.h.project, lambda: FunctionQueue(self.h)) def tearDown(self): progress_map.KEY_TTL = self.old_ttl progress_map._MAPS.clear() super().tearDown() def test_headline_units_tiers_and_cache(self): doc = self.map.get() self.assertEqual((doc["percent"], doc["logbook_percent"]), (83.33, 83.33)) # 150 of 180 game bytes self.assertEqual(doc["code"], {"exact": 100, "verified": 50, "readable": 0, "bytes": 180, "functions": 3, "decompiled_functions": 2}) self.assertEqual((doc["data"]["named"], doc["data"]["addresses"]), (1, 2)) self.assertEqual(doc["commit"]["sha"], "abc1234") self.assertEqual(doc["history"][-1][2:], ["abc1234", 180]) # history points keep the commit units = {u[0]: dict(zip(doc["unit_fields"], u)) for u in doc["units"]} self.assertEqual(set(units), {"a.cpp", "game · 0x400000 block (no source file)", "engine/NiFoo", "platform · 0x400000 block (no source file)", "unsorted · 0x400000 block (no source file)"}) self.assertEqual((units["a.cpp"]["kind"], units["a.cpp"]["counted"], units["a.cpp"]["exact"]), ("game", 150, 100)) self.assertEqual(units["engine/NiFoo"]["kind"], "engine") status = {f[1]: doc["statuses"][f[3]] for f in doc["functions"]} self.assertEqual(status, {0x401000: "exact", 0x401100: "verified", 0x401200: "raw", 0x402000: "engine", 0x403000: "library", 0x404000: "unsorted"}) self.assertEqual({f[1]: f[4] for f in doc["functions"]}[0x401000], "First") self.assertEqual(self.map.get(doc["version"]), {"version": doc["version"], "same": True}) self.assertEqual(self.map.computations, 1) FunctionQueue(self.h).defer("Claude", "0x401200", "x87", "precision differs") # a new note: recomputed doc2 = self.map.get(doc["version"]) self.assertNotEqual(doc2["version"], doc["version"]) self.assertEqual(doc2["statuses"][next(f for f in doc2["functions"] if f[1] == 0x401200)[3]], "deferred") def test_endpoint(self): progress_map._MAPS[self.h.project.name] = self.map status, body = progress_map.handle(self.h.project, {}) self.assertEqual((status, body["percent"]), (200, 83.33)) self.assertEqual(progress_map.handle(self.h.project, {"have": body["version"]})[1]["same"], True) self.h.project.reverse = {} self.assertEqual(progress_map.handle(self.h.project, {})[0], 404) if __name__ == "__main__": unittest.main()