mirror of
https://github.com/ruvnet/RuView.git
synced 2026-10-19 05:10:13 -05:00
Merge PR #2152 (aepod) into release integration
# Conflicts: # firmware/esp32-csi-node/provision.py
This commit is contained in:
ruv committed
commit
e0c4207bd5
2 files changed
+400
-3
No files matched your search
@@ -28,6 +28,11 @@ ADDITIVE-BY-DEFAULT (issue #391, #574 phase 1):
|
||||
3. Generate + flash NVS from the merged state.
|
||||
4. Write the merged state back to the state file.
|
||||
|
||||
State is keyed by the board's MAC, read with esptool, so a board plugged
|
||||
into a port another board used doesn't inherit its node_id (#1755).
|
||||
Port-keyed files from earlier versions move to the board's record the
|
||||
first time that board is provisioned.
|
||||
|
||||
Net effect: partial reconfigure works the way users expect. Pass `--reset`
|
||||
to wipe both the state file AND the device NVS for first-time provisioning
|
||||
of a recycled board.
|
||||
@@ -46,6 +51,7 @@ import io
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import re
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
@@ -291,6 +297,114 @@ def save_state(port: str, state_dir: str, state: dict) -> str:
|
||||
return path
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Board identity (#1755)
|
||||
# ---------------------------------------------------------------------------
|
||||
#
|
||||
# Serial port names get reused when boards are swapped, so a port-keyed state
|
||||
# file hands one board's node_id to the next. State is keyed by the chip's
|
||||
# base MAC instead. Each record also stores the MAC and the port it was last
|
||||
# written from, so --state can find it without opening the port.
|
||||
|
||||
STATE_MAC_KEY = "_chip_mac"
|
||||
STATE_PORT_KEY = "_port"
|
||||
|
||||
_ESPTOOL_MAC_RE = re.compile(
|
||||
r"^\s*(BASE MAC|MAC):\s*((?:[0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2})\s*$", re.MULTILINE
|
||||
)
|
||||
|
||||
|
||||
def normalize_mac(value: str):
|
||||
"""Return `value` as lowercase aa:bb:cc:dd:ee:ff, or None if it isn't a MAC."""
|
||||
parts = re.split(r"[:-]", value.strip())
|
||||
if len(parts) != 6 or not all(re.fullmatch(r"[0-9a-fA-F]{2}", p) for p in parts):
|
||||
return None
|
||||
return ":".join(p.lower() for p in parts)
|
||||
|
||||
|
||||
def parse_esptool_mac(output: str):
|
||||
"""Return the base MAC from `esptool read_mac` output, or None.
|
||||
|
||||
Most chips print one 6-byte "MAC:" line. EUI-64 chips (C5, C6, H2) print an
|
||||
8-byte "MAC:" line, then "BASE MAC:" with the 6-byte address.
|
||||
"""
|
||||
found = {label: mac.lower() for label, mac in _ESPTOOL_MAC_RE.findall(output)}
|
||||
return found.get("BASE MAC") or found.get("MAC")
|
||||
|
||||
|
||||
def read_chip_mac(port: str, baud: int, chip: str):
|
||||
"""Read the connected board's base MAC with esptool. None if it can't."""
|
||||
cmd = [
|
||||
sys.executable, "-m", "esptool",
|
||||
"--chip", chip,
|
||||
"--port", port,
|
||||
"--baud", str(baud),
|
||||
"read_mac",
|
||||
]
|
||||
try:
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return None
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
return parse_esptool_mac(result.stdout)
|
||||
|
||||
|
||||
def mac_state_key(mac: str) -> str:
|
||||
"""State-file key for a board, e.g. mac-aabbccddeeff."""
|
||||
return "mac-" + mac.replace(":", "")
|
||||
|
||||
|
||||
def boards_last_on_port(port: str, state_dir: str) -> list:
|
||||
"""MACs of boards whose state was last written from `port`, newest first."""
|
||||
if not os.path.isdir(state_dir):
|
||||
return []
|
||||
found = []
|
||||
for name in os.listdir(state_dir):
|
||||
if not (name.startswith("mac-") and name.endswith(".json")):
|
||||
continue
|
||||
data = load_state(name[:-len(".json")], state_dir)
|
||||
mac = data.get(STATE_MAC_KEY)
|
||||
if data.get(STATE_PORT_KEY) == port and mac:
|
||||
found.append((os.path.getmtime(os.path.join(state_dir, name)), mac))
|
||||
return [mac for _, mac in sorted(found, reverse=True)]
|
||||
|
||||
|
||||
def load_board_state(port: str, mac, state_dir: str):
|
||||
"""Return (prior state, legacy path) for the board `mac` on `port`.
|
||||
|
||||
Without a MAC this is the old port-keyed lookup. With one, the board's own
|
||||
record wins. If it has none but a port-keyed file from an earlier version
|
||||
exists, that file is returned for migration along with its path.
|
||||
"""
|
||||
if mac is None:
|
||||
return load_state(port, state_dir), None
|
||||
prior = load_state(mac_state_key(mac), state_dir)
|
||||
if prior:
|
||||
return prior, None
|
||||
legacy = _state_path_for(port, state_dir)
|
||||
if os.path.isfile(legacy):
|
||||
return load_state(port, state_dir), legacy
|
||||
return {}, None
|
||||
|
||||
|
||||
def save_board_state(port: str, mac, state_dir: str, state: dict, legacy=None) -> str:
|
||||
"""Persist `state` under the board's MAC (or the port if the MAC is unknown).
|
||||
|
||||
A migrated port-keyed file is removed afterwards, so no other board on that
|
||||
port can pick it up.
|
||||
"""
|
||||
if mac is None:
|
||||
return save_state(port, state_dir, state)
|
||||
state = dict(state)
|
||||
state[STATE_MAC_KEY] = mac
|
||||
state[STATE_PORT_KEY] = port
|
||||
path = save_state(mac_state_key(mac), state_dir, state)
|
||||
if legacy and os.path.isfile(legacy):
|
||||
os.unlink(legacy)
|
||||
return path
|
||||
|
||||
|
||||
def merge_state_into_args(args, prior: dict) -> dict:
|
||||
"""Overlay `args` onto `prior` for every MERGEABLE_ATTRS attribute.
|
||||
|
||||
@@ -514,6 +628,10 @@ def main():
|
||||
"previously-staged keys should NOT be re-applied.")
|
||||
parser.add_argument("--state-dir", default=_default_state_dir(),
|
||||
help="Override the per-user state directory (default: per-OS user config dir).")
|
||||
parser.add_argument("--mac", type=str,
|
||||
help="Board MAC (AA:BB:CC:DD:EE:FF) whose state to use. Normally read "
|
||||
"from the chip with esptool; --state and --dry-run never open the "
|
||||
"port, so pass it there to pick a board.")
|
||||
parser.add_argument("--state", action="store_true",
|
||||
help="Print the merged state that WOULD be flashed for this port and exit. "
|
||||
"Useful for debugging which keys are about to land on the device. "
|
||||
@@ -537,6 +655,33 @@ def main():
|
||||
print("WARNING: --password is visible in ps and shell history; "
|
||||
"use --password-file instead.", file=sys.stderr)
|
||||
cli_ssid, cli_password = args.ssid, args.password
|
||||
# --- Board identity (#1755) ---
|
||||
# State is keyed by the chip's MAC so a board swapped onto a reused port
|
||||
# doesn't inherit the previous board's node_id.
|
||||
if args.mac is not None:
|
||||
chip_mac = normalize_mac(args.mac)
|
||||
if chip_mac is None:
|
||||
parser.error(f"--mac must be in AA:BB:CC:DD:EE:FF format, got '{args.mac}'")
|
||||
elif args.state:
|
||||
# Inspection only: don't open the port; show the board last
|
||||
# provisioned from it.
|
||||
recent = boards_last_on_port(args.port, args.state_dir)
|
||||
chip_mac = recent[0] if recent else None
|
||||
if chip_mac:
|
||||
print(f"Showing board {chip_mac}, last provisioned on {args.port}. "
|
||||
f"Pass --mac to pick another.", file=sys.stderr)
|
||||
elif args.dry_run:
|
||||
# No board involved; keep the port-keyed state as before.
|
||||
chip_mac = None
|
||||
else:
|
||||
chip_mac = read_chip_mac(args.port, args.baud, args.chip)
|
||||
if chip_mac:
|
||||
print(f"Board MAC: {chip_mac}")
|
||||
else:
|
||||
print(f"WARNING: could not read the board's MAC with esptool, so state "
|
||||
f"stays keyed by port {args.port}. Pass --mac to key it by board.",
|
||||
file=sys.stderr)
|
||||
legacy = None
|
||||
|
||||
# --- Per-port state load + merge (additive-by-default, #391 / #574) ---
|
||||
if args.reset:
|
||||
@@ -548,7 +693,12 @@ def main():
|
||||
print(f"--reset: ignoring state file {path}", file=sys.stderr)
|
||||
prior = {}
|
||||
else:
|
||||
prior = load_state(args.port, args.state_dir)
|
||||
prior, legacy = load_board_state(args.port, chip_mac, args.state_dir)
|
||||
if legacy:
|
||||
print(f"Using port-keyed state {legacy} for board {chip_mac}; it moves to "
|
||||
f"the board's own record when state is next saved. If this isn't the "
|
||||
f"board last provisioned on {args.port}, rerun with --reset.",
|
||||
file=sys.stderr)
|
||||
merged = merge_state_into_args(args, prior)
|
||||
|
||||
# A new --ssid with no password given: ask for it on a terminal instead of
|
||||
@@ -585,11 +735,12 @@ def main():
|
||||
] if val is None or val == ""
|
||||
]
|
||||
if wifi_trio_missing and not args.force_partial:
|
||||
state_key = mac_state_key(chip_mac) if chip_mac else args.port
|
||||
parser.error(
|
||||
f"Missing required WiFi credentials after merging prior state: "
|
||||
f"{', '.join(wifi_trio_missing)}.\n"
|
||||
f"\n"
|
||||
f" No per-port state file at {_state_path_for(args.port, args.state_dir)}\n"
|
||||
f" No saved state at {_state_path_for(state_key, args.state_dir)}\n"
|
||||
f" and the CLI didn't include them. Either pass --ssid + --password + --target-ip\n"
|
||||
f" on this run, or add --force-partial to flash without WiFi.\n"
|
||||
)
|
||||
@@ -691,7 +842,7 @@ def main():
|
||||
# Persist merged state after a successful flash so future partial
|
||||
# invocations from this machine merge on top of what's actually on the
|
||||
# device. This is the heart of the additive-by-default fix (#391/#574).
|
||||
path = save_state(args.port, args.state_dir, merged)
|
||||
path = save_board_state(args.port, chip_mac, args.state_dir, merged, legacy)
|
||||
print(f"State persisted to {path}")
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,246 @@
|
||||
"""Tests for keying provision.py state by board MAC instead of serial port (#1755).
|
||||
|
||||
main() runs in-process. esptool's MAC read, the NVS generator and the flasher
|
||||
are all stubbed, so no serial port is opened and nothing is flashed. MACs and
|
||||
credentials are placeholders.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import csv
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
PROVISION_PATH = Path(__file__).resolve().parents[1] / "provision.py"
|
||||
SPEC = importlib.util.spec_from_file_location("provision", PROVISION_PATH)
|
||||
provision = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(provision)
|
||||
|
||||
PORT = "/dev/cu.usbmodem-test"
|
||||
OTHER_PORT = "/dev/cu.usbmodem-other"
|
||||
MAC_A = "02:00:00:00:00:0a"
|
||||
MAC_B = "02:00:00:00:00:0b"
|
||||
CREDS = ("--ssid", "test-ssid", "--password", "fake-pass-not-real",
|
||||
"--target-ip", "192.0.2.10")
|
||||
|
||||
# Shapes of `esptool read_mac` output (esptool/cmds.py read_mac in 4.x and 5.x).
|
||||
ESPTOOL_V5_S3 = "Connected to ESP32-S3 on /dev/x:\nMAC: 02:00:00:00:00:0a\n"
|
||||
ESPTOOL_V4_S3 = "Chip is ESP32-S3\nMAC: 02:00:00:00:00:0a\nHard resetting via RTS pin...\n"
|
||||
ESPTOOL_V5_C6 = (
|
||||
"MAC: 02:00:00:ff:fe:00:00:0a\n"
|
||||
"BASE MAC: 02:00:00:00:00:0a\n"
|
||||
"MAC_EXT: ff:fe\n"
|
||||
)
|
||||
|
||||
|
||||
class TestParseEsptoolMac(unittest.TestCase):
|
||||
def test_six_byte_mac_line(self):
|
||||
self.assertEqual(provision.parse_esptool_mac(ESPTOOL_V5_S3), MAC_A)
|
||||
self.assertEqual(provision.parse_esptool_mac(ESPTOOL_V4_S3), MAC_A)
|
||||
|
||||
def test_eui64_chips_use_the_base_mac(self):
|
||||
# C5/C6 print an 8-byte "MAC:" line first; that isn't the base MAC.
|
||||
self.assertEqual(provision.parse_esptool_mac(ESPTOOL_V5_C6), MAC_A)
|
||||
|
||||
def test_no_mac_in_output(self):
|
||||
self.assertIsNone(provision.parse_esptool_mac("A fatal error occurred\n"))
|
||||
|
||||
def test_normalize_mac(self):
|
||||
self.assertEqual(provision.normalize_mac("02-00-00-00-00-0A"), MAC_A)
|
||||
self.assertIsNone(provision.normalize_mac("02:00:00:00:00"))
|
||||
self.assertIsNone(provision.normalize_mac("02:00:00:00:00:zz"))
|
||||
|
||||
|
||||
class TestReadChipMac(unittest.TestCase):
|
||||
def test_runs_esptool_read_mac_on_the_port(self):
|
||||
done = subprocess.CompletedProcess([], 0, stdout=ESPTOOL_V5_C6, stderr="")
|
||||
with mock.patch.object(provision.subprocess, "run", return_value=done) as run:
|
||||
self.assertEqual(provision.read_chip_mac(PORT, 115200, "auto"), MAC_A)
|
||||
cmd = run.call_args.args[0]
|
||||
self.assertIn("read_mac", cmd)
|
||||
self.assertEqual(cmd[cmd.index("--port") + 1], PORT)
|
||||
|
||||
def test_failure_returns_none(self):
|
||||
failed = subprocess.CompletedProcess([], 2, stdout="", stderr="no port")
|
||||
with mock.patch.object(provision.subprocess, "run", return_value=failed):
|
||||
self.assertIsNone(provision.read_chip_mac(PORT, 115200, "auto"))
|
||||
with mock.patch.object(provision.subprocess, "run",
|
||||
side_effect=subprocess.TimeoutExpired("esptool", 30)):
|
||||
self.assertIsNone(provision.read_chip_mac(PORT, 115200, "auto"))
|
||||
|
||||
|
||||
class _MainCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.root = tempfile.mkdtemp(prefix="provision-mac-")
|
||||
self.state_dir = os.path.join(self.root, "state")
|
||||
self.flashed_csv = []
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.root, ignore_errors=True)
|
||||
|
||||
def run_main(self, *argv, mac=None):
|
||||
"""Run main() with the board `mac` attached; return (code, stdout, stderr)."""
|
||||
def fake_generate(csv_content, _size):
|
||||
self.flashed_csv.append(csv_content)
|
||||
return b"nvs"
|
||||
|
||||
out, err = io.StringIO(), io.StringIO()
|
||||
code = 0
|
||||
old_cwd = os.getcwd()
|
||||
os.chdir(self.root)
|
||||
try:
|
||||
with mock.patch.object(sys, "argv", ["provision.py", *argv,
|
||||
"--state-dir", self.state_dir]), \
|
||||
mock.patch.object(provision, "read_chip_mac", return_value=mac,
|
||||
create=True) as read, \
|
||||
mock.patch.object(provision, "generate_nvs_binary",
|
||||
side_effect=fake_generate), \
|
||||
mock.patch.object(provision, "flash_nvs"), \
|
||||
contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
|
||||
self.read_mac = read
|
||||
try:
|
||||
provision.main()
|
||||
except SystemExit as exc:
|
||||
code = exc.code
|
||||
finally:
|
||||
os.chdir(old_cwd)
|
||||
return code, out.getvalue(), err.getvalue()
|
||||
|
||||
def last_flashed(self):
|
||||
rows = csv.DictReader(io.StringIO(self.flashed_csv[-1]))
|
||||
return {row["key"]: row["value"] for row in rows}
|
||||
|
||||
def board_file(self, mac):
|
||||
return provision._state_path_for(provision.mac_state_key(mac), self.state_dir)
|
||||
|
||||
|
||||
class TestBoardKeyedState(_MainCase):
|
||||
def test_second_board_on_same_port_does_not_inherit_node_id(self):
|
||||
self.assertEqual(self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A)[0], 0)
|
||||
|
||||
code, _, _ = self.run_main("--port", PORT, *CREDS, mac=MAC_B)
|
||||
|
||||
self.assertEqual(code, 0)
|
||||
self.assertNotIn("node_id", self.last_flashed())
|
||||
|
||||
def test_second_board_without_credentials_is_not_given_the_first_boards(self):
|
||||
self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A)
|
||||
|
||||
code, _, err = self.run_main("--port", PORT, "--node-id", "3", mac=MAC_B)
|
||||
|
||||
self.assertEqual(code, 2)
|
||||
self.assertIn("Missing required WiFi credentials", err)
|
||||
|
||||
def test_board_keeps_its_state_on_a_different_port(self):
|
||||
self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A)
|
||||
|
||||
code, _, _ = self.run_main("--port", OTHER_PORT, "--zone", "lab", mac=MAC_A)
|
||||
|
||||
self.assertEqual(code, 0)
|
||||
flashed = self.last_flashed()
|
||||
self.assertEqual(flashed["node_id"], "2")
|
||||
self.assertEqual(flashed["ssid"], "test-ssid")
|
||||
|
||||
def test_record_carries_board_identity(self):
|
||||
self.run_main("--port", PORT, *CREDS, mac=MAC_A)
|
||||
|
||||
with open(self.board_file(MAC_A), encoding="utf-8") as f:
|
||||
saved = json.load(f)
|
||||
self.assertEqual(saved[provision.STATE_MAC_KEY], MAC_A)
|
||||
self.assertEqual(saved[provision.STATE_PORT_KEY], PORT)
|
||||
self.assertNotIn("_chip_mac", self.flashed_csv[-1])
|
||||
|
||||
def test_unreadable_mac_falls_back_to_port_keyed_state(self):
|
||||
code, _, err = self.run_main("--port", PORT, *CREDS, mac=None)
|
||||
|
||||
self.assertEqual(code, 0)
|
||||
self.assertIn("could not read the board's MAC", err)
|
||||
self.assertTrue(os.path.isfile(provision._state_path_for(PORT, self.state_dir)))
|
||||
|
||||
def test_reset_ignores_the_boards_record(self):
|
||||
self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A)
|
||||
|
||||
self.run_main("--port", PORT, "--reset", *CREDS, mac=MAC_A)
|
||||
|
||||
self.assertNotIn("node_id", self.last_flashed())
|
||||
|
||||
def test_invalid_mac_flag_is_rejected(self):
|
||||
code, _, err = self.run_main("--port", PORT, "--mac", "not-a-mac", "--state")
|
||||
self.assertEqual(code, 2)
|
||||
self.assertIn("--mac", err)
|
||||
|
||||
|
||||
class TestMigration(_MainCase):
|
||||
def write_legacy(self, port, state):
|
||||
os.makedirs(self.state_dir, exist_ok=True)
|
||||
path = provision._state_path_for(port, self.state_dir)
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
json.dump(state, f)
|
||||
return path
|
||||
|
||||
def test_port_keyed_record_moves_to_the_board_and_is_retired(self):
|
||||
legacy = self.write_legacy(PORT, {
|
||||
"ssid": "test-ssid", "password": "fake-pass-not-real",
|
||||
"target_ip": "192.0.2.10", "node_id": 5,
|
||||
})
|
||||
|
||||
code, _, err = self.run_main("--port", PORT, "--zone", "lab", mac=MAC_A)
|
||||
|
||||
self.assertEqual(code, 0)
|
||||
self.assertIn("port-keyed state", err)
|
||||
self.assertEqual(self.last_flashed()["node_id"], "5")
|
||||
self.assertFalse(os.path.exists(legacy))
|
||||
with open(self.board_file(MAC_A), encoding="utf-8") as f:
|
||||
self.assertEqual(json.load(f)["node_id"], 5)
|
||||
|
||||
# The next board on that port starts clean.
|
||||
self.run_main("--port", PORT, *CREDS, mac=MAC_B)
|
||||
self.assertNotIn("node_id", self.last_flashed())
|
||||
|
||||
def test_board_record_wins_over_a_port_keyed_file(self):
|
||||
self.run_main("--port", OTHER_PORT, *CREDS, "--node-id", "2", mac=MAC_A)
|
||||
self.write_legacy(PORT, {"node_id": 9})
|
||||
|
||||
code, _, _ = self.run_main("--port", PORT, "--zone", "lab", mac=MAC_A)
|
||||
|
||||
self.assertEqual(code, 0)
|
||||
self.assertEqual(self.last_flashed()["node_id"], "2")
|
||||
|
||||
|
||||
class TestNoSerialForInspection(_MainCase):
|
||||
def test_state_shows_last_board_on_port_without_opening_it(self):
|
||||
self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A)
|
||||
|
||||
code, out, err = self.run_main("--port", PORT, "--state", mac=MAC_B)
|
||||
|
||||
self.assertEqual(code, 0)
|
||||
self.read_mac.assert_not_called()
|
||||
self.assertEqual(json.loads(out)["node_id"], 2)
|
||||
self.assertIn(MAC_A, err)
|
||||
|
||||
def test_state_with_mac_flag_selects_that_board(self):
|
||||
self.run_main("--port", PORT, *CREDS, "--node-id", "2", mac=MAC_A)
|
||||
self.run_main("--port", PORT, *CREDS, "--node-id", "3", mac=MAC_B)
|
||||
|
||||
code, out, _ = self.run_main("--port", PORT, "--state", "--mac", MAC_A.upper())
|
||||
|
||||
self.assertEqual(code, 0)
|
||||
self.assertEqual(json.loads(out)["node_id"], 2)
|
||||
|
||||
def test_dry_run_does_not_open_the_port(self):
|
||||
code, _, _ = self.run_main("--port", PORT, *CREDS, "--dry-run", mac=MAC_A)
|
||||
|
||||
self.assertEqual(code, 0)
|
||||
self.read_mac.assert_not_called()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user