mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-10-15 17:19:55 -07:00
ScottW514
ed52e752ee
The catalog follows forgectrl's setup (openglow-org/forgefirm#4): the gate is the advisories with the press, the account, the required steps (preferences, machine, cloud), and the intake fan count; every other step is recommended; the setup sheet is retired for two self-contained burns and a dark lens travel check; the override file is gone. - setup.gate-blocks-controllers installs four records: without consent, without the account, and without the three required steps the controller is gated with the reason naming the cause; with them done the gate opens, the controller runs, and the other steps are recommended as not run. Clearing cool_intake_fans closes the gate live with its reason, and putting it back as found reopens it. - setup.override-until-reboot is deleted with its helpers, and forgetest.init no longer writes the override file at boot. - setup.first-run-flow follows the reason from the three required steps to open; setup.first-run-page walks the choice screen after them; setup.what-changed reads the one-level map with the gate unmoved. - New setup.check-intakes (the count by the tachs equals the machine's, the setting put back as found) and setup.check-lens (the window written equals the stops found or the fallback, and the settings and /status hold it; both keys restored). A precheck refuses the intake and airflow checks while cool_intake_fans is unset, so neither can leave it unset in the middle of a campaign. - setup_burns.py replaces setup_sheet.py: setup.focus and setup.floor answer the gantry stops, the home, the scrap and the thickness, and keep the preview, emission and window checks; motion.tray starts the focus burn for its tray-out refusal. Proof: the forgetest host suite (529 before the required-steps edits, and the full suite after them) and the manifest load pass; pyflakes reports no undefined names. The coverage lint is clean against forgectrl's working tree; against the current pin, setup.check-lens's src/lensfind.* glob selects no file until the forgectrl pin moves to the commit that adds it. Acceptance: these are the tests themselves; the campaign runs them with cool_intake_fans set on the bench.
510 lines
26 KiB
Python
510 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_gate_parts_and_the_optional_steps_are_never_recommended(self):
|
|
self.assertFalse(set(setup.RECOMMENDED_STEPS) & set(setup.NEVER_RECOMMENDED))
|
|
self.assertFalse(set(setup.RECOMMENDED_STEPS) & set(setup.REQUIRED_STEPS))
|
|
for wid in ("advisories", "account", "airflow.intakes", "laser.floor", "motion.envelope"):
|
|
self.assertIn(wid, setup.NEVER_RECOMMENDED)
|
|
for wid in ("lens.travel", "laser.focus", "airflow", "switches"):
|
|
self.assertIn(wid, setup.RECOMMENDED_STEPS)
|
|
self.assertEqual(setup.FORM_WIZARDS, ("advisories", "account", "preferences", "machine", "cloud"))
|
|
# in the order the gate's reason names them
|
|
self.assertEqual(setup.REQUIRED_STEPS, ("preferences", "machine", "cloud"))
|
|
|
|
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", "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["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()
|