File size: 11,797 Bytes
ad474c9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 | 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
<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):
# 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()
|