forgefirm/forgetest/tests/test_setup_suite.py

# 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, and the settle rule
for a gated supervisor."""
import json
import os
import shutil
import tempfile
import unittest

import helpers
from forgetest import baseline, catalog
from forgetest.suite import setup

IDS = ("setup.account-login", "setup.https-only-writes",
       "setup.ssh-until-reboot", "setup.factory-return", "setup.machine-name", "setup.first-run-flow",
       "setup.first-run-page", "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.first-run-flow", "setup.first-run-page", "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):
        export = set(self.reg["setup.record-export"].covers)
        mirror = set(self.reg["setup.mirror"].covers)
        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_recommended_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.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_intake_tests_start_from_a_count(self):
        reg = self.reg
        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


if __name__ == "__main__":
    unittest.main()