mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-10-15 17:19:55 -07:00
ScottW514
7246707df5
setup.gate-blocks-controllers asserted on the machine which steps the image's table requires (GET /wiz's missing list) and recommends (each not run yet, never the gate's parts or the optional steps). Both are the setup table's decisions over a record, software with no hardware in them, and forgectrl's setup_test now proves both on the image's own table: the required steps in order (10b), the missing list emptying when they are done (4b), and the recommended steps in order, each not run yet (10b). The catalog keeps what needs the machine: the supervisor holding and spawning the controller, the process, and the Grbl port. - The two missing-list checks and the recommended-list check leave the test's body; its description says where they are proven. - RECOMMENDED_STEPS and NEVER_RECOMMENDED go with them (no test read them any more), and the meta-test case that only compared them. Proof: the forgetest host suite passes under Linux (test_setup_suite with the case removed among it); the coverage lint is unchanged. Acceptance: every setup.py test (the constants were the module's shared text), setup.gate-blocks-controllers among them.
507 lines
26 KiB
Python
507 lines
26 KiB
Python
# Copyright 2026 514 LLC d/b/a OpenGlow
|
|
# Written by Scott Wiederhold
|
|
# https://community.openglow.org
|
|
# SPDX-License-Identifier: MIT
|
|
|
|
"""The setup.* suite on the host: the registration (ids, kinds, the
|
|
takeover tests, the operator tests' hands), the record builders, the
|
|
machine's name, the cookie parsing, the LED cue, the settle rule for a
|
|
gated supervisor, and the cloud-off surface test driven end to end
|
|
against the fake daemon."""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
|
|
import helpers
|
|
from forgetest import baseline, catalog
|
|
from forgetest.runner import Context, Failed, Run
|
|
from forgetest.suite import setup
|
|
|
|
IDS = ("setup.gate-blocks-controllers",
|
|
"setup.advisories-rehash", "setup.account-login", "setup.https-only-writes",
|
|
"setup.ssh-until-reboot", "setup.cloud-disabled-surface",
|
|
"setup.factory-return", "setup.machine-name", "setup.first-run-flow",
|
|
"setup.first-run-page", "setup.what-changed", "setup.record-export",
|
|
"setup.mirror")
|
|
OPERATOR = ("setup.first-run-flow", "setup.first-run-page")
|
|
|
|
|
|
class RegistrationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.reg = catalog.load_suite()
|
|
|
|
def test_every_test_is_registered_with_its_kind(self):
|
|
kinds = {tid: self.reg[tid].kind for tid in IDS if tid in self.reg}
|
|
self.assertEqual(sorted(kinds), sorted(IDS))
|
|
for tid in OPERATOR:
|
|
self.assertEqual(kinds[tid], "operator", tid)
|
|
for tid in IDS:
|
|
if tid not in OPERATOR:
|
|
self.assertEqual(kinds[tid], "auto", tid)
|
|
self.assertNotIn("setup.root-ssh-refused", self.reg) # folded into ssh-until-reboot
|
|
self.assertNotIn("setup.wizard-first-run", self.reg) # split into the flow and the page
|
|
self.assertNotIn("setup.override-until-reboot", self.reg) # the gate has nothing to lift
|
|
self.assertNotIn("setup.sheet", self.reg) # the two burns, each on its own scrap
|
|
|
|
def test_the_record_swapping_tests_are_takeovers(self):
|
|
for tid in ("setup.gate-blocks-controllers", "setup.advisories-rehash",
|
|
"setup.first-run-flow", "setup.first-run-page",
|
|
"setup.what-changed", "setup.mirror"):
|
|
self.assertEqual(self.reg[tid].hardware, "takeover", tid)
|
|
self.assertEqual(self.reg["setup.record-export"].hardware, "api")
|
|
|
|
def test_the_lifecycle_tests_cover_their_files(self):
|
|
changed = set(self.reg["setup.what-changed"].covers)
|
|
export = set(self.reg["setup.record-export"].covers)
|
|
mirror = set(self.reg["setup.mirror"].covers)
|
|
self.assertIn(("forgectrl", "src/ui/panel.js"), changed)
|
|
self.assertIn(("forgectrl", "src/recordhtml.*"), export)
|
|
self.assertIn(("forgectrl", "src/logs.*"), export)
|
|
self.assertIn(("forgectrl", "src/wizdark.*"), mirror)
|
|
self.assertIn("setup.check-sensors", self.reg["setup.mirror"].requires)
|
|
|
|
def test_flag_of_reads_the_status_document(self):
|
|
w = {"recommended": [{"id": "laser.floor", "reason": "the tube was replaced"},
|
|
{"id": "laser.focus", "reason": "the tray was replaced"},
|
|
{"id": "airflow", "reason": "not run yet"}, "stray"]}
|
|
self.assertEqual(setup.flag_of(w, "laser.floor"), "the tube was replaced")
|
|
self.assertEqual(setup.flag_of(w, "laser.focus"), "the tray was replaced")
|
|
self.assertEqual(setup.flag_of(w, "airflow"), "not run yet")
|
|
self.assertIsNone(setup.flag_of(w, "stray")) # only an entry with an id counts
|
|
self.assertIsNone(setup.flag_of(w, "cameras"))
|
|
self.assertIsNone(setup.flag_of({}, "x"))
|
|
self.assertEqual(setup.recommended(w), {"laser.floor": "the tube was replaced",
|
|
"laser.focus": "the tray was replaced",
|
|
"airflow": "not run yet"})
|
|
|
|
def test_the_form_wizards_and_the_required_steps(self):
|
|
self.assertEqual(setup.FORM_WIZARDS, ("advisories", "account", "preferences", "machine", "cloud"))
|
|
# in the order the gate's reason names them
|
|
self.assertEqual(setup.REQUIRED_FORM_STEPS, ("preferences", "machine", "cloud"))
|
|
self.assertEqual(setup.REQUIRED_STEPS, ("preferences", "machine", "cloud", "cooling.flow"))
|
|
self.assertIn("cooling.flow", setup.REQUIRED_STEPS)
|
|
|
|
def test_the_gate_test_starts_from_an_open_gate_and_the_intake_tests_from_a_count(self):
|
|
reg = self.reg
|
|
self.assertIs(reg["setup.gate-blocks-controllers"].precheck, setup.gate_is_open)
|
|
from forgetest.suite import setup_dark
|
|
for tid in ("setup.check-intakes", "setup.check-airflow"):
|
|
self.assertIs(reg[tid].precheck, setup_dark.intakes_counted, tid)
|
|
|
|
def test_the_first_run_flow_runs_unattended_with_the_fixture_and_the_page_walk_does_not(self):
|
|
flow = self.reg["setup.first-run-flow"]
|
|
self.assertEqual(list(flow.actions), ["button"])
|
|
self.assertFalse(flow.hands)
|
|
self.assertTrue(flow.fixture_runnable(("button", "lid", "interlock")))
|
|
page = self.reg["setup.first-run-page"]
|
|
self.assertTrue(page.hands)
|
|
self.assertFalse(page.fixture_runnable(("button", "lid", "interlock")))
|
|
self.assertIn("setup.first-run-flow", page.requires)
|
|
|
|
def test_the_first_run_split_keeps_the_backend_and_the_page_apart(self):
|
|
flow = set(self.reg["setup.first-run-flow"].covers)
|
|
page = set(self.reg["setup.first-run-page"].covers)
|
|
for want in (("forgectrl", "src/wiz.*"), ("forgectrl", "src/setup.*"), ("forgectrl", "src/users.*"),
|
|
("forgectrl", "src/button.*"), ("forgectrl", "src/led.*"), ("forgectrl", "src/advisories.*")):
|
|
self.assertIn(want, flow, want)
|
|
self.assertNotIn(want, page, want)
|
|
for want in (("forgectrl", "src/ui/wizard.*"), ("forgectrl", "src/ui/md.js"), ("forgectrl", "src/tls.*")):
|
|
self.assertIn(want, page, want)
|
|
|
|
def test_the_only_attended_tests_need_a_workstation_or_a_scrap(self):
|
|
# With the bench actuator up, four setup tests keep a person: the
|
|
# page walk, the print from the Glowforge app, and the two burns,
|
|
# each on a fresh scrap. Everything else runs from the queue.
|
|
attended = sorted(tid for tid, t in self.reg.items() if tid.startswith("setup.")
|
|
and t.kind != "auto" and not t.fixture_runnable(("button", "lid", "interlock")))
|
|
self.assertEqual(attended, ["setup.cloud-header-capture", "setup.first-run-page",
|
|
"setup.floor", "setup.focus"])
|
|
|
|
def test_the_factory_return_never_runs_the_return(self):
|
|
import inspect
|
|
t = self.reg["setup.factory-return"]
|
|
src = inspect.getsource(t.fn)
|
|
self.assertNotIn('"confirm": "1"', src) # the only argument that starts the return
|
|
self.assertIn('"confirm": "0"', src)
|
|
self.assertEqual(t.kind, "auto") # the return itself is a bench drill, not a test
|
|
self.assertFalse(t.hands)
|
|
|
|
def test_the_machine_name_covers_nothing_by_design(self):
|
|
self.assertEqual(self.reg["setup.machine-name"].covers, ())
|
|
|
|
def test_the_login_test_makes_its_own_account(self):
|
|
# No bench credentials, no precheck: the test installs a temporary
|
|
# account under a takeover and restores the real one.
|
|
t = self.reg["setup.account-login"]
|
|
self.assertIsNone(getattr(t, "precheck", None))
|
|
self.assertEqual(t.hardware, "takeover")
|
|
with open(setup.__file__) as f:
|
|
src = f.read()
|
|
for k in ("FORGETEST_LOGIN_NAME", "FORGETEST_LOGIN_PASSWORD"):
|
|
self.assertNotIn(k, src)
|
|
|
|
|
|
class RecordTests(unittest.TestCase):
|
|
STATUS = {"sheet_id": "ABCDE-FGHIJ",
|
|
"documents": [{"id": "safety-and-risk", "hash": "a" * 64, "consent": "typed"},
|
|
{"id": "licenses", "hash": "b" * 64, "consent": "check"}],
|
|
"wizards": [{"id": "advisories", "version": 1}, {"id": "account", "version": 1},
|
|
{"id": "machine", "version": 2}]}
|
|
|
|
def test_complete_record_counts_as_set_up(self):
|
|
rec = setup.complete_record(self.STATUS)
|
|
self.assertEqual(rec["schema"], 1)
|
|
self.assertEqual(rec["advisories"]["safety-and-risk"]["hash"], "a" * 64)
|
|
self.assertEqual(rec["advisories"]["safety-and-risk"]["method"], "typed")
|
|
self.assertIn("pressed_at", rec["acceptance"])
|
|
self.assertEqual(rec["account"]["name"], "bench")
|
|
self.assertEqual(rec["wizards"]["machine"]["version"], 2)
|
|
self.assertEqual(rec["sheet_id"], "ABCDE-FGHIJ")
|
|
self.assertTrue(rec["completed"])
|
|
json.dumps(rec)
|
|
|
|
def test_complete_record_keeps_the_base_account_and_machine(self):
|
|
base = {"account": {"name": "owner", "uid": 1000, "created": "x"}, "machine": {"model": "pro"},
|
|
"sheet_id": "KKKKK-LLLLL", "created": "then"}
|
|
rec = setup.complete_record(self.STATUS, base)
|
|
self.assertEqual(rec["account"]["name"], "owner")
|
|
self.assertEqual(rec["machine"], {"model": "pro"})
|
|
self.assertEqual(rec["sheet_id"], "KKKKK-LLLLL")
|
|
self.assertEqual(rec["created"], "then")
|
|
|
|
def test_without_wizards_keeps_consent_and_account(self):
|
|
rec = setup.complete_record(self.STATUS)
|
|
rec["flags"] = {"machine": {"reason": "x"}}
|
|
out = setup.record_without_wizards(rec)
|
|
self.assertEqual(out["wizards"], {})
|
|
self.assertEqual(out["flags"], {})
|
|
self.assertEqual(out["advisories"], rec["advisories"])
|
|
self.assertEqual(out["account"], rec["account"])
|
|
self.assertEqual(rec["flags"], {"machine": {"reason": "x"}}) # a copy
|
|
|
|
def test_with_required_steps_keeps_them_and_nothing_else(self):
|
|
rec = setup.complete_record(self.STATUS)
|
|
rec["wizards"]["machine"]["result"] = {"model": "basic"}
|
|
rec["wizards"]["switches"] = {"version": 1}
|
|
rec["flags"] = {"machine": {"reason": "the head was replaced"}}
|
|
out = setup.record_with_required_steps(rec)
|
|
self.assertEqual(sorted(out["wizards"]), ["cloud", "cooling.flow", "machine", "preferences"])
|
|
self.assertEqual(out["wizards"]["machine"]["result"], {"model": "basic"}) # kept from the record
|
|
self.assertEqual(out["wizards"]["preferences"]["version"], 1) # written at version 1
|
|
self.assertEqual(out["wizards"]["cooling.flow"]["version"], 1)
|
|
self.assertEqual(out["flags"], {})
|
|
self.assertEqual(out["account"], rec["account"])
|
|
self.assertIn("switches", rec["wizards"]) # a copy
|
|
|
|
def test_without_consent_keeps_the_account_and_the_wizards(self):
|
|
rec = setup.complete_record(self.STATUS)
|
|
out = setup.record_without_consent(rec)
|
|
self.assertEqual(out["advisories"], {})
|
|
self.assertNotIn("acceptance", out)
|
|
self.assertEqual(out["account"], rec["account"])
|
|
self.assertEqual(out["wizards"], rec["wizards"])
|
|
self.assertIn("pressed_at", rec["acceptance"]) # a copy
|
|
|
|
def test_without_account_keeps_the_consent(self):
|
|
rec = setup.complete_record(self.STATUS)
|
|
out = setup.record_without_account(rec)
|
|
self.assertNotIn("account", out)
|
|
self.assertEqual(out["advisories"], rec["advisories"])
|
|
self.assertEqual(out["acceptance"], rec["acceptance"])
|
|
self.assertEqual(rec["account"]["name"], "bench") # a copy
|
|
|
|
def test_write_file_and_remove(self):
|
|
tmp = tempfile.mkdtemp(prefix="forgetest-comm-")
|
|
try:
|
|
p = os.path.join(tmp, "sub", "setup.json")
|
|
setup.write_file(p, b"{}\n")
|
|
self.assertEqual(setup.read_file(p), b"{}\n")
|
|
if os.name == "posix": # a mode means nothing on a Windows host
|
|
self.assertEqual(oct(os.stat(p).st_mode & 0o777), oct(0o600))
|
|
setup.write_file(p, None)
|
|
self.assertIsNone(setup.read_file(p))
|
|
setup.write_file(p, None) # a second remove is silent
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
def test_paths_follow_the_daemon_environment(self):
|
|
os.environ["FORGECTRL_DATA_DIR"] = "/tmp/ffdata"
|
|
os.environ["GF_RUN_DIR"] = "/tmp/ffrun"
|
|
try:
|
|
self.assertEqual(setup.record_path(), "/tmp/ffdata/setup.json")
|
|
self.assertEqual(setup.users_path(), "/tmp/ffdata/users")
|
|
self.assertEqual(setup.ssh_flag_path(), "/tmp/ffrun/ssh-enabled")
|
|
finally:
|
|
os.environ.pop("FORGECTRL_DATA_DIR", None)
|
|
os.environ.pop("GF_RUN_DIR", None)
|
|
|
|
|
|
class MachineNameTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.root = tempfile.mkdtemp()
|
|
self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)
|
|
|
|
def _net(self, **devs):
|
|
"""A /sys/class/net tree; setup.mac_suffix reads it through
|
|
setup.read_file, which takes an absolute path."""
|
|
for dev, mac in devs.items():
|
|
d = os.path.join(self.root, dev)
|
|
os.makedirs(d)
|
|
with open(os.path.join(d, "address"), "w") as f:
|
|
f.write(mac + "\n")
|
|
real = setup.read_file
|
|
|
|
def read(path):
|
|
head = "/sys/class/net/"
|
|
if path.startswith(head):
|
|
return real(os.path.join(self.root, path[len(head):]))
|
|
return real(path)
|
|
|
|
setup.read_file = read
|
|
self.addCleanup(setattr, setup, "read_file", real)
|
|
|
|
def test_the_wifi_address_names_the_machine(self):
|
|
self._net(wlan0="2C:6B:7D:0D:B0:0A", eth0="00:11:22:33:44:55")
|
|
self.assertEqual(setup.mac_suffix(), "b00a")
|
|
|
|
def test_a_machine_with_no_wifi_falls_back_to_the_wired_address(self):
|
|
self._net(eth0="00:11:22:33:44:55")
|
|
self.assertEqual(setup.mac_suffix(), "4455")
|
|
|
|
def test_an_unread_address_is_no_address(self):
|
|
self._net(wlan0="00:00:00:00:00:00")
|
|
self.assertEqual(setup.mac_suffix(), "")
|
|
|
|
def test_no_interface_is_no_address(self):
|
|
self._net()
|
|
self.assertEqual(setup.mac_suffix(), "")
|
|
|
|
|
|
class SmallHelpersTests(unittest.TestCase):
|
|
def test_cookie_parsing(self):
|
|
sid = "ab" * 32
|
|
value = "ffsid=%s; Path=/; HttpOnly; Secure; SameSite=Strict; Max-Age=43200" % sid
|
|
self.assertEqual(setup.session_from_cookie(value), sid)
|
|
self.assertEqual(setup.cookie_flags(value), {"path", "httponly", "secure", "samesite", "max-age"})
|
|
self.assertIsNone(setup.session_from_cookie("ffsid=; Path=/; Max-Age=0"))
|
|
self.assertIsNone(setup.session_from_cookie(""))
|
|
|
|
def test_the_teal_breathe_cue(self):
|
|
self.assertTrue(setup.breathes_teal({"target": [0, 180, 200], "pulse_on": [0, 1400, 1400],
|
|
"pulse_off": [0, 1400, 1400]}))
|
|
self.assertFalse(setup.breathes_teal({"target": [0, 255, 40], "pulse_on": [0, 0, 0],
|
|
"pulse_off": [0, 0, 0]})) # solid green
|
|
self.assertFalse(setup.breathes_teal({"target": [200, 200, 200], "pulse_on": [1800, 1800, 1800],
|
|
"pulse_off": [1800] * 3})) # white
|
|
self.assertFalse(setup.breathes_teal({"target": [None, None, None], "pulse_on": [None] * 3}))
|
|
self.assertFalse(setup.breathes_teal({}))
|
|
|
|
def test_decode(self):
|
|
self.assertEqual(setup.decode(b'{"a": 1}'), {"a": 1})
|
|
self.assertEqual(setup.decode(b"cloud mode is not enabled"), "cloud mode is not enabled")
|
|
|
|
|
|
class GatedSettleTests(unittest.TestCase):
|
|
"""A supervisor that reports the setup gate closed is settled:
|
|
it spawns nothing until the gate opens, so a takeover that installs
|
|
a gating record must not wait the whole settle timeout for it."""
|
|
|
|
def setUp(self):
|
|
self.fake = helpers.FakeForgectrl().start()
|
|
baseline.Baseline._unreachable_until = 0.0
|
|
|
|
def tearDown(self):
|
|
self.fake.stop()
|
|
|
|
def test_gated_returns_at_once(self):
|
|
self.fake.state["mode"] = {"mode": "grbl", "controller": "gated", "pid": 0, "motion": "unverified",
|
|
"gated": True, "why": "the intake fans are not counted"}
|
|
lines = []
|
|
body = baseline.Baseline(lines.append).wait_settled(timeout=8)
|
|
self.assertEqual(body["controller"], "gated")
|
|
self.assertTrue(any("controller=gated" in ln for ln in lines))
|
|
|
|
|
|
class FirstRunSeedTests(unittest.TestCase):
|
|
"""The record the first-run tests install: a first run behind the
|
|
checks and the burns."""
|
|
|
|
STATUS = {"wizards": [{"id": "advisories", "version": 1, "class": "form"},
|
|
{"id": "account", "version": 1, "class": "form"},
|
|
{"id": "cloud", "version": 1, "class": "form"},
|
|
{"id": "switches", "version": 1, "class": "dark"},
|
|
{"id": "motion", "version": 2, "class": "dark"},
|
|
{"id": "laser.focus", "version": 1, "class": "live"}]}
|
|
|
|
def test_no_consent_no_account_not_complete_and_every_check_done(self):
|
|
base = {"created": "2026-09-01T00:00:00Z", "completed": "2026-09-02T00:00:00Z",
|
|
"advisories": {"safety-and-risk": {"hash": "x"}}, "acceptance": {"pressed_at": "t"},
|
|
"account": {"name": "scott", "uid": 1000}, "flags": {"flow_thin": True},
|
|
"wizards": {"advisories": {"version": 1}, "switches": {"version": 1, "result": {"lid": True}}}}
|
|
rec = setup.first_run_record(self.STATUS, base)
|
|
self.assertNotIn("completed", rec)
|
|
self.assertNotIn("acceptance", rec)
|
|
self.assertNotIn("account", rec)
|
|
self.assertEqual(rec["advisories"], {})
|
|
self.assertEqual(rec["created"], "2026-09-01T00:00:00Z")
|
|
self.assertEqual(rec["flags"], {"flow_thin": True})
|
|
self.assertEqual(sorted(rec["wizards"]), ["laser.focus", "motion", "switches"])
|
|
self.assertEqual(rec["wizards"]["switches"]["result"], {"lid": True}) # kept from the base
|
|
self.assertEqual(rec["wizards"]["motion"]["version"], 2) # written at the catalog version
|
|
|
|
def test_without_a_base_record(self):
|
|
rec = setup.first_run_record(self.STATUS, None)
|
|
self.assertEqual(sorted(rec["wizards"]), ["laser.focus", "motion", "switches"])
|
|
self.assertEqual(rec["wizards"]["laser.focus"]["version"], 1)
|
|
|
|
|
|
class SshdPolicyTests(unittest.TestCase):
|
|
def test_the_three_keys_out_of_sshd_t(self):
|
|
from forgetest import hw
|
|
calls = []
|
|
|
|
def fake_run(cmd, timeout=60):
|
|
calls.append(cmd)
|
|
return 0, "port 22\npermitrootlogin no\npasswordauthentication yes\npermitemptypasswords no\nx11forwarding no\n"
|
|
real = hw.run
|
|
hw.run = fake_run
|
|
try:
|
|
self.assertEqual(setup.sshd_policy(), {"permitrootlogin": "no", "passwordauthentication": "yes",
|
|
"permitemptypasswords": "no"})
|
|
finally:
|
|
hw.run = real
|
|
self.assertEqual(calls, [["/usr/sbin/sshd", "-T"]])
|
|
|
|
def test_a_failing_sshd_reports_the_error(self):
|
|
from forgetest import hw
|
|
real = hw.run
|
|
hw.run = lambda cmd, timeout=60: (1, "sshd: no hostkeys available")
|
|
try:
|
|
self.assertEqual(setup.sshd_policy(), {"error": "sshd: no hostkeys available"})
|
|
finally:
|
|
hw.run = real
|
|
|
|
|
|
class CloudDisabledSurfaceTests(unittest.TestCase):
|
|
"""The cloud-off surface test against the fake daemon: cloud_enabled
|
|
is turned off with one write that sweeps the cloud choices, the three
|
|
cloud-pointing writes and the phrase-less cloud_enabled=1 are refused,
|
|
the settings are left alone, and the prior values come back."""
|
|
|
|
def setUp(self):
|
|
self.fake = helpers.FakeForgectrl().start()
|
|
self.fake.state["settings"].update({"controller_mode": "grbl", "homing_mode": "none",
|
|
"cloud_enabled": "1", "ui_units": "metric"})
|
|
|
|
def on_post(path, form):
|
|
s = self.fake.state["settings"]
|
|
if path == "/settings":
|
|
# as the daemon rules it: on from off takes the typed phrase
|
|
if form.get("cloud_enabled") == "1" and s.get("cloud_enabled") != "1" \
|
|
and form.get("phrase") != "I UNDERSTAND":
|
|
return 400, "type I UNDERSTAND to turn cloud mode on"
|
|
# the request's own cloud choices, as main.c checks them
|
|
enabled = form.get("cloud_enabled", s.get("cloud_enabled")) == "1"
|
|
if not enabled and form.get("controller_mode") == "cloud":
|
|
return 409, "cloud mode is not enabled on this machine"
|
|
if not enabled and form.get("homing_mode") == "gfcloud":
|
|
return 409, "cloud homing needs cloud mode enabled"
|
|
for k, v in form.items():
|
|
if k == "phrase":
|
|
continue
|
|
if v == "":
|
|
s.pop(k, None)
|
|
else:
|
|
s[k] = v
|
|
# off sweeps what pointed at the cloud, as the cloud step does
|
|
if form.get("cloud_enabled") == "0":
|
|
if "homing_mode" not in form and s.get("homing_mode") == "gfcloud":
|
|
s["homing_mode"] = "none"
|
|
if "controller_mode" not in form and s.get("controller_mode") == "cloud":
|
|
s["controller_mode"] = "grbl"
|
|
return 200, dict(s)
|
|
if path == "/mode" and form.get("controller") == "cloud" and s.get("cloud_enabled") != "1":
|
|
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()
|
|
|
|
def run_test(self):
|
|
t = catalog.load_suite()["setup.cloud-disabled-surface"]
|
|
run = Run("test", t.id, t.title)
|
|
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"]
|
|
self.assertEqual(s["cloud_enabled"], "1") # restored, with the phrase
|
|
self.assertEqual(s["controller_mode"], "grbl")
|
|
self.assertEqual(s["homing_mode"], "none")
|
|
self.assertNotIn("phrase", s)
|
|
posts = [(p, f) for p, f in self.fake.posts]
|
|
self.assertIn(("/settings", {"cloud_enabled": "0"}), posts)
|
|
self.assertIn(("/settings", {"cloud_enabled": "1"}), posts) # refused: no phrase
|
|
self.assertIn(("/settings", {"cloud_enabled": "1", "phrase": "I UNDERSTAND"}), posts)
|
|
self.assertIn(("/mode", {"controller": "cloud"}), posts)
|
|
self.assertEqual(run.evidence["mode_cloud"]["status"], 409)
|
|
self.assertEqual(run.evidence["settings cloud_enabled=1 no phrase"], 400)
|
|
self.assertEqual(run.evidence["found"]["cloud_enabled"], "1")
|
|
|
|
def test_cloud_homing_and_boot_mode_are_swept_by_the_one_write_and_go_back(self):
|
|
# As the daemon rules it: cloud_enabled=0 takes the gfcloud homing
|
|
# and the cloud boot mode down with it, so the test writes the one
|
|
# key, checks the sweep, and puts the three back, cloud mode first.
|
|
self.fake.state["settings"].update({"controller_mode": "cloud", "homing_mode": "gfcloud"})
|
|
run = self.run_test()
|
|
s = self.fake.state["settings"]
|
|
self.assertEqual((s["cloud_enabled"], s["homing_mode"], s["controller_mode"]), ("1", "gfcloud", "cloud"))
|
|
posts = [(p, f) for p, f in self.fake.posts if p == "/settings"]
|
|
order = [f for _, f in posts]
|
|
self.assertEqual(order[:1], [{"cloud_enabled": "0"}])
|
|
self.assertEqual(order[-3:], [{"cloud_enabled": "1", "phrase": "I UNDERSTAND"},
|
|
{"homing_mode": "gfcloud"}, {"controller_mode": "cloud"}])
|
|
self.assertEqual(run.evidence["found"], {"cloud_enabled": "1", "homing_mode": "gfcloud",
|
|
"controller_mode": "cloud"})
|
|
|
|
def test_an_unset_value_is_cleared_back(self):
|
|
self.fake.state["settings"].pop("cloud_enabled")
|
|
self.run_test()
|
|
self.assertNotIn("cloud_enabled", self.fake.state["settings"])
|
|
self.assertIn(("/settings", {"cloud_enabled": ""}), self.fake.posts)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|