| 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): |
| |
| |
| 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 |
| <script>: the page still renders, every handler is missing, and the EXEC |
| button just submits the form and reloads. That is exactly what happened, |
| so the check is a real scanner rather than a substring guess. |
| """ |
|
|
| @classmethod |
| def setUpClass(cls): |
| from hdc.webui import PAGE |
|
|
| cls.page = PAGE |
| cls.script = PAGE.split("<script>")[1].split("</script>")[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): |
| |
| 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) |
| |
| |
| 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() |
|
|
