mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-27 08:41:13 -07:00
ScottW514
19ce4d78d2
forgectrl gains a table of built-in extensions, with cloud mode as entry one, GET /extensions to serve it, and POST /settings asking it which selections point at the cloud, how to refuse them, and what they fall back to. It also gains an example client under examples/, and its test tokens lose the names of clients nobody is building. setup.cloud-disabled-surface is the gate for "nothing points at the cloud while it is off", so it now holds the list to the settings three times: as found (enabled as cloud_enabled says, the two roles with their providers and fallbacks, each active exactly when its setting selects it), with the cloud off (not enabled, no role active), and as restored. The two refusals are held to the table's words. Its covers name src/builtin.*. The helper lives inside the test's own function, so no other test of the module changes its fingerprint. The unit test's fake daemon serves the route the way builtin.c does, and gains a case with two lists that lie (enabled after the cloud went off; a role that stays active), each of which must fail the test. forgectrl.tokens: the jog token is "forgetest jogger". manifest: forgectrl's examples/** joins the non-behavioral paths. They are clients that run on another computer: outside the image, outside every fingerprint, and outside the coverage lint. Proven. The unit suite passes (421) with no undefined name; the two lying lists fail the test as they should. On the bench reference, forgectrl's registry daemon hot-deployed over image 20260920152153, on a machine with cloud mode on: setup.cloud-disabled-surface PASS through the whole path (off, the sweep, the refusals, the restore), and forgectrl.tokens PASS with the renamed token. Acceptance. setup.cloud-disabled-surface is the gate for forgectrl's built-in table through the settings route.
456 lines
23 KiB
Python
456 lines
23 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.override-until-reboot",
|
|
"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
|
|
|
|
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 = {"required": ["airflow", {"id": "laser.floor", "reason": "the tube was replaced"}],
|
|
"recommended": [{"id": "laser.focus", "reason": "the tray was replaced"}]}
|
|
self.assertEqual(setup.flag_of(w, "required", "laser.floor"), "the tube was replaced")
|
|
self.assertEqual(setup.flag_of(w, "recommended", "laser.focus"), "the tray was replaced")
|
|
self.assertIsNone(setup.flag_of(w, "required", "airflow")) # a plain id is the table, not a flag
|
|
self.assertIsNone(setup.flag_of(w, "required", "laser.focus"))
|
|
self.assertIsNone(setup.flag_of({}, "recommended", "x"))
|
|
|
|
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_the_sheet(self):
|
|
# With the bench actuator up, three setup tests keep a
|
|
# person: the page walk, the print from the Glowforge app, and
|
|
# the sheet. 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.sheet"])
|
|
|
|
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": {"level": "required", "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": {"level": "required", "reason": "x"}}) # 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.override_path(), "/tmp/ffrun/setup-override")
|
|
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": "setup required: machine"}
|
|
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 sheet."""
|
|
|
|
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": "sheet.frame", "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"]), ["motion", "sheet.frame", "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"]), ["motion", "sheet.frame", "switches"])
|
|
self.assertEqual(rec["wizards"]["sheet.frame"]["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()
|