Merge PR #2111 (aepod) into release integration

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)

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()