review: refresh upgrade docs to validated state; assert appendData params in test

Addresses Qodo review on #3:
- upgrade_firmware docstring + confirm ValueError no longer claim the path is
  byte-unproven / never run on hardware (it is validated end-to-end on a
  GK7205V510); docstring now states the {"length":N}+binary contract and the
  prepare/appendData/execute state machine.
- test_upgrade_streams_file_in_chunks now records each appendData params and the
  real trailing-binary length and asserts params == {"length": N} with N equal
  to the payload for every chunk. FakeDHIPServer exposes the binary payload
  (__data__) so handlers can verify it. A revert to {"Offset","Length"} now
  fails the test (previously it passed).

Author: widgetii

dahua/client.py

@@ -472,18 +472,23 @@ def upgrade_firmware(self, path: str, *, confirm: bool = False,
         to). *path* is the vendor upgrade package (a Dahua "zzip" — see
         ``tools/zzip.py`` in the zenointel project), not a raw partition image.
 
+        ``appendData`` is sent params ``{"length": len(chunk)}`` (lowercase) with
+        the chunk as the trailing binary payload — the handler rejects the call
+        with ``400 "param error"`` unless ``params.length`` equals the actual
+        payload length. State machine: ``prepare`` (0→2) → ``appendData`` (2/4,
+        repeatable) → ``execute`` (4→0), with a ~60 s idle timeout.
+
         .. danger::
-           This can permanently **brick** the device. The orchestration is
-           validated against a mock and the method names are reversed from
-           firmware, but the JSON param names are not byte-proven and this has
-           deliberately never been run on hardware. Probe :meth:`firmware_state`
-           first, keep a UART/backup recovery path ready, and pass
-           ``confirm=True`` to proceed.
+           This can permanently **brick** the device. The flow is validated
+           end-to-end on hardware (a GK7205V510 flashed an OpenIPC package and
+           rebooted into it), but any wrong package for the target still bricks
+           it. Probe :meth:`firmware_state` first, keep a UART/backup recovery
+           path ready, and pass ``confirm=True`` to proceed.
         """
         if not confirm:
             raise ValueError(
-                "upgrade_firmware can brick the device and is untested on "
-                "hardware; pass confirm=True to proceed")
+                "upgrade_firmware writes device partitions and can brick the "
+                "device; pass confirm=True to proceed")
         import os
         total = os.path.getsize(path)
         self.call(const.UPGRADER_PREPARE, {"Type": fw_type})

tests/fake_server.py

@@ -57,7 +57,12 @@ def _read_frame(self, sock):
         _, magic, sess, rid, pkg, idx, mlen, dlen = struct.unpack(HEADER_FMT, hdr)
         assert magic == DHIP_MAGIC
         body = self._recv(sock, pkg) if pkg else b""
-        return json.loads(body[:mlen].decode()), sess
+        obj = json.loads(body[:mlen].decode())
+        if dlen:
+            # expose the trailing binary payload so handlers can verify a request
+            # whose params must describe it (e.g. upgrader.appendData length).
+            obj["__data__"] = body[mlen:mlen + dlen]
+        return obj, sess
 
     def _send(self, sock, obj, data=b"", index=0):
         body = json.dumps(obj, separators=(",", ":")).encode()

tests/test_dahua.py

@@ -316,18 +316,20 @@ def test_upgrade_requires_confirm(self):
                     cam.upgrade_firmware(__file__)  # no confirm=True
 
     def test_upgrade_streams_file_in_chunks(self):
-        received = bytearray()
+        append_params = []
 
         def send(req):
+            # The device rejects appendData unless params is exactly
+            # {"length": <payload length>}. Record params + the actual binary
+            # payload length so a regression to {"Offset","Length"} is caught.
+            append_params.append((req.get("params"), len(req.get("__data__", b""))))
             return {"result": True}
 
         handlers = {
             "upgrader.prepare": lambda r: {"result": True},
             "upgrader.appendData": send,
             "upgrader.execute": lambda r: {"result": True},
         }
-        # The fake server doesn't expose binary bodies to handlers, so assert
-        # the orchestration (start/send*/execute) and chunk count instead.
         with FakeDHIPServer(handlers) as srv:
             with DahuaClient("127.0.0.1", srv.port) as cam:
                 cam.login(USER, PASS, keep_alive=False)
@@ -344,6 +346,12 @@ def send(req):
                 self.assertIn("upgrader.prepare", srv.received)
                 self.assertIn("upgrader.execute", srv.received)
                 self.assertEqual(seen[-1], (10000, 10000))
+                # params must be exactly {"length": N} with N == the real payload
+                # length for every chunk (4096, 4096, 1808) — nothing else.
+                self.assertEqual([p for p, _ in append_params],
+                                 [{"length": 4096}, {"length": 4096}, {"length": 1808}])
+                self.assertEqual([(p["length"], n) for p, n in append_params],
+                                 [(4096, 4096), (4096, 4096), (1808, 1808)])
 
 
 class TestDiscovery(unittest.TestCase):