import json import os import shutil import subprocess import tempfile import threading import unittest import urllib.request from http.server import ThreadingHTTPServer from hdc.assistant import INTENTS, Assistant from hdc.terminal import Kernel, Terminal from hdc.webui import App, make_handler from hdc.x86 import Boot D = 2000 class TestTerminal(unittest.TestCase): def test_wrap_and_scroll(self): term = Terminal(cols=10, rows=3) term.write("abcdefghijK") self.assertEqual(term.cursor[0], 1) self.assertEqual(term.screen()[0], "abcdefghij") self.assertEqual(term.screen()[1].rstrip(), "K") def test_newline_and_carriage_return(self): term = Terminal() term.write("ab\rc\r\nde") self.assertEqual(term.screen()[0], "cb") self.assertEqual(term.screen()[1].rstrip(), "de") def test_clear_resets(self): term = Terminal() term.write("something") term.clear() self.assertTrue(all(not line.strip() for line in term.screen())) self.assertEqual(term.cursor, [0, 0]) def test_cells_carry_attributes(self): term = Terminal(cols=4, rows=1) term.write("ab") cells = term.cells() self.assertEqual(cells[0][0], ("a", "green")) class TestKernel(unittest.TestCase): def setUp(self): self.kernel = Kernel(d=D) boot = Boot() boot.post() boot.boot() self.kernel.boot(boot) def test_boot_prints_a_banner(self): self.assertTrue(self.kernel.booted) self.assertIn("HDCSTORAGE kernel", "\n".join(self.kernel.term.screen())) def test_asm_then_run_outputs_42(self): self.kernel.execute("asm") output = self.kernel.execute("run") self.assertIn("42", output) self.assertIn("halted", output) def test_step_reports_one_instruction(self): self.kernel.execute("asm") before = self.kernel.machine.pc self.kernel.execute("step") self.assertEqual(self.kernel.machine.pc, before + 1) def test_unknown_command_is_reported_not_raised(self): output = self.kernel.execute("frobnicate") self.assertIn("unknown command", output) def test_clear_removes_previous_output(self): self.kernel.execute("help") self.assertIn("commands (the pipeline", "\n".join(self.kernel.term.screen())) self.kernel.execute("clear") self.assertNotIn("commands (the pipeline", "\n".join(self.kernel.term.screen())) def test_help_lists_the_pipeline(self): self.assertIn("asm", self.kernel.execute("help")) def test_boot_command_replays_the_bios(self): output = self.kernel.execute("boot") self.assertIn("HDCSTORAGE BIOS", output) def test_stat_reports_no_error(self): output = self.kernel.execute("stat") self.assertNotIn("error", output) self.assertIn("cells=", output) class TestAssistant(unittest.TestCase): def setUp(self): self.kernel = Kernel(d=D) self.assistant = Assistant(self.kernel, d=D) def test_known_utterances_classify(self): intent, score = self.assistant.classify("run the program now") self.assertEqual(intent, "run") self.assertGreater(score, 0.3) def test_nonsense_is_rejected(self): intent, score = self.assistant.classify("purple tractor umbrella") self.assertLess(score, self.assistant.threshold) def test_chat_executes_the_command(self): record = self.assistant.chat("show the registers") self.assertEqual(record["executed"], "regs") self.assertTrue(record["output"]) def test_chat_history_grows(self): self.assistant.chat("help") self.assistant.chat("run the program") self.assertEqual(len(self.assistant.history), 2) def test_every_intent_has_examples_and_a_command(self): for name, (examples, command) in INTENTS.items(): self.assertTrue(examples, msg=name) self.assertIsInstance(command, str, msg=name) def test_report_mentions_the_encoder(self): self.assertIn("intents", self.assistant.report()) class TestApp(unittest.TestCase): @classmethod def setUpClass(cls): cls.app = App(d=D) cls.server = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(cls.app)) cls.port = cls.server.server_address[1] cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True) cls.thread.start() @classmethod def tearDownClass(cls): cls.server.shutdown() cls.server.server_close() def get(self, path): url = "http://127.0.0.1:%d%s" % (self.port, path) with urllib.request.urlopen(url, timeout=60) as response: return response.status, response.read() def post(self, path, payload): url = "http://127.0.0.1:%d%s" % (self.port, path) request = urllib.request.Request( url, json.dumps(payload).encode(), {"Content-Type": "application/json"}) with urllib.request.urlopen(request, timeout=120) as response: return response.status, json.load(response) def test_page_is_served(self): status, body = self.get("/") self.assertEqual(status, 200) self.assertIn(b"CRT", body) self.assertIn(b"/api/exec", body) def test_state_has_boot_and_screen(self): status, body = self.get("/api/state") state = json.loads(body) self.assertEqual(status, 200) self.assertTrue(state["boot"]) self.assertEqual(len(state["screen"]), 25) def test_exec_runs_a_command(self): status, result = self.post("/api/exec", {"line": "asm"}) self.assertEqual(status, 200) self.assertIn("LDI", result["output"]) def test_autonomy_state_reports_the_readahead(self): status, body = self.get("/api/autonomy") stats = json.loads(body)["stats"] self.assertEqual(status, 200) for key in ("graded", "hits", "accuracy", "chance", "readahead", "autonomous_actions", "cached_commands"): self.assertIn(key, stats) def test_autonomy_can_be_toggled(self): status, body = self.post("/api/autonomy", {"enabled": True}) self.assertEqual(status, 200) self.assertTrue(json.loads(json.dumps(body))["stats"]["enabled"]) status, body = self.post("/api/autonomy", {"enabled": False}) self.assertFalse(json.loads(json.dumps(body))["stats"]["enabled"]) def test_chat_grades_the_next_turn(self): for _ in range(4): status, body = self.post("/api/chat", {"text": "show the statistics"}) graded = json.loads(json.dumps(body)).get("graded") self.assertIn("predicted", graded) self.assertIn("hit", graded) def test_chat_executes_and_returns_screen(self): status, result = self.post("/api/chat", {"text": "show the tools"}) self.assertEqual(status, 200) self.assertEqual(result["executed"], "tools") self.assertEqual(len(result["screen"]), 25) def test_pipeline_lists_every_stage(self): status, body = self.get("/api/pipeline") self.assertEqual(status, 200) stages = json.loads(body)["stages"] self.assertGreaterEqual(len(stages), 10) self.assertIn("BIOS", stages[0]["name"]) def test_visualisations_render(self): for name in ("shader2d", "shader3d", "fractal", "ring", "state"): status, body = self.get("/api/viz/" + name) self.assertEqual(status, 200, msg=name) self.assertTrue(json.loads(body)["body"], msg=name) def test_unknown_visualisation_is_404(self): try: self.get("/api/viz/nope") except urllib.error.HTTPError as error: self.assertEqual(error.code, 404) error.close() else: self.fail("expected a 404") def test_diagnostics_cover_the_layers(self): # /api/diag lists the sections; each one is computed on demand so a # click stays responsive instead of blocking on all eleven reports status, body = self.get("/api/diag") sections = json.loads(body)["sections"] self.assertEqual(status, 200) for key in ("isa", "shader", "ring", "state", "topology"): self.assertIn(key, sections) status, body = self.get("/api/diag/isa") self.assertEqual(status, 200) self.assertTrue(json.loads(body)["body"]) def test_boot_endpoint_restarts_the_machine(self): status, body = self.post("/api/boot", {}) self.assertEqual(status, 200) self.assertTrue(json.loads(json.dumps(body))["boot"]) class TestPageScript(unittest.TestCase): """The page has to be parseable JavaScript. A single literal newline inside a quoted string silently kills the whole ")[0] @unittest.skipUnless(shutil.which("node"), "node is not installed") def test_script_parses(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "page.js") with open(path, "w", encoding="utf-8", newline="") as fh: fh.write(self.script) check = subprocess.run([shutil.which("node"), "--check", path], capture_output=True, text=True) self.assertEqual(check.returncode, 0, "the page's script does not parse:\n" + check.stderr[:1200]) def test_page_has_no_console_only_silence(self): """Every failure path in the page has to say something on screen; a silently swallowed error is what made the buttons look dead.""" self.assertIn("note('error')", self.script) self.assertIn("nothing to run", self.script) self.assertIn("[error]", self.script) def test_exec_form_posts_instead_of_reloading(self): self.assertIn('onsubmit="send(); return false;"', self.page) self.assertIn('id="cmd"', self.page) def test_terminal_fits_itself_to_the_available_width(self): # the font size is measured, not guessed in a CSS clamp self.assertIn("function fitCrt()", self.script) self.assertIn("data-font", self.page) self.assertIn("crtZoom", self.script) self.assertNotIn("clamp(9px", self.page) def test_terminal_fits_itself_to_the_available_height_too(self): """Width alone was not enough: a 30-row desktop frame ran past a 76vh box and paint() scrolled it to the bottom, hiding the menu bar. The fit has to satisfy both axes, and the box needs a definite height so the measurement is stable.""" self.assertIn("byHeight", self.script) self.assertIn("gridRows", self.script) self.assertIn("Math.min(byWidth, byHeight)", self.script) self.assertNotIn("max-height:76vh", self.page) # 30 rows at the 8px floor still fit a tiny window without the grid # scrolling; that is what the min-height is sized for. self.assertIn("body.desktop #screen{height:calc(100vh - 116px);min-height:344px}", self.page) def test_paint_scrolls_the_container_not_the_pre(self): self.assertIn("scr.scrollTop=scr.scrollHeight", self.script) self.assertNotIn("crt.scrollTop", self.script) if __name__ == "__main__": unittest.main()