mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-28 01:01:12 -07:00
forgetest: the setup suite, and the sensors check asks nothing
The commission.* tests are setup.* (subsystem "setup"): suite/setup.py, setup_dark.py, setup_sheet.py, and the host tests test_setup_*.py. The coverage maps name src/setup.* in place of src/commission.*, the record is setup.json, and the bench seed in forgetest.init creates /run/forgefirm/setup-override. setup.check-sensors follows the check as it is now: no question is asked and no setting is written. The test reads the settings before and after and fails on any change, and fails at once if the check opens a prompt. The host test drives the fake daemon with no prompt and proves both outcomes. Proof: test_setup_dark.py, test_setup_sheet.py, and test_setup_suite.py pass (46 tests). The coverage lint names src/commission.c and .h as uncovered until the forgectrl pin moves to the revision that carries the rename.
This commit is contained in:
ScottW514
@@ -1,8 +1,8 @@
|
||||
"""The setup's checks in the catalog: registered as automatic tests with
|
||||
no precheck, the takeover ones marked so, every covers map naming the
|
||||
check runner; and the check driver itself against the fake daemon: a
|
||||
start, a prompt answered, the result taken, the record read at version
|
||||
1, and the settings a check may write put back as found."""
|
||||
start, the run polled to its end with no question asked, the result
|
||||
taken, the record read at version 1, and the settings read as before."""
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
@@ -12,8 +12,8 @@ import helpers # noqa: E402
|
||||
from forgetest import catalog # noqa: E402
|
||||
from forgetest.runner import Context, Run # noqa: E402
|
||||
|
||||
IDS = ["commission.check-switches", "commission.check-sensors", "commission.check-airflow",
|
||||
"commission.check-cameras", "commission.check-motion", "commission.check-flow-verify"]
|
||||
IDS = ["setup.check-switches", "setup.check-sensors", "setup.check-airflow",
|
||||
"setup.check-cameras", "setup.check-motion", "setup.check-flow-verify"]
|
||||
|
||||
|
||||
class RegistrationTests(unittest.TestCase):
|
||||
@@ -25,11 +25,11 @@ class RegistrationTests(unittest.TestCase):
|
||||
# bench fixture performs; the rest need no hands at all.
|
||||
for tid in IDS:
|
||||
self.assertIn(tid, self.reg)
|
||||
for tid in ("commission.check-switches", "commission.check-cameras"):
|
||||
for tid in ("setup.check-switches", "setup.check-cameras"):
|
||||
self.assertEqual(self.reg[tid].kind, "operator", tid)
|
||||
self.assertTrue(self.reg[tid].fixture_runnable(("button", "lid", "interlock")), tid)
|
||||
for tid in ("commission.check-sensors", "commission.check-airflow", "commission.check-motion",
|
||||
"commission.check-flow-verify"):
|
||||
for tid in ("setup.check-sensors", "setup.check-airflow", "setup.check-motion",
|
||||
"setup.check-flow-verify"):
|
||||
self.assertEqual(self.reg[tid].kind, "auto", tid)
|
||||
|
||||
def test_no_check_asks_the_operator_to_prepare_anything(self):
|
||||
@@ -37,7 +37,7 @@ class RegistrationTests(unittest.TestCase):
|
||||
self.assertIsNone(getattr(self.reg[tid], "precheck", None), tid)
|
||||
|
||||
def test_the_checks_that_take_the_machine_say_so(self):
|
||||
for tid in ("commission.check-airflow", "commission.check-motion", "commission.check-flow-verify"):
|
||||
for tid in ("setup.check-airflow", "setup.check-motion", "setup.check-flow-verify"):
|
||||
self.assertEqual(self.reg[tid].hardware, "takeover", tid)
|
||||
|
||||
def test_every_check_covers_the_runner_and_the_page(self):
|
||||
@@ -62,9 +62,17 @@ class SensorsCheckTests(unittest.TestCase):
|
||||
"shots": {"lid": False, "head": False}}
|
||||
self.versions = {}
|
||||
self.answers = []
|
||||
self.polls = 0
|
||||
|
||||
def on_get(path, q):
|
||||
if path == "/wiz/dark":
|
||||
# One poll sees the sampling; the next sees the result.
|
||||
if self.dark["running"]:
|
||||
self.polls += 1
|
||||
if self.polls >= 2:
|
||||
self.dark.update({"running": False, "prompt": None, "progress": 100,
|
||||
"result": dict(self.RESULT)})
|
||||
self.versions["sensors"] = 1
|
||||
return 200, dict(self.dark)
|
||||
if path == "/wiz":
|
||||
return 200, {"versions": dict(self.versions), "required": [], "completed": True}
|
||||
@@ -74,19 +82,11 @@ class SensorsCheckTests(unittest.TestCase):
|
||||
s = self.fake.state["settings"]
|
||||
if path == "/wiz/sensors/start":
|
||||
self.dark.update({"id": "sensors", "running": True, "phase": "sampling",
|
||||
"prompt": {"seq": 7, "id": "room-temp", "kind": "number",
|
||||
"text": "Optional: the room temperature.",
|
||||
"options": ["Set", "Skip"]}})
|
||||
"prompt": None})
|
||||
return 200, {"started": True, "id": "sensors"}
|
||||
if path == "/wiz/sensors/answer":
|
||||
self.answers.append(dict(form))
|
||||
if form.get("seq") != "7":
|
||||
return 409, {"error": "no such prompt is open"}
|
||||
if form.get("value") == "Skip":
|
||||
self.dark.update({"running": False, "prompt": None, "progress": 100,
|
||||
"result": dict(self.RESULT)})
|
||||
self.versions["sensors"] = 1
|
||||
return 200, {"ok": True}
|
||||
return 409, {"error": "no such prompt is open"}
|
||||
if path == "/settings":
|
||||
for k, v in form.items():
|
||||
if v == "":
|
||||
@@ -101,15 +101,35 @@ class SensorsCheckTests(unittest.TestCase):
|
||||
def tearDown(self):
|
||||
self.fake.stop()
|
||||
|
||||
def test_start_prompt_answer_result_and_restore(self):
|
||||
t = catalog.load_suite()["commission.check-sensors"]
|
||||
def test_start_no_question_result_and_settings_untouched(self):
|
||||
t = catalog.load_suite()["setup.check-sensors"]
|
||||
run = Run("test", t.id, t.title)
|
||||
t.fn(Context(run, None, t))
|
||||
self.assertIn(("/wiz/sensors/start", {}), [(p, f) for p, f in self.fake.posts])
|
||||
self.assertEqual(self.answers, [{"seq": "7", "value": "Skip"}])
|
||||
self.assertEqual(self.answers, [])
|
||||
self.assertEqual(run.evidence["result"]["laser_pgood"], 1)
|
||||
self.assertEqual(run.evidence["settings_before"], {"cool_temp_offset_c": "0.5"})
|
||||
self.assertEqual(run.evidence["settings_before"].get("cool_temp_offset_c"), "0.5")
|
||||
self.assertEqual(self.fake.state["settings"]["cool_temp_offset_c"], "0.5")
|
||||
self.assertNotIn("/settings", [p for p, _ in self.fake.posts])
|
||||
|
||||
def test_a_question_fails_the_check(self):
|
||||
t = catalog.load_suite()["setup.check-sensors"]
|
||||
run = Run("test", t.id, t.title)
|
||||
real_get = self.fake.on_get
|
||||
|
||||
def on_get(path, q):
|
||||
if path == "/wiz/dark" and self.dark["running"]:
|
||||
d = dict(self.dark)
|
||||
d["prompt"] = {"seq": 3, "id": "room-temp", "kind": "number", "text": "?",
|
||||
"options": ["Set", "Skip"]}
|
||||
return 200, d
|
||||
return real_get(path, q)
|
||||
self.fake.on_get = on_get
|
||||
with self.assertRaises(Exception) as cm:
|
||||
t.fn(Context(run, None, t))
|
||||
self.assertIn("asked a question", str(cm.exception))
|
||||
self.assertEqual(self.answers, [])
|
||||
self.assertIn(("/wiz/sensors/abort", {}), [(p, f) for p, f in self.fake.posts])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
@@ -14,7 +14,7 @@ sys.path.insert(0, os.path.dirname(__file__))
|
||||
import helpers # noqa: E402
|
||||
from forgetest import catalog # noqa: E402
|
||||
from forgetest.runner import Context, Run # noqa: E402
|
||||
from forgetest.suite import commission_sheet # noqa: E402
|
||||
from forgetest.suite import setup_sheet # noqa: E402
|
||||
|
||||
LIVE = ["sheet.frame", "laser.focus", "laser.floor", "laser.dose-curve", "laser.corner", "cooling.flow-load"]
|
||||
# The focus card's program is read again after it ran: the served ladder
|
||||
@@ -107,7 +107,7 @@ FOUND = {"lens_hall_edge_z_mm": "3.35", "lens_stop_below_steps": "10", "lens_sto
|
||||
|
||||
class RegistrationTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.t = catalog.load_suite()["commission.sheet"]
|
||||
self.t = catalog.load_suite()["setup.sheet"]
|
||||
|
||||
def test_one_live_test_on_the_button_with_the_sheet_at_hand(self):
|
||||
t = self.t
|
||||
@@ -118,12 +118,12 @@ class RegistrationTests(unittest.TestCase):
|
||||
self.assertTrue(t.hands)
|
||||
self.assertFalse(t.fixture_runnable(("button", "lid", "interlock")))
|
||||
self.assertIsNone(getattr(t, "precheck", None))
|
||||
for want in ("commission.check-motion", "laser.emission-witness", "cooling.flow-verify"):
|
||||
for want in ("setup.check-motion", "laser.emission-witness", "cooling.flow-verify"):
|
||||
self.assertIn(want, t.requires)
|
||||
|
||||
def test_no_other_sheet_test_remains(self):
|
||||
ids = [i for i in catalog.load_suite() if i.startswith("commission.sheet")]
|
||||
self.assertEqual(ids, ["commission.sheet"])
|
||||
ids = [i for i in catalog.load_suite() if i.startswith("setup.sheet")]
|
||||
self.assertEqual(ids, ["setup.sheet"])
|
||||
|
||||
def test_covers_the_live_runner_the_renderer_the_sender_the_page_and_the_driver(self):
|
||||
covers = set(self.t.covers)
|
||||
@@ -137,7 +137,7 @@ class RegistrationTests(unittest.TestCase):
|
||||
text = " ".join(self.t.steps).lower()
|
||||
self.assertIn("press the machine's button once", text)
|
||||
self.assertNotIn("when the page", text)
|
||||
self.assertEqual(len(commission_sheet.CARDS), 6)
|
||||
self.assertEqual(len(setup_sheet.CARDS), 6)
|
||||
|
||||
|
||||
class ScriptedDaemon:
|
||||
@@ -246,7 +246,7 @@ class SheetRunTests(unittest.TestCase):
|
||||
self.fake.stop()
|
||||
|
||||
def test_the_whole_sheet_on_one_piece(self):
|
||||
t = catalog.load_suite()["commission.sheet"]
|
||||
t = catalog.load_suite()["setup.sheet"]
|
||||
run = Run("test", t.id, t.title)
|
||||
run.unattended = True # the ready gate passes; the arm cue is a notice
|
||||
t.fn(Context(run, None, t))
|
||||
@@ -270,12 +270,12 @@ class SheetRunTests(unittest.TestCase):
|
||||
# frame, the frame's program ran in the fallback window, the focus
|
||||
# card's window was in the settings when the floor card started,
|
||||
# and the window check filed what it compared.
|
||||
for k in commission_sheet.LENS_SETTINGS:
|
||||
for k in setup_sheet.LENS_SETTINGS:
|
||||
self.assertNotIn(k, d.settings_at_start["sheet.frame"], k)
|
||||
self.assertEqual(run.evidence["lens_fresh"]["stops_found"], False)
|
||||
self.assertEqual(run.evidence["program_z"]["sheet.frame"]["reach"], [-0.07, 7.46])
|
||||
floor_start = d.settings_at_start["laser.floor"]
|
||||
self.assertEqual({k: floor_start[k] for k in commission_sheet.LENS_SETTINGS},
|
||||
self.assertEqual({k: floor_start[k] for k in setup_sheet.LENS_SETTINGS},
|
||||
{"lens_hall_edge_z_mm": "6.96", "lens_stop_below_steps": "14", "lens_stop_above_steps": "20"})
|
||||
fw = run.evidence["focus_window"]
|
||||
self.assertEqual(fw["window"], FOCUS_WINDOW)
|
||||
@@ -294,7 +294,7 @@ class SheetRunTests(unittest.TestCase):
|
||||
place_prompts, place_result = SCRIPT["sheet.place"]
|
||||
SCRIPT["sheet.place"] = (place_prompts, dict(place_result, thickness_mm=3.175))
|
||||
try:
|
||||
t = catalog.load_suite()["commission.sheet"]
|
||||
t = catalog.load_suite()["setup.sheet"]
|
||||
run = Run("test", t.id, t.title)
|
||||
run.unattended = True
|
||||
t.fn(Context(run, None, t))
|
||||
@@ -310,7 +310,7 @@ class SheetRunTests(unittest.TestCase):
|
||||
place_prompts, place_result = SCRIPT["sheet.place"]
|
||||
SCRIPT["sheet.place"] = (place_prompts, dict(place_result, thickness_mm=0.0))
|
||||
try:
|
||||
t = catalog.load_suite()["commission.sheet"]
|
||||
t = catalog.load_suite()["setup.sheet"]
|
||||
run = Run("test", t.id, t.title)
|
||||
run.unattended = True
|
||||
with self.assertRaises(Failed) as cm:
|
||||
@@ -326,7 +326,7 @@ class SheetRunTests(unittest.TestCase):
|
||||
SCRIPT_DARK["emission"] = {"hv_max": 0, "laser_on_samples": 0, "thermopile_delta": 0}
|
||||
SCRIPT["laser.floor"] = (SCRIPT["laser.floor"][0], SCRIPT_DARK)
|
||||
try:
|
||||
t = catalog.load_suite()["commission.sheet"]
|
||||
t = catalog.load_suite()["setup.sheet"]
|
||||
run = Run("test", t.id, t.title)
|
||||
run.unattended = True
|
||||
with self.assertRaises(Failed):
|
||||
@@ -347,7 +347,7 @@ class SheetRunTests(unittest.TestCase):
|
||||
saved = dict(START_WRITES["laser.focus"])
|
||||
START_WRITES["laser.focus"] = {}
|
||||
try:
|
||||
t = catalog.load_suite()["commission.sheet"]
|
||||
t = catalog.load_suite()["setup.sheet"]
|
||||
run = Run("test", t.id, t.title)
|
||||
run.unattended = True
|
||||
with self.assertRaises(Failed) as cm:
|
||||
@@ -364,7 +364,7 @@ class SheetRunTests(unittest.TestCase):
|
||||
# refused before the card starts: nothing burns.
|
||||
from forgetest.runner import Failed
|
||||
self.daemon.stray_z = 10.6
|
||||
t = catalog.load_suite()["commission.sheet"]
|
||||
t = catalog.load_suite()["setup.sheet"]
|
||||
run = Run("test", t.id, t.title)
|
||||
run.unattended = True
|
||||
with self.assertRaises(Failed) as cm:
|
||||
@@ -1,4 +1,4 @@
|
||||
"""The commission.* suite on the host: the registration (ids, kinds, the
|
||||
"""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
|
||||
@@ -12,15 +12,15 @@ import unittest
|
||||
import helpers
|
||||
from forgetest import baseline, catalog
|
||||
from forgetest.runner import Context, Run
|
||||
from forgetest.suite import commission
|
||||
from forgetest.suite import setup
|
||||
|
||||
IDS = ("commission.gate-blocks-controllers", "commission.override-until-reboot",
|
||||
"commission.advisories-rehash", "commission.account-login", "commission.https-only-writes",
|
||||
"commission.ssh-until-reboot", "commission.cloud-disabled-surface",
|
||||
"commission.factory-return", "commission.machine-name", "commission.first-run-flow",
|
||||
"commission.first-run-page", "commission.what-changed", "commission.record-export",
|
||||
"commission.mirror")
|
||||
OPERATOR = ("commission.first-run-flow", "commission.first-run-page")
|
||||
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):
|
||||
@@ -35,49 +35,49 @@ class RegistrationTests(unittest.TestCase):
|
||||
for tid in IDS:
|
||||
if tid not in OPERATOR:
|
||||
self.assertEqual(kinds[tid], "auto", tid)
|
||||
self.assertNotIn("commission.root-ssh-refused", self.reg) # folded into ssh-until-reboot
|
||||
self.assertNotIn("commission.wizard-first-run", self.reg) # split into the flow and the page
|
||||
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 ("commission.gate-blocks-controllers", "commission.advisories-rehash",
|
||||
"commission.first-run-flow", "commission.first-run-page",
|
||||
"commission.what-changed", "commission.mirror"):
|
||||
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["commission.record-export"].hardware, "api")
|
||||
self.assertEqual(self.reg["setup.record-export"].hardware, "api")
|
||||
|
||||
def test_the_lifecycle_tests_cover_their_files(self):
|
||||
changed = set(self.reg["commission.what-changed"].covers)
|
||||
export = set(self.reg["commission.record-export"].covers)
|
||||
mirror = set(self.reg["commission.mirror"].covers)
|
||||
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("commission.check-sensors", self.reg["commission.mirror"].requires)
|
||||
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(commission.flag_of(w, "required", "laser.floor"), "the tube was replaced")
|
||||
self.assertEqual(commission.flag_of(w, "recommended", "laser.focus"), "the tray was replaced")
|
||||
self.assertIsNone(commission.flag_of(w, "required", "airflow")) # a plain id is the table, not a flag
|
||||
self.assertIsNone(commission.flag_of(w, "required", "laser.focus"))
|
||||
self.assertIsNone(commission.flag_of({}, "recommended", "x"))
|
||||
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["commission.first-run-flow"]
|
||||
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["commission.first-run-page"]
|
||||
page = self.reg["setup.first-run-page"]
|
||||
self.assertTrue(page.hands)
|
||||
self.assertFalse(page.fixture_runnable(("button", "lid", "interlock")))
|
||||
self.assertIn("commission.first-run-flow", page.requires)
|
||||
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["commission.first-run-flow"].covers)
|
||||
page = set(self.reg["commission.first-run-page"].covers)
|
||||
for want in (("forgectrl", "src/wiz.*"), ("forgectrl", "src/commission.*"), ("forgectrl", "src/users.*"),
|
||||
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)
|
||||
@@ -85,17 +85,17 @@ class RegistrationTests(unittest.TestCase):
|
||||
self.assertIn(want, page, want)
|
||||
|
||||
def test_the_only_attended_tests_need_a_workstation_or_the_sheet(self):
|
||||
# With the bench actuator up, three commissioning tests keep a
|
||||
# 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("commission.")
|
||||
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, ["commission.cloud-header-capture", "commission.first-run-page",
|
||||
"commission.sheet"])
|
||||
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["commission.factory-return"]
|
||||
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)
|
||||
@@ -103,15 +103,15 @@ class RegistrationTests(unittest.TestCase):
|
||||
self.assertFalse(t.hands)
|
||||
|
||||
def test_the_machine_name_covers_nothing_by_design(self):
|
||||
self.assertEqual(self.reg["commission.machine-name"].covers, ())
|
||||
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["commission.account-login"]
|
||||
t = self.reg["setup.account-login"]
|
||||
self.assertIsNone(getattr(t, "precheck", None))
|
||||
self.assertEqual(t.hardware, "takeover")
|
||||
with open(commission.__file__) as f:
|
||||
with open(setup.__file__) as f:
|
||||
src = f.read()
|
||||
for k in ("FORGETEST_LOGIN_NAME", "FORGETEST_LOGIN_PASSWORD"):
|
||||
self.assertNotIn(k, src)
|
||||
@@ -124,8 +124,8 @@ class RecordTests(unittest.TestCase):
|
||||
"wizards": [{"id": "advisories", "version": 1}, {"id": "account", "version": 1},
|
||||
{"id": "machine", "version": 2}]}
|
||||
|
||||
def test_complete_record_counts_as_commissioned(self):
|
||||
rec = commission.complete_record(self.STATUS)
|
||||
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")
|
||||
@@ -139,16 +139,16 @@ class RecordTests(unittest.TestCase):
|
||||
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 = commission.complete_record(self.STATUS, base)
|
||||
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 = commission.complete_record(self.STATUS)
|
||||
rec = setup.complete_record(self.STATUS)
|
||||
rec["flags"] = {"machine": {"level": "required", "reason": "x"}}
|
||||
out = commission.record_without_wizards(rec)
|
||||
out = setup.record_without_wizards(rec)
|
||||
self.assertEqual(out["wizards"], {})
|
||||
self.assertEqual(out["flags"], {})
|
||||
self.assertEqual(out["advisories"], rec["advisories"])
|
||||
@@ -158,14 +158,14 @@ class RecordTests(unittest.TestCase):
|
||||
def test_write_file_and_remove(self):
|
||||
tmp = tempfile.mkdtemp(prefix="forgetest-comm-")
|
||||
try:
|
||||
p = os.path.join(tmp, "sub", "commissioning.json")
|
||||
commission.write_file(p, b"{}\n")
|
||||
self.assertEqual(commission.read_file(p), b"{}\n")
|
||||
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))
|
||||
commission.write_file(p, None)
|
||||
self.assertIsNone(commission.read_file(p))
|
||||
commission.write_file(p, None) # a second remove is silent
|
||||
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)
|
||||
|
||||
@@ -173,10 +173,10 @@ class RecordTests(unittest.TestCase):
|
||||
os.environ["FORGECTRL_DATA_DIR"] = "/tmp/ffdata"
|
||||
os.environ["GF_RUN_DIR"] = "/tmp/ffrun"
|
||||
try:
|
||||
self.assertEqual(commission.record_path(), "/tmp/ffdata/commissioning.json")
|
||||
self.assertEqual(commission.users_path(), "/tmp/ffdata/users")
|
||||
self.assertEqual(commission.override_path(), "/tmp/ffrun/commissioning-override")
|
||||
self.assertEqual(commission.ssh_flag_path(), "/tmp/ffrun/ssh-enabled")
|
||||
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)
|
||||
@@ -188,14 +188,14 @@ class MachineNameTests(unittest.TestCase):
|
||||
self.addCleanup(shutil.rmtree, self.root, ignore_errors=True)
|
||||
|
||||
def _net(self, **devs):
|
||||
"""A /sys/class/net tree; commission.mac_suffix reads it through
|
||||
commission.read_file, which takes an absolute path."""
|
||||
"""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 = commission.read_file
|
||||
real = setup.read_file
|
||||
|
||||
def read(path):
|
||||
head = "/sys/class/net/"
|
||||
@@ -203,52 +203,52 @@ class MachineNameTests(unittest.TestCase):
|
||||
return real(os.path.join(self.root, path[len(head):]))
|
||||
return real(path)
|
||||
|
||||
commission.read_file = read
|
||||
self.addCleanup(setattr, commission, "read_file", real)
|
||||
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(commission.mac_suffix(), "b00a")
|
||||
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(commission.mac_suffix(), "4455")
|
||||
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(commission.mac_suffix(), "")
|
||||
self.assertEqual(setup.mac_suffix(), "")
|
||||
|
||||
def test_no_interface_is_no_address(self):
|
||||
self._net()
|
||||
self.assertEqual(commission.mac_suffix(), "")
|
||||
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(commission.session_from_cookie(value), sid)
|
||||
self.assertEqual(commission.cookie_flags(value), {"path", "httponly", "secure", "samesite", "max-age"})
|
||||
self.assertIsNone(commission.session_from_cookie("ffsid=; Path=/; Max-Age=0"))
|
||||
self.assertIsNone(commission.session_from_cookie(""))
|
||||
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(commission.breathes_teal({"target": [0, 180, 200], "pulse_on": [0, 1400, 1400],
|
||||
self.assertTrue(setup.breathes_teal({"target": [0, 180, 200], "pulse_on": [0, 1400, 1400],
|
||||
"pulse_off": [0, 1400, 1400]}))
|
||||
self.assertFalse(commission.breathes_teal({"target": [0, 255, 40], "pulse_on": [0, 0, 0],
|
||||
self.assertFalse(setup.breathes_teal({"target": [0, 255, 40], "pulse_on": [0, 0, 0],
|
||||
"pulse_off": [0, 0, 0]})) # solid green
|
||||
self.assertFalse(commission.breathes_teal({"target": [200, 200, 200], "pulse_on": [1800, 1800, 1800],
|
||||
self.assertFalse(setup.breathes_teal({"target": [200, 200, 200], "pulse_on": [1800, 1800, 1800],
|
||||
"pulse_off": [1800] * 3})) # white
|
||||
self.assertFalse(commission.breathes_teal({"target": [None, None, None], "pulse_on": [None] * 3}))
|
||||
self.assertFalse(commission.breathes_teal({}))
|
||||
self.assertFalse(setup.breathes_teal({"target": [None, None, None], "pulse_on": [None] * 3}))
|
||||
self.assertFalse(setup.breathes_teal({}))
|
||||
|
||||
def test_decode(self):
|
||||
self.assertEqual(commission.decode(b'{"a": 1}'), {"a": 1})
|
||||
self.assertEqual(commission.decode(b"cloud mode is not enabled"), "cloud mode is not enabled")
|
||||
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 commissioning gate closed is settled:
|
||||
"""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."""
|
||||
|
||||
@@ -261,7 +261,7 @@ class GatedSettleTests(unittest.TestCase):
|
||||
|
||||
def test_gated_returns_at_once(self):
|
||||
self.fake.state["mode"] = {"mode": "grbl", "controller": "gated", "pid": 0, "motion": "unverified",
|
||||
"gated": True, "why": "commissioning required: machine"}
|
||||
"gated": True, "why": "setup required: machine"}
|
||||
lines = []
|
||||
body = baseline.Baseline(lines.append).wait_settled(timeout=8)
|
||||
self.assertEqual(body["controller"], "gated")
|
||||
@@ -284,7 +284,7 @@ class FirstRunSeedTests(unittest.TestCase):
|
||||
"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 = commission.first_run_record(self.STATUS, base)
|
||||
rec = setup.first_run_record(self.STATUS, base)
|
||||
self.assertNotIn("completed", rec)
|
||||
self.assertNotIn("acceptance", rec)
|
||||
self.assertNotIn("account", rec)
|
||||
@@ -296,7 +296,7 @@ class FirstRunSeedTests(unittest.TestCase):
|
||||
self.assertEqual(rec["wizards"]["motion"]["version"], 2) # written at the catalog version
|
||||
|
||||
def test_without_a_base_record(self):
|
||||
rec = commission.first_run_record(self.STATUS, None)
|
||||
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)
|
||||
|
||||
@@ -312,7 +312,7 @@ class SshdPolicyTests(unittest.TestCase):
|
||||
real = hw.run
|
||||
hw.run = fake_run
|
||||
try:
|
||||
self.assertEqual(commission.sshd_policy(), {"permitrootlogin": "no", "passwordauthentication": "yes",
|
||||
self.assertEqual(setup.sshd_policy(), {"permitrootlogin": "no", "passwordauthentication": "yes",
|
||||
"permitemptypasswords": "no"})
|
||||
finally:
|
||||
hw.run = real
|
||||
@@ -323,7 +323,7 @@ class SshdPolicyTests(unittest.TestCase):
|
||||
real = hw.run
|
||||
hw.run = lambda cmd, timeout=60: (1, "sshd: no hostkeys available")
|
||||
try:
|
||||
self.assertEqual(commission.sshd_policy(), {"error": "sshd: no hostkeys available"})
|
||||
self.assertEqual(setup.sshd_policy(), {"error": "sshd: no hostkeys available"})
|
||||
finally:
|
||||
hw.run = real
|
||||
|
||||
@@ -375,7 +375,7 @@ class CloudDisabledSurfaceTests(unittest.TestCase):
|
||||
self.fake.stop()
|
||||
|
||||
def run_test(self):
|
||||
t = catalog.load_suite()["commission.cloud-disabled-surface"]
|
||||
t = catalog.load_suite()["setup.cloud-disabled-surface"]
|
||||
run = Run("test", t.id, t.title)
|
||||
t.fn(Context(run, None, t))
|
||||
return run
|
||||
Reference in New Issue
Block a user