fix(lf): BigBuf download spoke the wrong protocol — LF sample download hung

download_samples sent a DOWNLOAD_BIGBUF request per chunk and read replies with the
NG-only reader. But the firmware answers ONE request by streaming the data as OLD-format
CMD_DOWNLOADED_BIGBUF frames (no magic) then a single NG ACK (appmain.c) — so the NG
reader scanned for a magic that never came and timed out (hardware repro: lf.read() ok,
download_samples -> Response timeout). This blocked all LF identify on real hardware.

- transport.download_bulk: send the request once, then read with the existing NG/OLD-
  agnostic reader, accumulating CMD_DOWNLOADED_BIGBUF payloads (trimmed to oldarg[1], not
  the 512-byte padding) until the ACK or `count` bytes; drains the trailing ACK so it
  can't poison the next command. Reusable for EML/SPIFFS/flashmem bulk reads.
- download_samples: one call into download_bulk instead of the broken per-chunk loop.

Tests: count-terminated + ACK-terminated streams, oldarg-length trimming, single request.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
michael
2026-07-14 16:11:11 -07:00
parent 5dd8e086cc
commit 2f221644f9
3 changed files with 97 additions and 27 deletions

View File

@@ -79,3 +79,57 @@ def test_decode_response_frame_error_status():
resp = decode_response_frame(frame)
assert resp["status"] == -4 # ETIMEOUT
assert resp["data"] == b""
# --- bulk download (BigBuf): one request -> OLD data frames -> ACK -----------------------------
import asyncio
from unittest.mock import AsyncMock
from pm3py.core.transport import PM3Transport, PM3Response
from pm3py.core.protocol import PM3_CMD_DATA_SIZE
def _run(coro):
return asyncio.get_event_loop().run_until_complete(coro)
def _bigbuf_frame(offset, data):
# OLD-format CMD_DOWNLOADED_BIGBUF: oldarg=[offset, len, tracelen], data padded to 512
return PM3Response(cmd=int(Cmd.DOWNLOADED_BIGBUF), status=0, reason=0, ng=False,
data=data + b"\x00" * (PM3_CMD_DATA_SIZE - len(data)),
oldarg=[offset, len(data), 4242])
def _ack():
return PM3Response(cmd=int(Cmd.ACK), status=0, reason=0, ng=False, data=b"", oldarg=[1, 0, 0])
def _transport_with(frames):
t = PM3Transport("/dev/null")
t.send_frame = AsyncMock()
t._read_any_response = AsyncMock(side_effect=frames)
return t
def test_download_bulk_count_terminated():
payload = bytes((i * 7) % 256 for i in range(1224))
t = _transport_with([_bigbuf_frame(0, payload[0:512]), _bigbuf_frame(512, payload[512:1024]),
_bigbuf_frame(1024, payload[1024:1224]), _ack()])
got = _run(t.download_bulk(Cmd.DOWNLOAD_BIGBUF, 0, 1224, Cmd.DOWNLOADED_BIGBUF, Cmd.ACK))
assert got == payload # exactly count bytes, reassembled in order
assert t.send_frame.call_count == 1 # ONE request, not one per chunk
def test_download_bulk_ack_terminated_early():
# device has fewer bytes than requested -> stops at the ACK, returns what came
payload = bytes(range(200))
t = _transport_with([_bigbuf_frame(0, payload), _ack()])
got = _run(t.download_bulk(Cmd.DOWNLOAD_BIGBUF, 0, 99999, Cmd.DOWNLOADED_BIGBUF, Cmd.ACK))
assert got == payload
def test_download_bulk_uses_oldarg_length_not_padding():
# the OLD frame is always 512 bytes; only oldarg[1] bytes are valid (rest is padding)
t = _transport_with([_bigbuf_frame(0, b"\xAA\xBB\xCC"), _ack()])
got = _run(t.download_bulk(Cmd.DOWNLOAD_BIGBUF, 0, 3, Cmd.DOWNLOADED_BIGBUF, Cmd.ACK))
assert got == b"\xAA\xBB\xCC" # trimmed to oldarg length, not 512 of padding