From 5a36be17d24c5c8580671c5ec87278b32d147af4 Mon Sep 17 00:00:00 2001 From: Mathew Beane Date: Fri, 2 Oct 2026 16:03:30 -0400 Subject: [PATCH] fix(provision): make --dry-run and failed --reset leave state intact --dry-run persisted the merged state record, so the next real run merged on top of values that never reached the device. It now skips save_state. The NVS binary is still written to nvs_provision.bin because ADR-061 and scripts/qemu-mesh-test.sh consume it. --reset deleted the per-port state file before argument validation, so a run that then failed (e.g. missing WiFi credentials) still lost the record. It now only ignores the prior state; a successful flash overwrites the file as before. Also fix the module docstring's package name: the PyPI package is esp-idf-nvs-partition-gen, matching the runtime hint. Fixes #2096 Co-Authored-By: claude-flow --- firmware/esp32-csi-node/provision.py | 18 +-- .../tests/test_provision_dry_run_reset.py | 133 ++++++++++++++++++ 2 files changed, 143 insertions(+), 8 deletions(-) create mode 100644 firmware/esp32-csi-node/tests/test_provision_dry_run_reset.py diff --git a/firmware/esp32-csi-node/provision.py b/firmware/esp32-csi-node/provision.py index 88172fe4e..0b2f9c582 100644 --- a/firmware/esp32-csi-node/provision.py +++ b/firmware/esp32-csi-node/provision.py @@ -11,7 +11,7 @@ Usage: --password "..." --target-ip 192.168.1.20 Requirements: - pip install 'esptool>=5.0' nvs-partition-gen + pip install 'esptool>=5.0' esp-idf-nvs-partition-gen (or use the nvs_partition_gen.py bundled with ESP-IDF) ADDITIVE-BY-DEFAULT (issue #391, #574 phase 1): @@ -360,7 +360,8 @@ def main(): "error when no prior state file exists. The script now merges " "with prior state by default, so this flag is rarely needed.") parser.add_argument("--reset", action="store_true", - help="Wipe this machine's per-port state file before merging. " + help="Ignore this machine's per-port state file when merging; it is " + "replaced after a successful flash. " "Use for first-time provisioning of a recycled board where " "previously-staged keys should NOT be re-applied.") parser.add_argument("--state-dir", default=_default_state_dir(), @@ -373,10 +374,12 @@ def main(): # --- Per-port state load + merge (additive-by-default, #391 / #574) --- if args.reset: + # Don't delete the state file here: validation below can still fail, + # and a failed run must not lose the record. A successful flash + # overwrites it with the reset (CLI-only) state. path = _state_path_for(args.port, args.state_dir) if os.path.isfile(path): - os.unlink(path) - print(f"--reset: removed state file {path}", file=sys.stderr) + print(f"--reset: ignoring state file {path}", file=sys.stderr) prior = {} else: prior = load_state(args.port, args.state_dir) @@ -502,10 +505,9 @@ def main(): print(f"NVS binary saved to {out} ({len(nvs_bin)} bytes)") print(f"Flash manually: python -m esptool --chip {args.chip} --port {args.port} " f"write_flash 0x9000 {out}") - # Persist merged state even on dry-run so a subsequent real flash from - # this machine sees the same staged config. - path = save_state(args.port, args.state_dir, merged) - print(f"State persisted to {path}") + # Don't persist state on dry-run: nothing reached the device, so the + # next real run must not merge on top of values that were never flashed. + print("Dry run: state file not updated.") return flash_nvs(args.port, args.baud, nvs_bin, args.chip) diff --git a/firmware/esp32-csi-node/tests/test_provision_dry_run_reset.py b/firmware/esp32-csi-node/tests/test_provision_dry_run_reset.py new file mode 100644 index 000000000..da240cb7f --- /dev/null +++ b/firmware/esp32-csi-node/tests/test_provision_dry_run_reset.py @@ -0,0 +1,133 @@ +"""Side-effect tests for provision.py --dry-run and --reset (#2096). + +main() runs in-process with the NVS generator and the flasher stubbed out, so +no serial port is opened and no ESP-IDF tooling is needed. State lives in a +per-test temp dir passed via --state-dir. +""" + +import contextlib +import importlib.util +import io +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/null-test" +FAKE_NVS = b"\xab" * 64 +PRIOR = { + "ssid": "TESTSSID", + "password": "TESTPASS", + "target_ip": "127.0.0.1", + "node_id": 1, +} +WIFI = ["--ssid", "TESTSSID", "--password", "TESTPASS", "--target-ip", "127.0.0.1"] + + +class ProvisionSideEffectTests(unittest.TestCase): + def setUp(self): + self.root = tempfile.mkdtemp(prefix="provision-side-effects-") + self.state_dir = os.path.join(self.root, "state") + self.cwd = os.path.join(self.root, "cwd") + os.makedirs(self.cwd) + self.state_path = provision._state_path_for(PORT, self.state_dir) + self.flash = mock.Mock() + + def tearDown(self): + shutil.rmtree(self.root, ignore_errors=True) + + def run_main(self, *argv): + """Run provision.main() with stubs; return the SystemExit code or None.""" + old_cwd = os.getcwd() + os.chdir(self.cwd) + try: + with mock.patch.object(sys, "argv", ["provision.py", "--port", PORT, + "--state-dir", self.state_dir, *argv]), \ + mock.patch.object(provision, "generate_nvs_binary", + return_value=FAKE_NVS), \ + mock.patch.object(provision, "flash_nvs", self.flash), \ + contextlib.redirect_stdout(io.StringIO()), \ + contextlib.redirect_stderr(io.StringIO()): + provision.main() + except SystemExit as exc: + return exc.code + finally: + os.chdir(old_cwd) + return None + + def seed_state(self): + provision.save_state(PORT, self.state_dir, PRIOR) + with open(self.state_path, "rb") as f: + return f.read() + + def read_state_bytes(self): + with open(self.state_path, "rb") as f: + return f.read() + + def test_dry_run_does_not_create_state(self): + self.assertIsNone(self.run_main(*WIFI, "--node-id", "1", "--dry-run")) + self.assertFalse(os.path.exists(self.state_path)) + self.assertFalse(os.path.exists(self.state_dir)) + self.flash.assert_not_called() + + def test_dry_run_does_not_modify_existing_state(self): + before = self.seed_state() + self.assertIsNone(self.run_main("--node-id", "7", "--dry-run")) + self.assertEqual(self.read_state_bytes(), before) + self.flash.assert_not_called() + + def test_dry_run_still_writes_requested_binary(self): + # The binary is the dry-run's documented output (ADR-061, + # scripts/qemu-mesh-test.sh), so it is kept. + self.run_main(*WIFI, "--dry-run") + with open(os.path.join(self.cwd, "nvs_provision.bin"), "rb") as f: + self.assertEqual(f.read(), FAKE_NVS) + + def test_failed_reset_keeps_state(self): + before = self.seed_state() + # The issue's repro: no WiFi trio after reset, so validation fails. + self.assertEqual(self.run_main("--reset", "--node-id", "3"), 2) + self.assertEqual(self.read_state_bytes(), before) + self.flash.assert_not_called() + + def test_reset_dry_run_keeps_state(self): + before = self.seed_state() + self.assertIsNone(self.run_main("--reset", *WIFI, "--node-id", "3", "--dry-run")) + self.assertEqual(self.read_state_bytes(), before) + + def test_reset_with_failed_flash_keeps_state(self): + before = self.seed_state() + self.flash.side_effect = subprocess.CalledProcessError(2, "esptool") + with self.assertRaises(subprocess.CalledProcessError): + self.run_main("--reset", *WIFI, "--node-id", "3") + self.assertEqual(self.read_state_bytes(), before) + + def test_reset_with_successful_flash_replaces_state(self): + provision.save_state(PORT, self.state_dir, {**PRIOR, "zone": "lobby"}) + self.assertIsNone(self.run_main("--reset", *WIFI, "--node-id", "3")) + self.flash.assert_called_once() + state = provision.load_state(PORT, self.state_dir) + self.assertEqual(state["node_id"], 3) + self.assertNotIn("zone", state) + + def test_real_flash_still_persists_merged_state(self): + self.seed_state() + self.assertIsNone(self.run_main("--zone", "lobby")) + self.flash.assert_called_once() + state = provision.load_state(PORT, self.state_dir) + self.assertEqual(state["zone"], "lobby") + for key, value in PRIOR.items(): + self.assertEqual(state[key], value) + + +if __name__ == "__main__": + unittest.main()