diff --git a/forgetest/forgetest/manifest.py b/forgetest/forgetest/manifest.py index aecb3cf..8cccdb3 100644 --- a/forgetest/forgetest/manifest.py +++ b/forgetest/forgetest/manifest.py @@ -93,6 +93,7 @@ NON_BEHAVIORAL = [ ("*", ".vscode/**"), ("*", ".env.example"), ("forgectrl", "tools/**"), # host-side dev tools (panel dev server) + ("forgectrl", "examples/**"), # clients of the remote API, run on another computer ] diff --git a/forgetest/forgetest/suite/forgectrl.py b/forgetest/forgetest/suite/forgectrl.py index b02747a..71a1801 100644 --- a/forgetest/forgetest/suite/forgectrl.py +++ b/forgetest/forgetest/suite/forgectrl.py @@ -741,7 +741,7 @@ def scoped_tokens(ctx): ctx.check(st == 400 and "not a capability" in _words_of(body), "a capability outside the list -> %s %s", st, _words_of(body)[:120]) hub = create("forgetest hub", "machine.read,camera.lid") - pend = create("forgetest pendant", "motion.jog") + pend = create("forgetest jogger", "motion.jog") view = create("forgetest viewer", "camera.lid") passed = {} @@ -774,7 +774,7 @@ def scoped_tokens(ctx): lambda: lan(None, "POST", "/settings?token=" + hub, data={"ui_units": units})), ("a stranger's token", "authentication required", lambda: lan("fft_" + "0" * 32, "GET", "/status")), - ("the pendant on /status", "does not hold machine.read", lambda: lan(pend, "GET", "/status"))): + ("the jog token on /status", "does not hold machine.read", lambda: lan(pend, "GET", "/status"))): st, body = call() refused[name] = [st, _words_of(body)[:120]] ctx.check(st == 403 and want in _words_of(body), "%s -> %s %s (wanted 403 %r)", name, st, @@ -784,12 +784,12 @@ def scoped_tokens(ctx): # A write with a scoped token and nothing else: no session, no panel token. st, body = lan(pend, "POST", "/motion/jog", data={"x": "1", "feed": "1200"}) - ctx.check(st == 200, "the pendant token's jog -> %s %s", st, _words_of(body)[:120]) + ctx.check(st == 200, "the jog token's jog -> %s %s", st, _words_of(body)[:120]) ctx.wait_for(lambda: abs(kernel_xy_mm(ctx)[0] - x0 - 1.0) < 0.02, 10, poll=0.1) machine_idle(ctx) moved = kernel_xy_mm(ctx)[0] - x0 ev["jog_mm"] = round(moved, 3) - ctx.log("the pendant token jogged the head %.3f mm from the LAN with no session", moved) + ctx.log("the jog token jogged the head %.3f mm from the LAN with no session", moved) ctx.check(abs(moved - 1.0) < 0.02, "the jog moved %.3f mm, not 1", moved) # Plain HTTP from the LAN: refused, and the log says which token crossed in the clear. @@ -827,10 +827,10 @@ def scoped_tokens(ctx): st, listed = fc.get("/tokens") mine = {t["name"]: t for t in listed.get("tokens", []) if t.get("name", "").startswith("forgetest ")} ev["listed"] = {n: {k: t[k] for k in ("id", "caps", "last_used")} for n, t in mine.items()} - ctx.check(set(mine) == {"forgetest hub", "forgetest pendant", "forgetest viewer"}, + ctx.check(set(mine) == {"forgetest hub", "forgetest jogger", "forgetest viewer"}, "the list: %s", sorted(mine)) ctx.check(mine["forgetest hub"]["caps"] == ["machine.read", "camera.lid"] and - mine["forgetest pendant"]["caps"] == ["motion.jog"], "the capabilities listed: %s", ev["listed"]) + mine["forgetest jogger"]["caps"] == ["motion.jog"], "the capabilities listed: %s", ev["listed"]) ctx.check(all(t["last_used"] > 0 for t in mine.values()), "a used token reads as never used: %s", ev["listed"]) text = json.dumps(listed) ctx.check(hub[4:] not in text and pend[4:] not in text and view[4:] not in text and "sha" not in text and diff --git a/forgetest/forgetest/suite/setup.py b/forgetest/forgetest/suite/setup.py index a782d8f..d4055b9 100644 --- a/forgetest/forgetest/suite/setup.py +++ b/forgetest/forgetest/suite/setup.py @@ -896,12 +896,17 @@ def ssh_until_reboot(ctx): @test("setup.cloud-disabled-surface", title="Nothing points at the cloud while it is off", subsystem="setup", kind="auto", mode="grbl", est_min=1, covers=[("forgectrl", "src/main.c"), ("forgectrl", "src/super.c"), ("forgectrl", "src/settings.*"), - ("forgectrl", "src/wiz.c"), ("forgectrl", "src/hooks.h")], + ("forgectrl", "src/wiz.c"), ("forgectrl", "src/hooks.h"), ("forgectrl", "src/builtin.*")], requires=["forgectrl.settings-bounds"], description="The test turns cloud mode off itself with one write and puts every setting " - "back as found. cloud_enabled=0 takes the gfcloud homing and the cloud boot mode " - "down with it, as the cloud step does. With it at 0, POST /settings " - "controller_mode=cloud and homing_mode=gfcloud are refused (409) and leave the " + "back as found. GET /extensions lists cloud mode as a built-in with its two " + "roles (homing gfcloud, controller cloud), enabled as cloud_enabled says and each " + "role active exactly when its setting selects it. cloud_enabled=0 takes the " + "gfcloud homing and the cloud boot mode " + "down with it, as the cloud step does, and the list then reads off with no role " + "active. With it at 0, POST /settings " + "controller_mode=cloud and homing_mode=gfcloud are refused (409, in the words the " + "table gives each role) and leave the " "settings unchanged, cloud_enabled=1 without the typed phrase is refused (400) " "and leaves it at 0, and POST /mode controller=cloud is refused (409) with a " "message that names cloud mode, the machine staying in GRBL mode.") @@ -912,6 +917,29 @@ def cloud_disabled_surface(ctx): found = fc.settings() prior = found.get("cloud_enabled") or "" ev["found"] = {k: found.get(k) for k in ("cloud_enabled", "homing_mode", "controller_mode")} + + def cloud_entry(): + st, doc = fc.get("/extensions") + ctx.check(st == 200 and isinstance(doc, dict), "GET /extensions -> %s", st) + ext = [e for e in doc.get("extensions", []) if e.get("id") == "cloud"] + ctx.check(len(ext) == 1, "GET /extensions lists cloud mode %d times", len(ext)) + return ext[0] + + # The list agrees with the settings as found. + e = cloud_entry() + ev["extension_found"] = {"enabled": e.get("enabled"), "roles": e.get("roles")} + ctx.check(e.get("builtin") is True and e.get("enable_key") == "cloud_enabled" and e.get("tab") == "gfcloud" + and e.get("setup_step") == "cloud", "the cloud entry: %s", e) + ctx.check(e.get("enabled") is (prior == "1"), "the list says enabled=%s with cloud_enabled=%r", + e.get("enabled"), prior) + roles = {r["role"]: r for r in e.get("roles", [])} + ctx.check(set(roles) == {"homing", "controller"} and roles["homing"]["provider"] == "gfcloud" + and roles["homing"]["fallback"] == "none" and roles["controller"]["provider"] == "cloud" + and roles["controller"]["fallback"] == "grbl", "the cloud entry's roles: %s", e.get("roles")) + for r in roles.values(): + want = prior == "1" and found.get(r["select_key"]) == r["provider"] + ctx.check(r["active"] is want, "role %s reads active=%s with %s=%r", r["role"], r["active"], + r["select_key"], found.get(r["select_key"])) try: if prior != "0": # the one write sweeps what pointed at the cloud @@ -927,12 +955,18 @@ def cloud_disabled_surface(ctx): "cloud_enabled=0 left controller_mode at %r, expected grbl", swept.get("controller_mode")) ctx.check(fc.settings().get("cloud_enabled") == "0", "cloud_enabled does not read 0") + off = cloud_entry() + ev["extension_off"] = {"enabled": off.get("enabled"), "active": [r["active"] for r in off["roles"]]} + ctx.check(off.get("enabled") is False and not any(r["active"] for r in off["roles"]), + "the list with cloud mode off: %s", ev["extension_off"]) before = fc.settings() - for key, val in (("controller_mode", "cloud"), ("homing_mode", "gfcloud")): + for key, val, words in (("controller_mode", "cloud", "cloud mode is not enabled on this machine"), + ("homing_mode", "gfcloud", "cloud homing needs cloud mode enabled")): st, body = fc.post("/settings", data={key: val}) ev["settings %s=%s" % (key, val)] = st ctx.log("POST /settings %s=%s -> %s %s", key, val, st, body if isinstance(body, str) else "") ctx.check(st == 409, "%s=%s with cloud off -> %s, expected 409", key, val, st) + ctx.check(isinstance(body, str) and words in body, "%s=%s was refused in other words: %r", key, val, body) # the pair in one request is refused too: the request's own switch counts st, body = fc.post("/settings", data={"cloud_enabled": "0", "controller_mode": "cloud"}) ctx.check(st == 409, "cloud_enabled=0 with controller_mode=cloud -> %s, expected 409", st) @@ -972,6 +1006,9 @@ def cloud_disabled_surface(ctx): for key in ("cloud_enabled", "homing_mode", "controller_mode"): ctx.check((after.get(key) or "") == (found.get(key) or ""), "%s not restored: %r, was %r", key, after.get(key), found.get(key)) + back = cloud_entry() + ctx.check(back.get("enabled") == e.get("enabled") and back.get("roles") == e.get("roles"), + "the list did not come back as found: %s, was %s", back, e) # ---------------------------------------------------------- the operator diff --git a/forgetest/tests/test_setup_suite.py b/forgetest/tests/test_setup_suite.py index 2412b85..8e5acb1 100644 --- a/forgetest/tests/test_setup_suite.py +++ b/forgetest/tests/test_setup_suite.py @@ -16,7 +16,7 @@ import unittest import helpers from forgetest import baseline, catalog -from forgetest.runner import Context, Run +from forgetest.runner import Context, Failed, Run from forgetest.suite import setup IDS = ("setup.gate-blocks-controllers", "setup.override-until-reboot", @@ -375,6 +375,26 @@ class CloudDisabledSurfaceTests(unittest.TestCase): return 409, "cloud mode is not enabled on this machine" return None self.fake.on_post = on_post + self.list_lies = None # a test sets this to break the list one way + + def on_get(path, query): + if path != "/extensions": + return None + s = self.fake.state["settings"] + on = s.get("cloud_enabled") == "1" + roles = [{"role": role, "provider": prov, "kind": kind, "select_key": key, "fallback": back, + "active": on and s.get(key) == prov} + for role, prov, kind, key, back in (("homing", "gfcloud", "runner-fd", "homing_mode", "none"), + ("controller", "cloud", "supervised", "controller_mode", + "grbl"))] + if self.list_lies == "stale-active": + roles[0]["active"] = True + if self.list_lies == "always-on": + on = True + return 200, {"extensions": [{"id": "cloud", "name": "Glowforge cloud mode", "builtin": True, + "enabled": on, "enable_key": "cloud_enabled", "setup_step": "cloud", + "tab": "gfcloud", "roles": roles}]} + self.fake.on_get = on_get def tearDown(self): self.fake.stop() @@ -385,6 +405,13 @@ class CloudDisabledSurfaceTests(unittest.TestCase): t.fn(Context(run, None, t)) return run + def test_a_list_that_disagrees_with_the_settings_fails(self): + for lie in ("always-on", "stale-active"): + self.list_lies = lie + self.fake.state["settings"].update({"cloud_enabled": "1", "homing_mode": "gfcloud"}) + with self.assertRaises(Failed, msg=lie): + self.run_test() + def test_refusals_and_restore(self): run = self.run_test() s = self.fake.state["settings"]