forked from Keelrleek/Harbor
190 lines
11 KiB
Python
190 lines
11 KiB
Python
"""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()
|