From 861bd2aad684c09cb5c2df77c088e7e13a85e6f4 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Thu, 15 Oct 2026 11:05:00 -0700 Subject: [PATCH] Preserve BLE command replies and respect negotiated mOTA MTU --- .github/workflows/run-unit-tests.yml | 1 + examples/companion_radio/MyMesh.cpp | 16 ++ examples/companion_radio/main.cpp | 12 +- src/helpers/CompanionBleBench.h | 39 ++++ src/helpers/esp32/SerialBLEInterface.cpp | 34 ++- src/helpers/esp32/SerialBLEInterface.h | 1 + src/helpers/nrf52/SerialBLEInterface.cpp | 26 ++- src/helpers/nrf52/SerialBLEInterface.h | 2 + src/helpers/ota/MotaSourceSerial.cpp | 20 +- src/helpers/ota/MotaSourceSerial.h | 7 + .../nimble_companion/mocks/NimBLEDevice.h | 2 + .../test_nimble_companion.cpp | 162 +++++++++++++- test/test_ble_visibility_probe.py | 69 ++++++ test/test_companion_ble_bench.py | 181 ++++++++++++++++ test/test_mota_serial_deadline.py | 161 ++++++++++++++ test/test_nimble_companion.py | 26 +++ test/test_nrf52_ble_backpressure.py | 199 ++++++++++++++++++ tools/hil/ble_visibility_probe.py | 163 ++++++++++++++ 18 files changed, 1111 insertions(+), 10 deletions(-) create mode 100644 src/helpers/CompanionBleBench.h create mode 100644 test/test_ble_visibility_probe.py create mode 100644 test/test_companion_ble_bench.py create mode 100644 test/test_nrf52_ble_backpressure.py create mode 100644 tools/hil/ble_visibility_probe.py diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 236513f5f..548c8ca5e 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -622,6 +622,7 @@ jobs: python3 -B test/test_local_release_capacity_archive.py python3 -B test/test_t096_full_memory.py python3 -B test/test_nrf52_ble_startup.py + python3 -B test/test_nrf52_ble_backpressure.py python3 -B test/test_nrf52_ble_trace.py python3 -B test/test_nrf52_ble_debug_tools.py python3 -B test/test_nrf52_usb_power.py diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 58a8fdf2b..0f8b0fb93 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -25,6 +25,9 @@ #endif #include #include +#if defined(MESH_HIL_BLE_BENCH) && MESH_HIL_BLE_BENCH +#include +#endif #include #include #include @@ -5742,6 +5745,19 @@ void MyMesh::handleCmdFrame(size_t len) { return; } +#if defined(MESH_HIL_BLE_BENCH) && MESH_HIL_BLE_BENCH + if (cmd_frame[0] == mesh::companion::kBleBenchCommand) { + // Direct Companion transport benchmark only: no RF dispatch, settings, + // identity reads or persistent state. Reuse the normal reply queue and + // its required-frame admission/backpressure, including the source route. + const size_t echoed = mesh::companion::buildBleBenchEcho( + cmd_frame, len, out_frame, MAX_FRAME_SIZE); + if (echoed) _serial->writeFrame(out_frame, echoed); + else writeErrFrame(ERR_CODE_ILLEGAL_ARG); + return; + } +#endif + servicePendingSerialReply(); const bool starts_long_lived_request = (cmd_frame[0] == CMD_SEND_LOGIN && len >= 1 + PUB_KEY_SIZE) diff --git a/examples/companion_radio/main.cpp b/examples/companion_radio/main.cpp index 21fbf5d49..e560df6b7 100644 --- a/examples/companion_radio/main.cpp +++ b/examples/companion_radio/main.cpp @@ -469,7 +469,17 @@ public: } private: - mesh::ota::SerialMotaSource _source; + class Source : public mesh::ota::SerialMotaSource { + public: + using mesh::ota::SerialMotaSource::SerialMotaSource; + protected: + uint16_t responseFrameCapacity() const override { + return bluetooth_interface.motaResponseCapacity(); + } + }; + static_assert(sizeof(Source) == sizeof(mesh::ota::SerialMotaSource), + "BLE source MTU policy must not allocate additional state"); + Source _source; uint32_t _packets_sent_at_start; uint32_t _last_packets_sent; }; diff --git a/src/helpers/CompanionBleBench.h b/src/helpers/CompanionBleBench.h new file mode 100644 index 000000000..0913dd9b7 --- /dev/null +++ b/src/helpers/CompanionBleBench.h @@ -0,0 +1,39 @@ +#pragma once + +// Explicit lab images only. Normal release builds expose no echo command, +// allocate no benchmark state and retain their existing command behavior. +#if defined(MESH_HIL_BLE_BENCH) && MESH_HIL_BLE_BENCH +#if MESH_HIL_BLE_BENCH != 1 +#error "MESH_HIL_BLE_BENCH must be explicitly 0 or 1" +#endif + +#include +#include +#include +#include "CompanionFrameLimits.h" + +namespace mesh { +namespace companion { + +// 0x78..0x7F already have legacy hardware aliases. 0x70 is lab-only and is +// distinct from normal responses/pushes; nonce and sequence belong to the host. +static constexpr uint8_t kBleBenchCommand = 0x70; +static constexpr size_t kBleBenchHeaderBytes = 15; + +// Wire: 70, "BLE1", runIdLE32, sequenceLE32, payloadLengthLE16, payload. +// Echo the entire request exactly once, without generic OK or radio traffic. +// The host also limits frame size to its negotiated ATT MTU minus three. +inline size_t buildBleBenchEcho(const uint8_t* request, size_t length, + uint8_t* response, size_t capacity) { + if (!request || !response || length < kBleBenchHeaderBytes + || length > MAX_FRAME_SIZE || length > capacity + || request[0] != kBleBenchCommand || memcmp(request + 1, "BLE1", 4)) return 0; + const size_t payload_length = size_t(request[13]) | (size_t(request[14]) << 8); + if (payload_length != length - kBleBenchHeaderBytes) return 0; + memcpy(response, request, length); + return length; +} + +} // namespace companion +} // namespace mesh +#endif diff --git a/src/helpers/esp32/SerialBLEInterface.cpp b/src/helpers/esp32/SerialBLEInterface.cpp index bb278319d..2d14bb8a9 100644 --- a/src/helpers/esp32/SerialBLEInterface.cpp +++ b/src/helpers/esp32/SerialBLEInterface.cpp @@ -977,14 +977,38 @@ size_t SerialBLEInterface::writeFrame(const uint8_t src[], size_t len) { return 0; } -#define BLE_WRITE_MIN_INTERVAL 60 +#if defined(MESH_HIL_BLE_INTERVAL_MS) +#if !defined(MESH_HIL_BLE_BENCH) || MESH_HIL_BLE_BENCH != 1 +#error "MESH_HIL_BLE_INTERVAL_MS requires the explicit MESH_HIL_BLE_BENCH=1 lab image" +#endif +#if MESH_HIL_BLE_INTERVAL_MS < 7 || MESH_HIL_BLE_INTERVAL_MS > 60 +#error "MESH_HIL_BLE_INTERVAL_MS must be in 7..60 milliseconds" +#endif +#define BLE_WRITE_MIN_INTERVAL MESH_HIL_BLE_INTERVAL_MS +#else +// Preserve the shipped pacing until a separate lab image has measured a safe +// replacement. One value governs admission and both successful/failed sends. +#define BLE_WRITE_MIN_INTERVAL 60 +#endif bool SerialBLEInterface::isReadBusy() const { return uxQueueMessagesWaiting(recv_queue) > 0; } +bool SerialBLEInterface::hasRequiredReplyCapacity() const { + if (send_queue_len < FRAME_QUEUE_SIZE) return true; + // Match enqueueCompanionFrame's admission rule: a required response can + // displace best-effort traffic, but must never displace another required one. + for (int i = 0; i < send_queue_len; ++i) { + if (!mesh::companionFrameRequiresDelivery(send_queue[i].buf, send_queue[i].len)) + return true; + } + return false; +} + bool SerialBLEInterface::isWriteBusy() const { - return !mesh::bleElapsedAtLeast((uint32_t)millis(), _last_write, + return !hasRequiredReplyCapacity() + || !mesh::bleElapsedAtLeast((uint32_t)millis(), _last_write, BLE_WRITE_MIN_INTERVAL); } @@ -1052,7 +1076,11 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) { } Frame frame; - if (deviceConnected && xQueueReceive(recv_queue, &frame, 0) == pdTRUE) { + // Service TX above even when full, then retain a command until its required + // reply can be admitted. Dequeuing while all TX slots are required causes + // one-shot replies (including CMD66) to be silently lost by writeFrame(). + if (deviceConnected && hasRequiredReplyCapacity() + && xQueueReceive(recv_queue, &frame, 0) == pdTRUE) { memcpy(dest, frame.buf, frame.len); BLE_DEBUG_PRINTLN("readBytes: sz=%d, hdr=%d", (uint32_t) frame.len, (uint32_t) dest[0]); return frame.len; diff --git a/src/helpers/esp32/SerialBLEInterface.h b/src/helpers/esp32/SerialBLEInterface.h index 6385b5501..669493cb3 100644 --- a/src/helpers/esp32/SerialBLEInterface.h +++ b/src/helpers/esp32/SerialBLEInterface.h @@ -67,6 +67,7 @@ class SerialBLEInterface : public BaseSerialInterface, Frame send_queue[FRAME_QUEUE_SIZE]; void clearBuffers(); + bool hasRequiredReplyCapacity() const; void servicePendingTxReset(); void scheduleAdvertisingRestart(uint32_t now); void recoverStalledTx(const char* cause); diff --git a/src/helpers/nrf52/SerialBLEInterface.cpp b/src/helpers/nrf52/SerialBLEInterface.cpp index 4e0f62e8f..a3817d158 100644 --- a/src/helpers/nrf52/SerialBLEInterface.cpp +++ b/src/helpers/nrf52/SerialBLEInterface.cpp @@ -1194,6 +1194,17 @@ void SerialBLEInterface::loop() { } } +bool SerialBLEInterface::hasRequiredReplyCapacity() const { + if (send_queue_len < FRAME_QUEUE_SIZE) return true; + // Required responses may evict an unsent packet log, never another required + // reply. Match enqueueCompanionFrame without growing either bounded queue. + for (uint8_t i = 0; i < send_queue_len; ++i) { + if (!mesh::companionFrameRequiresDelivery(send_queue[i].buf, send_queue[i].len)) + return true; + } + return false; +} + size_t SerialBLEInterface::writeFrame(const uint8_t src[], size_t len) { if (len > MAX_FRAME_SIZE) { BLE_DEBUG_PRINTLN("writeFrame(), frame too big, len=%u", (unsigned)len); @@ -1285,7 +1296,9 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) { _tx_stall_watchdog.reset(); } - if (recv_queue_len > 0) { + // TX/recovery above must run even when full. Retain RX until a required + // command reply can be admitted instead of silently losing that reply. + if (recv_queue_len > 0 && hasRequiredReplyCapacity()) { size_t len = recv_queue[0].len; memcpy(dest, recv_queue[0].buf, len); @@ -1366,6 +1379,17 @@ bool SerialBLEInterface::isMotaChannelReady() { && _conn_handle != BLE_CONN_HANDLE_INVALID && _mota_request.notifyEnabled(_conn_handle); } + +uint16_t SerialBLEInterface::motaResponseCapacity() { + if (!isMotaChannelReady() || !_mota_stream.isActive()) return 0; + BLEConnection* connection = Bluefruit.Connection(_conn_handle); + if (!connection || !connection->connected()) return 0; + const uint16_t mtu = connection->getMtu(); + if (mtu < 23) return 0; + const uint16_t capacity = (uint16_t)(mtu - 3u); + return capacity < mesh::ota::BLE_MOTA_RESPONSE_MAX + ? capacity : mesh::ota::BLE_MOTA_RESPONSE_MAX; +} #endif bool SerialBLEInterface::isReadBusy() const { diff --git a/src/helpers/nrf52/SerialBLEInterface.h b/src/helpers/nrf52/SerialBLEInterface.h index 6fd91ec5d..e7dfbdcd3 100644 --- a/src/helpers/nrf52/SerialBLEInterface.h +++ b/src/helpers/nrf52/SerialBLEInterface.h @@ -109,6 +109,7 @@ class SerialBLEInterface : public BaseSerialInterface { Frame recv_queue[FRAME_QUEUE_SIZE]; void clearBuffers(); + bool hasRequiredReplyCapacity() const; void shiftSendQueueLeft(); void shiftRecvQueueLeft(); size_t writeBleUartFrame(const Frame& frame); @@ -209,6 +210,7 @@ public: #if COMPANION_FEATURE_BLE_MOTA_SOURCE Stream& motaStream() { return _mota_stream; } bool isMotaChannelReady(); + uint16_t motaResponseCapacity(); bool isMotaStreamActive() const { return _mota_stream.isActive(); } void setMotaStreamActive(bool active) { _mota_stream.setActive(active); diff --git a/src/helpers/ota/MotaSourceSerial.cpp b/src/helpers/ota/MotaSourceSerial.cpp index 6f51f0971..535c3600a 100644 --- a/src/helpers/ota/MotaSourceSerial.cpp +++ b/src/helpers/ota/MotaSourceSerial.cpp @@ -202,13 +202,21 @@ bool SerialMotaSource::describe(uint8_t idx, MotaDesc& out) { return true; } +uint16_t SerialMotaSource::responsePayloadCapacity() const { + uint16_t frame = responseFrameCapacity(); + if (frame < 8) return 0; // even a nonempty DEFLATE response must fit + if (frame > MOTA_SEEDER_READ_MAX + 5u) frame = MOTA_SEEDER_READ_MAX + 5u; + return (uint16_t)(frame - 5u); +} + bool SerialMotaSource::read(uint8_t idx, uint32_t off, uint8_t* buf, uint32_t len) { if (!buf && len != 0) return false; if (len > UINT32_MAX - off) return false; uint32_t done = 0; while (done < len) { - uint16_t chunk = (uint16_t)(len - done > MOTA_SEEDER_READ_MAX - ? MOTA_SEEDER_READ_MAX : len - done); + const uint16_t maximum = responsePayloadCapacity(); + if (maximum == 0) return false; // disconnected/invalid BLE session + uint16_t chunk = (uint16_t)(len - done > maximum ? maximum : len - done); uint8_t args[7]; args[0] = idx; wr_u32le(args + 1, off + done); @@ -223,6 +231,7 @@ bool SerialMotaSource::read_deflated_block(uint8_t idx, uint16_t block, uint8_t* uint16_t cap, uint16_t* len) { if (!buf || !len || cap == 0) return false; *len = 0; + if (responsePayloadCapacity() < 3) return false; uint8_t args[7] = { idx }; args[1] = (uint8_t)(block & 0xFF); args[2] = (uint8_t)(block >> 8); @@ -235,8 +244,11 @@ bool SerialMotaSource::read_deflated_block(uint8_t idx, uint16_t block, uint8_t* uint16_t done = 0; while (done < total) { - const uint16_t chunk = (uint16_t)((total - done > MOTA_SEEDER_DEFLATE_CHUNK_MAX) - ? MOTA_SEEDER_DEFLATE_CHUNK_MAX : total - done); + const uint16_t payload_capacity = responsePayloadCapacity(); + if (payload_capacity < 3) return false; + const uint16_t maximum = (uint16_t)(payload_capacity - 2u); + const uint16_t chunk = (uint16_t)((total - done > maximum) + ? maximum : total - done); args[3] = (uint8_t)(done & 0xFF); args[4] = (uint8_t)(done >> 8); args[5] = (uint8_t)(chunk & 0xFF); args[6] = (uint8_t)(chunk >> 8); uint8_t response[2 + MOTA_SEEDER_DEFLATE_CHUNK_MAX]; diff --git a/src/helpers/ota/MotaSourceSerial.h b/src/helpers/ota/MotaSourceSerial.h index f4888df95..62a4f0643 100644 --- a/src/helpers/ota/MotaSourceSerial.h +++ b/src/helpers/ota/MotaSourceSerial.h @@ -38,7 +38,14 @@ public: bool hasPendingResponse() const { return _response_pending; } void resetSessionState(); +protected: + // Full framed response capacity, including magic/op/status/checksum. The + // default preserves USB/UART/TCP geometry. A BLE source can override this + // using its current ATT MTU without storing another buffer or callback. + virtual uint16_t responseFrameCapacity() const { return 197; } + private: + uint16_t responsePayloadCapacity() const; // Send a request (op+args) and read its response header; on OK, `payload` (if non-null) receives // `payload_len` bytes. Returns true iff a well-formed OK response for `op` arrived in time. bool txn(uint8_t op, const uint8_t* args, uint8_t arglen, uint8_t* payload, uint32_t payload_len); diff --git a/test/fixtures/nimble_companion/mocks/NimBLEDevice.h b/test/fixtures/nimble_companion/mocks/NimBLEDevice.h index 2b77b4aad..bcbd49e9d 100644 --- a/test/fixtures/nimble_companion/mocks/NimBLEDevice.h +++ b/test/fixtures/nimble_companion/mocks/NimBLEDevice.h @@ -61,6 +61,7 @@ inline uint32_t pin = 0; inline bool bonding = false, mitm = false, sc = false; inline std::vector> transmitted; inline unsigned disconnects = 0; +inline unsigned notification_attempts = 0; } inline int ble_gap_wl_set(const ble_addr_t* peers, uint8_t count) { assert(count == 1); @@ -91,6 +92,7 @@ struct NimBLECharacteristic { void setValue(const uint8_t* data, size_t size) { value.assign(data, data + size); } bool notify(uint16_t handle = 0xffff) const { assert(handle == 7); + ++fake_nimble::notification_attempts; if (!fake_nimble::accept_notification) return false; fake_nimble::transmitted.push_back(value); return true; diff --git a/test/fixtures/nimble_companion/test_nimble_companion.cpp b/test/fixtures/nimble_companion/test_nimble_companion.cpp index ea9d5aa27..552441c80 100644 --- a/test/fixtures/nimble_companion/test_nimble_companion.cpp +++ b/test/fixtures/nimble_companion/test_nimble_companion.cpp @@ -10,7 +10,7 @@ using namespace fake_nimble; static void reset() { NimBLEDevice::deinit(true); failure.clear(); calls.clear(); transmitted.clear(); - persisted = true; accept_notification = true; disconnects = 0; + persisted = true; accept_notification = true; disconnects = 0; notification_attempts = 0; test_millis = 100; } @@ -199,11 +199,171 @@ static void test_real_transport_frames() { NimBLEDevice::deinit(true); } +static void test_required_reply_backpressure_services_tx_before_rx() { + reset(); SerialBLEInterface interface; + assert(interface.begin("MeshCore-", "backpressure", 123456)); + interface.enable(); + NimBLEConnInfo info; + authenticate(interface, info); + auto* server = NimBLEDevice::server.get(); + auto* tx = server->service->characteristics[0].get(); + auto* rx = server->service->characteristics[1].get(); + tx->callbacks->onSubscribe(tx, info, 1); + mtu = 179; + for (uint8_t i = 0; i < 4; ++i) { + const uint8_t reply[] = {0x0d, i}; + assert(interface.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + } + const uint8_t command[] = {66, 'A', '7', '|', 'g', 'e', 't', ' ', 'l', 'a', 'b', '.', 't', 'x'}; + rx->setValue(command, sizeof(command)); + rx->callbacks->onWrite(rx, info); + uint8_t received[MAX_FRAME_SIZE] = {}; + accept_notification = false; + // A command must retain its RX slot until its required reply can fit. + assert(interface.checkRecvFrame(received) == 0); + assert(interface.isReadBusy() && transmitted.empty()); + test_millis += 61; + assert(interface.isWriteBusy()); // Capacity, not just the 60 ms TX throttle. + assert(interface.checkRecvFrame(received) == 0); + assert(interface.isReadBusy() && transmitted.empty()); + + accept_notification = true; + test_millis += 61; + // TX must be serviced BEFORE the full-queue RX gate, or this deadlocks. + assert(interface.checkRecvFrame(received) == sizeof(command)); + assert(memcmp(received, command, sizeof(command)) == 0); + assert(!interface.isReadBusy() && transmitted.size() == 1); + const uint8_t cli_reply[] = {29, 'A', '7', '|', 'O', 'K'}; + assert(interface.writeFrame(cli_reply, sizeof(cli_reply)) == sizeof(cli_reply)); + for (unsigned i = 0; i < 6; ++i) { + test_millis += 61; + assert(interface.checkRecvFrame(received) == 0); + } + assert(transmitted.size() == 5); + for (uint8_t i = 0; i < 4; ++i) { + assert((transmitted[i] == std::vector{0x0d, i})); + } + assert(transmitted[4] == std::vector(cli_reply, cli_reply + sizeof(cli_reply))); + assert(!interface.hasPendingIO()); + NimBLEDevice::deinit(true); +} + +static void test_full_queue_best_effort_eviction_still_admits_rx() { + reset(); SerialBLEInterface interface; + assert(interface.begin("MeshCore-", "eviction", 123456)); + interface.enable(); + NimBLEConnInfo info; + authenticate(interface, info); + auto* server = NimBLEDevice::server.get(); + auto* tx = server->service->characteristics[0].get(); + auto* rx = server->service->characteristics[1].get(); + tx->callbacks->onSubscribe(tx, info, 1); + mtu = 179; + for (uint8_t i = 0; i < 3; ++i) { + const uint8_t log[] = {0x88, i}; + assert(interface.writeFrame(log, sizeof(log)) == sizeof(log)); + } + const uint8_t extra_log[] = {0x88, 9}; + assert(interface.writeFrame(extra_log, sizeof(extra_log)) == 0); + const uint8_t previous_reply[] = {0x0d, 7}; + assert(interface.writeFrame(previous_reply, sizeof(previous_reply)) == sizeof(previous_reply)); + const uint8_t command[] = {66, 'B', '8', '|', 'v', 'e', 'r'}; + rx->setValue(command, sizeof(command)); + rx->callbacks->onWrite(rx, info); + uint8_t received[MAX_FRAME_SIZE] = {}; + accept_notification = false; + // Full is not blocked when a required response can evict a packet log. + assert(!interface.isWriteBusy()); + assert(interface.checkRecvFrame(received) == sizeof(command)); + assert(memcmp(received, command, sizeof(command)) == 0); + const uint8_t cli_reply[] = {29, 'B', '8', '|', 'O', 'K'}; + assert(interface.writeFrame(cli_reply, sizeof(cli_reply)) == sizeof(cli_reply)); + assert(transmitted.empty()); + accept_notification = true; + for (unsigned i = 0; i < 6; ++i) { + test_millis += 61; + assert(interface.checkRecvFrame(received) == 0); + } + assert(transmitted.size() == 4); + assert(transmitted[0] == std::vector(previous_reply, previous_reply + sizeof(previous_reply))); + assert(transmitted[1] == std::vector(cli_reply, cli_reply + sizeof(cli_reply))); + assert((transmitted[2] == std::vector{0x88, 0})); + assert((transmitted[3] == std::vector{0x88, 1})); + assert(!interface.hasPendingIO()); + NimBLEDevice::deinit(true); +} + +static void test_exact_notify_interval_failure_retention_and_rollover() { +#if defined(MESH_HIL_BLE_INTERVAL_MS) + constexpr uint32_t interval = MESH_HIL_BLE_INTERVAL_MS; +#else + constexpr uint32_t interval = 60; +#endif + reset(); SerialBLEInterface interface; + assert(interface.begin("MeshCore-", "interval", 123456)); + interface.enable(); + NimBLEConnInfo info; authenticate(interface, info); + auto* tx = NimBLEDevice::server->service->characteristics[0].get(); + uint8_t first[MAX_FRAME_SIZE] = {0x70, 1}, second[MAX_FRAME_SIZE] = {0x70, 2}; + uint8_t received[MAX_FRAME_SIZE] = {}; + assert(interface.writeFrame(first, sizeof(first)) == sizeof(first)); + interface.checkRecvFrame(received); + assert(notification_attempts == 0); // CCCD is not ready. + tx->callbacks->onSubscribe(tx, info, 1); + mtu = 23; interface.checkRecvFrame(received); + assert(notification_attempts == 0); // Never truncate to fit a smaller MTU. + mtu = 179; accept_notification = false; + interface.checkRecvFrame(received); + assert(notification_attempts == 1 && transmitted.empty() && interface.hasPendingIO()); + test_millis += interval - 1; + assert(interface.isWriteBusy()); + interface.checkRecvFrame(received); + assert(notification_attempts == 1); // Failed attempts must also be paced. + ++test_millis; + assert(!interface.isWriteBusy()); + accept_notification = true; interface.checkRecvFrame(received); + assert(notification_attempts == 2 && transmitted.size() == 1); + assert(transmitted[0] == std::vector(first, first + sizeof(first))); + assert(!interface.hasPendingIO()); // One accepted copy, no retry/late-status duplicate. + assert(interface.writeFrame(second, sizeof(second)) == sizeof(second)); + test_millis += interval - 1; interface.checkRecvFrame(received); + assert(notification_attempts == 2 && transmitted.size() == 1); + ++test_millis; interface.checkRecvFrame(received); + assert(notification_attempts == 3 && transmitted.size() == 2); + assert(transmitted[1] == std::vector(second, second + sizeof(second))); + // Real adapter arithmetic must remain correct when millis wraps at UINT32. + test_millis = UINT32_MAX - 7; + assert(interface.writeFrame(first, sizeof(first)) == sizeof(first)); + interface.checkRecvFrame(received); + assert(notification_attempts == 4); + assert(interface.writeFrame(second, sizeof(second)) == sizeof(second)); + test_millis += interval - 1; + assert(interface.isWriteBusy()); interface.checkRecvFrame(received); + assert(notification_attempts == 4); + ++test_millis; + assert(!interface.isWriteBusy()); interface.checkRecvFrame(received); + assert(notification_attempts == 5 && transmitted.size() == 4); + interface.checkRecvFrame(received); + assert(notification_attempts == 5 && !interface.hasPendingIO()); + // A faster lab pace must not shorten or remove the existing stall recovery. + assert(interface.writeFrame(first, sizeof(first)) == sizeof(first)); + accept_notification = false; test_millis += interval; + interface.checkRecvFrame(received); + test_millis += 9999; interface.checkRecvFrame(received); + assert(disconnects == 0 && interface.hasPendingIO()); + test_millis += interval; interface.checkRecvFrame(received); + assert(disconnects == 1 && !interface.isConnected()); + NimBLEDevice::deinit(true); +} + int main() { test_identity_modes(); test_pairing_wake_for_random_and_saved_pins(); test_startup_failures(); test_authentication_and_bond_lifecycle(); test_real_transport_frames(); + test_required_reply_backpressure_services_tx_before_rx(); + test_full_queue_best_effort_eviction_still_admits_rx(); + test_exact_notify_interval_failure_retention_and_rollover(); puts("NimBLE adapter: identity, faults, bonding, stealth and full frames PASS"); } diff --git a/test/test_ble_visibility_probe.py b/test/test_ble_visibility_probe.py new file mode 100644 index 000000000..eb3ad845b --- /dev/null +++ b/test/test_ble_visibility_probe.py @@ -0,0 +1,69 @@ +import importlib.util +from pathlib import Path +import unittest + +SPEC = importlib.util.spec_from_file_location("ble_visibility_probe", Path(__file__).resolve().parents[1] / "tools/hil/ble_visibility_probe.py") +probe = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(probe) +TARGET = "D2:90:05:46:31:01" + + +class VisibilityTests(unittest.TestCase): + def test_info_not_available_is_not_a_cached_device(self): + for code in (0, 1): + self.assertFalse(probe.is_cached_device(f"Device {TARGET} not available\n", TARGET, code)) + self.assertFalse(probe.is_cached_device(f"Device {TARGET}\n", TARGET, 1)) + self.assertFalse(probe.is_cached_device(f"Name: Device {TARGET}\n", TARGET, 0)) + + def test_successful_info_requires_exact_header(self): + for header in (f"Device {TARGET}", f"Device {TARGET} (random)", f"Device {TARGET} (public)"): + self.assertTrue(probe.is_cached_device(f"\x1b[0;92m{header}\x1b[0m\n\tName: test\n", TARGET, 0)) + + def test_cache_query_errors_are_not_verified_absence(self): + self.assertIsNone(probe.device_cache_state("org.bluez failed", TARGET, 1)) + self.assertIsNone(probe.device_cache_state(f"Device {TARGET}\n", TARGET, 1)) + self.assertIsNone(probe.device_cache_state(f"Device {TARGET}\nDevice {TARGET} not available\n", TARGET, 0)) + self.assertIs(probe.device_cache_state(f"Device {TARGET} not available\n", TARGET, 1), False) + + def test_new_cache_record_is_evidence_without_fabricated_events_or_rssi(self): + scan = probe.parse_discovery(f"[NEW] Device {TARGET} name\nDiscovery started\n", TARGET) + result = probe.with_cache_evidence(scan, False, True) + self.assertTrue(result["advertisement_evidence"]) + self.assertTrue(result["new_target_record_during_scan"]) + self.assertFalse(result["post_start_event_evidence"]) + self.assertEqual(result["target_events"], 0) + self.assertEqual(result["rssi_dbm"], []) + self.assertEqual(result["evidence_sources"], ["absent_before_present_after_successful_discovery"]) + + def test_existing_or_unknown_cache_and_failed_discovery_not_new_evidence(self): + scan = probe.parse_discovery("Discovery started\n", TARGET) + for before, after in ((True, True), (None, True), (False, None), (False, False)): + result = probe.with_cache_evidence(scan, before, after) + self.assertFalse(result["advertisement_evidence"]) + failed = probe.parse_discovery("Failed to start discovery\n", TARGET) + self.assertFalse(probe.with_cache_evidence(failed, False, True)["advertisement_evidence"]) + + def test_fresh_rssi_remains_separate_from_cache_transition(self): + scan = probe.parse_discovery(f"Discovery started\n[CHG] Device {TARGET} RSSI: -40\n", TARGET) + result = probe.with_cache_evidence(scan, True, True) + self.assertTrue(result["post_start_event_evidence"]) + self.assertFalse(result["new_target_record_during_scan"]) + self.assertEqual(result["evidence_sources"], ["post_start_rssi_update"]) + + def test_cached_enumeration_is_not_actual_discovery(self): + result = probe.parse_discovery(f"[NEW] Device {TARGET} cached\nDiscovery started\n", TARGET) + self.assertFalse(result["advertisement_evidence"]) + self.assertEqual(result["target_events"], 0) + + def test_filters_other_devices_and_parses_target_rssi(self): + result = probe.parse_discovery(f"Discovery started\n[CHG] Device AA:00:00:00:00:01 RSSI: -12\n\x1b[0;92m[CHG]\x1b[0m Device {TARGET} RSSI: -54\n[CHG] Device {TARGET} RSSI: 0xffffffcb (-53)\n", TARGET) + self.assertTrue(result["advertisement_evidence"]) + self.assertEqual(result["rssi_dbm"], [-54, -53]) + self.assertEqual(result["target_events"], 2) + + def test_scan_start_failure_does_not_use_stale_events(self): + self.assertFalse(probe.parse_discovery(f"Failed to start discovery\n[NEW] Device {TARGET} cached\n", TARGET)["advertisement_evidence"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_companion_ble_bench.py b/test/test_companion_ble_bench.py new file mode 100644 index 000000000..4e4f2ef2c --- /dev/null +++ b/test/test_companion_ble_bench.py @@ -0,0 +1,181 @@ +"""Actual lab echo/dispatch boundary; no BLE, serial hardware or radio.""" +from pathlib import Path +import re +import shutil +import subprocess +import tempfile +import unittest + +from test_replay_reset_integration import extract_braced + +ROOT = Path(__file__).resolve().parents[1] + + +class CompanionBleBenchTest(unittest.TestCase): + def compile(self, source, *flags, expect_failure=False): + compiler = shutil.which("c++") or shutil.which("g++") + if not compiler: + self.skipTest("a native C++ compiler is required") + with tempfile.TemporaryDirectory(prefix="mesh-ble-bench-") as folder: + cpp, binary = Path(folder) / "test.cpp", Path(folder) / "test" + cpp.write_text(source) + result = subprocess.run([compiler, "-std=c++11", "-Wall", "-Wextra", "-Werror", + "-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie", + "-I", str(ROOT / "src"), *flags, str(cpp), "-o", str(binary)], + capture_output=True, text=True, timeout=30) + if expect_failure: + self.assertNotEqual(result.returncode, 0) + return result.stderr + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + run = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10) + self.assertEqual(run.returncode, 0, run.stdout + run.stderr) + + def test_production_echo_exact_binary_maximum_and_malformed_bounds(self): + self.compile(r''' +#include +#include +#include +#include +using namespace mesh::companion; +int main() { + static_assert(kBleBenchCommand == 0x70 && kBleBenchHeaderBytes == 15, "wire contract"); + std::array request{}; + std::array output{}; + for (size_t total : {size_t(15), size_t(20), size_t(64), size_t(172), size_t(176)}) { + request[0] = kBleBenchCommand; memcpy(request.data() + 1, "BLE1", 4); + for (size_t i = 5; i < total; ++i) request[i] = uint8_t(i * 71); + const size_t payload = total - kBleBenchHeaderBytes; + request[13] = payload; request[14] = payload >> 8; + output.fill(0xa5); + assert(buildBleBenchEcho(request.data(), total, output.data() + 1, total) == total); + assert(!memcmp(request.data(), output.data() + 1, total)); + assert(output[0] == 0xa5 && output[total + 1] == 0xa5); + output.fill(0xa5); + assert(buildBleBenchEcho(request.data(), total, output.data(), total - 1) == 0); + for (uint8_t value : output) assert(value == 0xa5); + } + // Full binary byte range, including NUL/FEND/FESC/FF, is neither escaped nor + // interpreted. Two sequences cover all possible byte values in payloads. + for (unsigned part : {0U, 128U}) { + const size_t total = 143; + request[13] = 128; request[14] = 0; + for (size_t i = 0; i < 128; ++i) request[15 + i] = uint8_t(part + i); + assert(buildBleBenchEcho(request.data(), total, output.data(), output.size()) == total); + assert(!memcmp(request.data(), output.data(), total)); + } + for (size_t total = 0; total < kBleBenchHeaderBytes; ++total) + assert(buildBleBenchEcho(request.data(), total, output.data(), output.size()) == 0); + assert(buildBleBenchEcho(nullptr, 15, output.data(), output.size()) == 0); + assert(buildBleBenchEcho(request.data(), 15, nullptr, 176) == 0); + assert(buildBleBenchEcho(request.data(), 177, output.data(), output.size()) == 0); + request[13] = 0; request[14] = 0; + for (unsigned fault = 0; fault < 6; ++fault) { + auto bad = request; + if (fault == 0) bad[0] ^= 1; + if (fault == 1) bad[1] ^= 1; + if (fault == 2) bad[4] = '2'; + if (fault == 3) bad[13] = 1; + if (fault == 4) bad[14] = 1; + if (fault == 5) bad[13] = bad[14] = 255; + assert(buildBleBenchEcho(bad.data(), 15, output.data(), output.size()) == 0); + } +} +''', "-DMESH_HIL_BLE_BENCH=1") + + def test_real_dispatch_returns_once_before_rf_or_pending_operation_service(self): + source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text() + dispatch = extract_braced(source, + "if (cmd_frame[0] == mesh::companion::kBleBenchCommand)") + handler = extract_braced(source, "void MyMesh::handleCmdFrame(") + self.assertLess(handler.index(dispatch), handler.index("servicePendingSerialReply();")) + self.assertRegex(handler, r"#if defined\(MESH_HIL_BLE_BENCH\) && MESH_HIL_BLE_BENCH\s+" + r"if \(cmd_frame\[0\] == mesh::companion::kBleBenchCommand\)") + for forbidden in ("sendDirect", "sendPacket", "self_id", "savePrefs", "_radio", "new ", "malloc"): + self.assertNotIn(forbidden, dispatch) + self.compile(r''' +#include +#include +#include +static const uint8_t ERR_CODE_ILLEGAL_ARG = 6; +struct Serial { + bool busy = false; + unsigned calls = 0; + std::vector frame; + size_t writeFrame(const uint8_t* bytes, size_t count) { + ++calls; + if (busy) return 0; // No retry loop, queue growth or fallback broadcast. + frame.assign(bytes, bytes + count); return count; + } +}; +struct Handler { + uint8_t cmd_frame[MAX_FRAME_SIZE + 1]{}, out_frame[MAX_FRAME_SIZE + 1]{}; + Serial serial; Serial* _serial = &serial; + unsigned errors = 0, normal_path = 0; + void writeErrFrame(uint8_t code) { assert(code == 6); ++errors; } + void run(size_t len) { + @DISPATCH@ + ++normal_path; + } +}; +int main() { + Handler h; h.cmd_frame[0] = 0x70; memcpy(h.cmd_frame + 1, "BLE1", 4); + h.run(15); + assert(h.serial.calls == 1 && h.serial.frame.size() == 15 && h.normal_path == 0 && h.errors == 0); + h.serial.busy = true; h.run(15); + assert(h.serial.calls == 2 && h.normal_path == 0 && h.errors == 0); + h.run(177); + assert(h.serial.calls == 2 && h.errors == 1 && h.normal_path == 0); + h.cmd_frame[13] = 1; h.run(15); + assert(h.serial.calls == 2 && h.errors == 2 && h.normal_path == 0); + h.cmd_frame[0] = 0x05; h.run(15); + assert(h.normal_path == 1 && h.serial.calls == 2 && h.errors == 2); +} +'''.replace("@DISPATCH@", dispatch), "-DMESH_HIL_BLE_BENCH=1") + + def test_release_zero_effect_and_invalid_flag_rejected(self): + source = "#include \nint main() {}\n" + self.compile(source) + self.compile(source, "-DMESH_HIL_BLE_BENCH=0") + failure = self.compile(source, "-DMESH_HIL_BLE_BENCH=2", expect_failure=True) + self.assertIn("must be explicitly 0 or 1", failure) + # No echo symbols/constants exist at all in normal release compilation. + failure = self.compile(source.replace("int main() {}", "int main() { return mesh::companion::kBleBenchCommand; }"), + expect_failure=True) + self.assertIn("mesh", failure) + + def test_echo_uses_existing_required_response_route_and_queue_admission(self): + self.compile(r''' +#include +#include +#include +struct Frame { uint8_t buf[MAX_FRAME_SIZE]{}; size_t len = 0; }; +int main() { + const uint8_t echo[] = {0x70, 'B','L','E','1', 1,0,0,0, 2,0,0,0, 0,0}; + assert(mesh::companionFrameRequiresDelivery(echo, sizeof(echo))); + assert(mesh::companionFrameUsesRequesterRoute(echo, sizeof(echo))); + Frame queue[2]; size_t count = 0; + assert(mesh::enqueueCompanionFrame(queue, count, 2, echo, sizeof(echo))); + assert(mesh::enqueueCompanionFrame(queue, count, 2, echo, sizeof(echo))); + assert(count == 2); + assert(!mesh::enqueueCompanionFrame(queue, count, 2, echo, sizeof(echo))); + assert(count == 2); // Never grows or evicts another required response. + const uint8_t log[] = {0x88, 0}; + count = 0; + assert(mesh::enqueueCompanionFrame(queue, count, 2, log, sizeof(log))); + assert(mesh::enqueueCompanionFrame(queue, count, 2, echo, sizeof(echo))); + assert(mesh::enqueueCompanionFrame(queue, count, 2, echo, sizeof(echo))); + assert(count == 2 && queue[0].buf[0] == 0x70 && queue[1].buf[0] == 0x70); +} +''', "-DMESH_HIL_BLE_BENCH=1") + + def test_lab_opcode_does_not_collide_with_shipped_protocol_commands(self): + source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text() + for token in re.findall(r"^#define CMD_\w+\s+(0[xX][0-9a-fA-F]+|[0-9]+)", source, re.MULTILINE): + self.assertNotEqual(int(token, 0), 0x70) + for name in ("CompanionHardwareCommandCompat.h", "CompanionMotaControl.h"): + source = (ROOT / "src/helpers" / name).read_text() + self.assertNotRegex(source, r"=\s*(?:0x70|112)\s*;") + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_mota_serial_deadline.py b/test/test_mota_serial_deadline.py index d13987005..1f4c551eb 100644 --- a/test/test_mota_serial_deadline.py +++ b/test/test_mota_serial_deadline.py @@ -5,12 +5,81 @@ from pathlib import Path import subprocess import tempfile import unittest +from test_t096_full_memory import method ROOT = Path(__file__).resolve().parents[1] class MotaSerialDeadlineTests(unittest.TestCase): + def test_actual_nrf_ble_capacity_uses_current_ready_connection(self): + serial = (ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp").read_text() + getter = method(serial, "uint16_t SerialBLEInterface::motaResponseCapacity()") + harness = r''' +#include +#include +#include "helpers/BleMotaProtocol.h" +struct BLEConnection { + bool online = true; + uint16_t mtu = 247; + bool connected() const { return online; } + uint16_t getMtu() const { return mtu; } +}; +struct { + BLEConnection connection; + bool available = true; + uint16_t last_handle = 0; + BLEConnection* Connection(uint16_t handle) { + last_handle = handle; + return available ? &connection : nullptr; + } +} Bluefruit; +struct SerialBLEInterface { + bool ready = true; + uint16_t _conn_handle = 77; + struct { bool active = true; bool isActive() const { return active; } } _mota_stream; + bool isMotaChannelReady() const { return ready; } + uint16_t motaResponseCapacity(); +}; +@GETTER@ +int main() { + SerialBLEInterface port; + for (uint16_t mtu : {uint16_t(247), uint16_t(185), uint16_t(179), uint16_t(23)}) { + Bluefruit.connection.mtu = mtu; + const uint16_t expected = mtu == 247 ? 197 : mtu - 3; + assert(port.motaResponseCapacity() == expected); + assert(Bluefruit.last_handle == 77); + } + Bluefruit.connection.mtu = 22; + assert(port.motaResponseCapacity() == 0); + Bluefruit.connection.mtu = 247; + Bluefruit.connection.online = false; + assert(port.motaResponseCapacity() == 0); + Bluefruit.connection.online = true; + Bluefruit.available = false; + assert(port.motaResponseCapacity() == 0); + Bluefruit.available = true; + port.ready = false; + assert(port.motaResponseCapacity() == 0); + port.ready = true; + port._mota_stream.active = false; + assert(port.motaResponseCapacity() == 0); +} +'''.replace("@GETTER@", getter).replace("#include ", + "#include \n#include ") + with tempfile.TemporaryDirectory() as directory: + cpp = Path(directory) / "capacity.cpp" + binary = Path(directory) / "capacity" + cpp.write_text(harness) + built = subprocess.run([ + "g++", "-std=c++17", "-fsanitize=address,undefined", + "-fno-sanitize-recover=all", "-fno-pie", "-no-pie", + "-I", str(ROOT / "src"), str(cpp), "-o", str(binary)], + capture_output=True, text=True) + self.assertEqual(built.returncode, 0, built.stdout + built.stderr) + result = subprocess.run([str(binary)], capture_output=True, text=True) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + def test_actual_native_serial_cases(self): native = (ROOT / "test/test_ota/test_ota_core.cpp").read_text() start = native.index("class FakeMotaSeederStream") @@ -32,6 +101,98 @@ using namespace mesh::ota; ''' source += native[start:end] source += r''' +class MtuMotaSource : public SerialMotaSource { +public: + using SerialMotaSource::SerialMotaSource; + uint16_t frame_capacity = 197; + mutable unsigned checks = 0; + unsigned disconnect_after = 0; +protected: + uint16_t responseFrameCapacity() const override { + if (disconnect_after && ++checks >= disconnect_after) return 0; + return frame_capacity; + } +}; + +TEST(MotaSourceMtu, RawRequestsFitOneActualAttPayload) { + for (uint16_t mtu : {247, 185, 179, 23}) { + resetArduinoMock(); + FakeMotaSeederStream stream; + MtuMotaSource source(stream, MotaStreamWritePolicy::NoFlush); + source.frame_capacity = std::min(197, mtu - 3); + std::array bytes{}; + ASSERT_TRUE(source.read(0, 99, bytes.data(), bytes.size())); + const uint16_t limit = source.frame_capacity - 5; + EXPECT_EQ(stream.lengths.size(), (bytes.size() + limit - 1) / limit); + uint32_t offset = 99; + for (size_t i = 0; i < stream.lengths.size(); ++i) { + EXPECT_EQ(stream.offsets[i], offset); + EXPECT_LE(stream.lengths[i] + 5u, source.frame_capacity); + offset += stream.lengths[i]; + } + for (size_t i = 0; i < bytes.size(); ++i) EXPECT_EQ(bytes[i], uint8_t(99 + i)); + if (mtu == 247) EXPECT_EQ(stream.lengths.size(), 11u); + if (mtu == 185) EXPECT_EQ(stream.lengths.size(), 12u); + if (mtu == 179) EXPECT_EQ(stream.lengths.size(), 12u); + } +} + +TEST(MotaSourceMtu, DeflateRequestsIncludeSizeAndFrameOverheads) { + for (uint16_t mtu : {247, 185, 179, 23}) { + resetArduinoMock(); + FakeDeflateMotaSeederStream stream; + stream.encoded.resize(2048); + for (size_t i = 0; i < stream.encoded.size(); ++i) stream.encoded[i] = uint8_t(i * 17); + MtuMotaSource source(stream, MotaStreamWritePolicy::NoFlush); + source.frame_capacity = std::min(197, mtu - 3); + std::array bytes{}; + uint16_t length = 0; + ASSERT_TRUE(source.read_deflated_block(0, 7, bytes.data(), bytes.size(), &length)); + const uint16_t limit = source.frame_capacity - 7; + EXPECT_EQ(stream.lengths.size(), 1u + (bytes.size() + limit - 1) / limit); + EXPECT_EQ(stream.lengths[0], 0u); // bounded size query + uint16_t offset = 0; + for (size_t i = 1; i < stream.lengths.size(); ++i) { + EXPECT_EQ(stream.offsets[i], offset); + EXPECT_LE(stream.lengths[i] + 7u, source.frame_capacity); + offset += stream.lengths[i]; + } + EXPECT_EQ(length, bytes.size()); + EXPECT_EQ(0, memcmp(bytes.data(), stream.encoded.data(), bytes.size())); + } +} + +TEST(MotaSourceMtu, InvalidOrDisconnectedCapacityFailsBeforeNextRequest) { + for (uint16_t invalid : {0, 1, 5, 7}) { + FakeMotaSeederStream stream; + MtuMotaSource source(stream, MotaStreamWritePolicy::NoFlush); + source.frame_capacity = invalid; + uint8_t byte = 0; + EXPECT_FALSE(source.read(0, 0, &byte, 1)); + EXPECT_TRUE(stream.lengths.empty()); + uint16_t length = 99; + EXPECT_FALSE(source.read_deflated_block(0, 0, &byte, 1, &length)); + EXPECT_EQ(length, 0u); + EXPECT_TRUE(stream.lengths.empty()); + } + FakeMotaSeederStream stream; + MtuMotaSource source(stream, MotaStreamWritePolicy::NoFlush); + source.disconnect_after = 2; + std::array bytes{}; + EXPECT_FALSE(source.read(0, 0, bytes.data(), bytes.size())); + EXPECT_EQ(stream.lengths, (std::vector{192})); +} + +TEST(MotaSourceMtu, OversizedCapacityCannotIncreaseLegacyTransportMaximum) { + FakeMotaSeederStream stream; + MtuMotaSource source(stream, MotaStreamWritePolicy::NoFlush); + source.frame_capacity = UINT16_MAX; + std::array bytes{}; + ASSERT_TRUE(source.read(0, 0, bytes.data(), bytes.size())); + EXPECT_EQ(stream.lengths, (std::vector{192, 192, 192, 192, 192, 64})); +} +''' + source += r''' int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); diff --git a/test/test_nimble_companion.py b/test/test_nimble_companion.py index 8b965ce6a..8be66579f 100644 --- a/test/test_nimble_companion.py +++ b/test/test_nimble_companion.py @@ -10,11 +10,37 @@ ROOT = Path(__file__).resolve().parents[1] class NimbleCompanionTests(unittest.TestCase): def test_identity_security_and_data_path_under_sanitizers(self): + self.run_adapter() + + def test_lab15_interval_real_adapter_under_sanitizers(self): + self.run_adapter("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=15") + + def test_fast_interval_is_lab_only_and_bounded(self): + invalid = ( + ("-DMESH_HIL_BLE_INTERVAL_MS=15",), + ("-DMESH_HIL_BLE_BENCH=0", "-DMESH_HIL_BLE_INTERVAL_MS=15"), + ("-DMESH_HIL_BLE_BENCH=2", "-DMESH_HIL_BLE_INTERVAL_MS=15"), + ("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=6"), + ("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=61"), + ) + for flags in invalid: + with self.subTest(flags=flags): + result = subprocess.run([ + "c++", "-std=c++17", "-fsyntax-only", "-DMESH_USE_NIMBLE_ARDUINO=1", *flags, + "-I", str(ROOT / "test/fixtures/nimble_companion/mocks"), + "-I", str(ROOT / "src"), + str(ROOT / "src/helpers/esp32/SerialBLEInterface.cpp"), + ], text=True, capture_output=True, timeout=30) + self.assertNotEqual(result.returncode, 0) + self.assertIn("MESH_HIL_BLE_INTERVAL_MS", result.stderr) + + def run_adapter(self, *flags): with tempfile.TemporaryDirectory(prefix="meshcore-nimble-") as temp: binary = Path(temp) / "nimble-companion" result = subprocess.run([ "c++", "-std=c++17", "-g", "-fsanitize=address,undefined", "-DMESH_USE_NIMBLE_ARDUINO=1", + *flags, "-I", str(ROOT / "test/fixtures/nimble_companion/mocks"), "-I", str(ROOT / "src"), str(ROOT / "src/helpers/esp32/SerialBLEInterface.cpp"), diff --git a/test/test_nrf52_ble_backpressure.py b/test/test_nrf52_ble_backpressure.py new file mode 100644 index 000000000..f47f5c801 --- /dev/null +++ b/test/test_nrf52_ble_backpressure.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +"""Execute production nRF52 BLE queue/dispatch methods with simulated UART IO. + +This covers real admission, RX-dequeue, retry-throttle and recovery ordering, +not Bluefruit/SoftDevice callbacks or physical ATT delivery. +""" +from pathlib import Path +import os +import subprocess +import tempfile +import unittest + +from test_t096_full_memory import method + +ROOT = Path(__file__).resolve().parents[1] +SOURCE = ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp" + +HARNESS = r''' +#include +#include +#include +#include +#include +#include + +#define MAX_FRAME_SIZE 176 +#define FRAME_QUEUE_SIZE 12 +#define BLE_RETRY_THROTTLE_MS 250 +#define BLE_DEBUG_PRINTLN(...) do {} while (0) +#define meshBleTrace(...) do {} while (0) +static uint32_t current_millis = 100; +uint32_t millis() { return current_millis; } + +struct SerialBLEInterface { + struct Frame { uint8_t len = 0; uint8_t buf[MAX_FRAME_SIZE] = {}; }; + uint8_t send_queue_len = 0, recv_queue_len = 0; + Frame send_queue[FRAME_QUEUE_SIZE], recv_queue[FRAME_QUEUE_SIZE]; + unsigned long _last_retry_attempt = 0; + mesh::BleTxStallWatchdog _tx_stall_watchdog; + mesh::BleDisconnectRecovery _tx_disconnect_recovery; + bool connected = true, allow_write = false, partial_write = false; + unsigned write_attempts = 0, recovery_services = 0, recoveries = 0; + std::vector> transmitted; + bool isConnected() const { return connected && !_tx_disconnect_recovery.pending(); } + void serviceBondedOnlyTransition() {} + void serviceTxRecovery(uint32_t) { ++recovery_services; } + void recoverStalledTx(const char*) { ++recoveries; _tx_disconnect_recovery.begin(); } + size_t writeBleUartFrame(const Frame& frame) { + ++write_attempts; + if (partial_write) return 1; + if (!allow_write) return 0; + transmitted.emplace_back(frame.buf, frame.buf + frame.len); + return frame.len; + } + void receive(const uint8_t* bytes, size_t size) { + assert(recv_queue_len < FRAME_QUEUE_SIZE && size <= MAX_FRAME_SIZE); + recv_queue[recv_queue_len].len = size; + memcpy(recv_queue[recv_queue_len++].buf, bytes, size); + } + void shiftSendQueueLeft(); + void shiftRecvQueueLeft(); + bool hasRequiredReplyCapacity() const; + bool isWriteBusy() const; + size_t writeFrame(const uint8_t*, size_t); + size_t checkRecvFrame(uint8_t*); +}; + +@METHODS@ + +static void full_required_queue() { + current_millis = 100; + SerialBLEInterface link; + for (uint8_t i = 0; i < FRAME_QUEUE_SIZE; ++i) { + const uint8_t reply[] = {0x0d, i}; + assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + } + const uint8_t command[] = {66, 'A', '7', '|', 'v', 'e', 'r'}; + link.receive(command, sizeof(command)); + uint8_t received[MAX_FRAME_SIZE] = {}; + assert(link.checkRecvFrame(received) == 0); + assert(link.recv_queue_len == 1 && link.send_queue_len == FRAME_QUEUE_SIZE); + assert(link.write_attempts == 1 && link.transmitted.empty()); + ++current_millis; + assert(link.checkRecvFrame(received) == 0); + assert(link.write_attempts == 1); // Existing 250 ms retry throttle is retained. + assert(link.recv_queue_len == 1 && link.isWriteBusy()); + + current_millis = 351; + link.allow_write = true; + // A completed TX frees admission on this SAME call, before RX is examined. + assert(link.checkRecvFrame(received) == sizeof(command)); + assert(memcmp(received, command, sizeof(command)) == 0); + assert(link.recv_queue_len == 0 && link.send_queue_len == FRAME_QUEUE_SIZE - 1); + const uint8_t reply[] = {29, 'A', '7', '|', 'O', 'K'}; + assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + for (unsigned i = 0; i < FRAME_QUEUE_SIZE + 2; ++i) { + ++current_millis; + assert(link.checkRecvFrame(received) == 0); + } + assert(link.transmitted.size() == FRAME_QUEUE_SIZE + 1); + for (uint8_t i = 0; i < FRAME_QUEUE_SIZE; ++i) + assert((link.transmitted[i] == std::vector{0x0d, i})); + assert(link.transmitted.back() == std::vector(reply, reply + sizeof(reply))); + assert(link.send_queue_len == 0 && link.recv_queue_len == 0); +} + +static void best_effort_eviction() { + current_millis = 100; + SerialBLEInterface link; + for (uint8_t i = 0; i < FRAME_QUEUE_SIZE - 1; ++i) { + const uint8_t log[] = {0x88, i}; + assert(link.writeFrame(log, sizeof(log)) == sizeof(log)); + } + const uint8_t log[] = {0x88, 99}; + assert(link.writeFrame(log, sizeof(log)) == 0); // One required slot stays reserved. + const uint8_t previous[] = {0x0d, 99}; + assert(link.writeFrame(previous, sizeof(previous)) == sizeof(previous)); + const uint8_t command[] = {66, 'B', '8', '|', 'v', 'e', 'r'}; + link.receive(command, sizeof(command)); + uint8_t received[MAX_FRAME_SIZE] = {}; + assert(link.isWriteBusy()); // Preserve the nRF52 >=8/12 producer high-water mark. + assert(link.checkRecvFrame(received) == sizeof(command)); + assert(memcmp(received, command, sizeof(command)) == 0); + const uint8_t reply[] = {29, 'B', '8', '|', 'O', 'K'}; + assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + current_millis = 351; + link.allow_write = true; + for (unsigned i = 0; i < FRAME_QUEUE_SIZE + 2; ++i) { + assert(link.checkRecvFrame(received) == 0); + ++current_millis; + } + assert(link.transmitted.size() == FRAME_QUEUE_SIZE); + assert(link.transmitted[0] == std::vector(previous, previous + sizeof(previous))); + assert(link.transmitted[1] == std::vector(reply, reply + sizeof(reply))); + for (uint8_t i = 0; i < FRAME_QUEUE_SIZE - 2; ++i) + assert((link.transmitted[i + 2] == std::vector{0x88, i})); +} + +static void producer_threshold_and_recovery_ordering() { + current_millis = 100; + SerialBLEInterface link; + const uint8_t reply[] = {0x0d, 1}; + for (unsigned i = 0; i < 7; ++i) assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + assert(!link.isWriteBusy()); + assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + assert(link.isWriteBusy()); + const uint8_t command[] = {66, 'v'}; + link.receive(command, sizeof(command)); + link._tx_disconnect_recovery.begin(); + uint8_t received[MAX_FRAME_SIZE] = {}; + assert(link.checkRecvFrame(received) == 0); + assert(link.recovery_services == 1 && link.write_attempts == 0 && link.recv_queue_len == 1); + + SerialBLEInterface partial; + partial.partial_write = true; + assert(partial.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + partial.receive(command, sizeof(command)); + assert(partial.checkRecvFrame(received) == 0); + assert(partial.recoveries == 1 && partial.write_attempts == 1); + assert(partial.recv_queue_len == 1); // Recovery is invoked instead of dispatch. +} + +int main() { + full_required_queue(); + best_effort_eviction(); + producer_threshold_and_recovery_ordering(); +} +''' + + +class Nrf52BleBackpressureTests(unittest.TestCase): + def test_real_queue_dispatch_throttle_and_recovery_under_sanitizers(self): + source = SOURCE.read_text() + signatures = ["void SerialBLEInterface::shiftSendQueueLeft()", + "void SerialBLEInterface::shiftRecvQueueLeft()", + "size_t SerialBLEInterface::writeFrame(", + "size_t SerialBLEInterface::checkRecvFrame(", + "bool SerialBLEInterface::isWriteBusy() const"] + capacity = "bool SerialBLEInterface::hasRequiredReplyCapacity() const" + if capacity in source: + signatures.append(capacity) + extracted = "\n".join(method(source, signature) for signature in signatures) + sanitizers = [] if os.name == "nt" else ["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"] + with tempfile.TemporaryDirectory(prefix="meshcore-nrf-ble-backpressure-") as temp: + binary = Path(temp) / "backpressure" + built = subprocess.run(["c++", "-std=c++17", "-x", "c++", "-", "-I" + str(ROOT / "src"), + *sanitizers, "-o", str(binary)], input=HARNESS.replace("@METHODS@", extracted), + capture_output=True, text=True, timeout=30) + self.assertEqual(built.returncode, 0, built.stderr) + result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_queue_capacity_remains_twelve(self): + header = (ROOT / "src/helpers/nrf52/SerialBLEInterface.h").read_text() + self.assertRegex(header, r"#define\s+FRAME_QUEUE_SIZE\s+12\b") + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/hil/ble_visibility_probe.py b/tools/hil/ble_visibility_probe.py new file mode 100644 index 000000000..2017b17da --- /dev/null +++ b/tools/hil/ble_visibility_probe.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +"""Owned, bounded BlueZ discovery; print only one explicitly selected peer. + +Does not pair/connect, power adapters, open USB, or stop shared discovery. +All nearby-device output stays in process memory and is discarded. +""" +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path +import re +import subprocess + + +ANSI = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") +MAC = re.compile(r"^(?:[0-9A-F]{2}:){5}[0-9A-F]{2}$") + + +def device_cache_state(output: str, target: str, returncode: int) -> bool | None: + """True=exact successful info, False=explicit unavailable, None=query error.""" + header = re.compile(r"^Device\s+" + re.escape(target) + r"(?:\s+\((?:public|random)\))?\s*$", re.I) + unavailable = re.compile(r"^Device\s+" + re.escape(target) + r"\s+not available\s*$", re.I) + lines = ANSI.sub("", output).splitlines() + headers = sum(bool(header.fullmatch(line)) for line in lines) + absent = sum(bool(unavailable.fullmatch(line)) for line in lines) + if returncode == 0 and headers == 1 and not absent: + return True + if returncode in (0, 1) and absent == 1 and not headers: + return False + return None + + +def is_cached_device(output: str, target: str, returncode: int) -> bool: + """Only a successful info header proves a cached record, never its error.""" + return device_cache_state(output, target, returncode) is True + + +def parse_discovery(output: str, target: str) -> dict: + started = False + events = 0 + new_events = 0 + rssi = [] + for line in ANSI.sub("", output).splitlines(): + if "Discovery started" in line: + started = True + if not started or not re.search(r"\bDevice " + re.escape(target) + r"(?:\s|$)", line, re.I): + continue + if "[NEW]" in line or "[CHG]" in line: + events += 1 + new_events += int("[NEW]" in line) + value = re.search(r"RSSI:\s*(?:0x[0-9a-f]+\s*\()?(-\d+)\)?", line, re.I) + if value: + rssi.append(int(value[1])) + return { + "discovery_started": started, + "target_events": events, + "new_target_events": new_events, + "rssi_updates": len(rssi), + "rssi_dbm": rssi, + # Startup cached-device enumeration is ignored until StartDiscovery succeeds. + "advertisement_evidence": bool(rssi or new_events), + } + + +def with_cache_evidence(scan: dict, before: bool | None, after: bool | None) -> dict: + """Do not invent event counts when BlueZ prints its start marker late. + + A verified absent -> present transition during successful discovery is + separate evidence of a newly learned target. It does not prove the exact + advertising time/RSSI, nor attribution if another discovery client exists. + Unknown/failed info queries never constitute verified prior absence. + """ + result = dict(scan) + created = bool(scan["discovery_started"] and before is False and after is True) + event_evidence = bool(scan["advertisement_evidence"]) + sources = [] + if scan["rssi_updates"]: + sources.append("post_start_rssi_update") + if scan["new_target_events"]: + sources.append("post_start_new_target_event") + if created: + sources.append("absent_before_present_after_successful_discovery") + result.update( + cached_before=before is True, + cached_after=after is True, + cache_query_before_verified=before is not None, + cache_query_after_verified=after is not None, + post_start_event_evidence=event_evidence, + new_target_record_during_scan=created, + advertisement_evidence=event_evidence or created, + evidence_sources=sources, + ) + return result + + +def usb_identity(alias: str, expected_serial: str) -> dict: + path = Path(alias) + if path.parent != Path("/dev/serial/by-id") or expected_serial not in path.name: + raise ValueError("Explicit matching USB by-id identity required") + resolved = path.resolve(strict=True) + if not re.fullmatch(r"ttyACM\d+", resolved.name): + raise ValueError("Expected native USB ACM radio") + device = Path("/sys/class/tty") / resolved.name / "device" + for parent in (device.resolve(), *device.resolve().parents): + if (parent / "serial").is_file(): + serial = (parent / "serial").read_text().strip() + vid = (parent / "idVendor").read_text().strip().lower() + pid = (parent / "idProduct").read_text().strip().lower() + if (serial, vid, pid) != (expected_serial, "239a", "8029"): + raise ValueError("USB identity mismatch") + return {"serial": serial, "vid": vid, "pid": pid, "port": str(resolved)} + raise ValueError("USB identity unavailable") + + +def bluetooth(*args: str, timeout: int = 8) -> subprocess.CompletedProcess: + return subprocess.run( + ["bluetoothctl", *args], stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, text=True, timeout=timeout, + env={**os.environ, "LC_ALL": "C"}, check=False, + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--target", required=True) + parser.add_argument("--controller", required=True) + parser.add_argument("--usb-by-id", required=True) + parser.add_argument("--usb-serial", required=True) + parser.add_argument("--seconds", type=int, default=20, choices=range(1, 21)) + args = parser.parse_args() + target, controller = args.target.upper(), args.controller.upper() + result = {"target": target, "duration_seconds": args.seconds} + try: + if not MAC.fullmatch(target) or not MAC.fullmatch(controller): + raise ValueError("Exact peer and controller MAC required") + result["usb_before"] = usb_identity(args.usb_by_id, args.usb_serial) + adapter = bluetooth("show", controller) + if not re.search(r"Powered:\s*yes\b", adapter.stdout): + raise ValueError("Expected controller is not already powered") + # Verify this is the selected/default controller, rather than changing it. + default = bluetooth("show") + if not re.search(r"Controller\s+" + re.escape(controller) + r"\b", default.stdout, re.I): + raise ValueError("Expected controller is not the default") + before = bluetooth("info", target) + before_state = device_cache_state(before.stdout, target, before.returncode) + scan = bluetooth("--timeout", str(args.seconds), "scan", "on", timeout=args.seconds + 5) + result["scan_exit_code"] = scan.returncode + after = bluetooth("info", target) + after_state = device_cache_state(after.stdout, target, after.returncode) + result.update(with_cache_evidence(parse_discovery(scan.stdout, target), before_state, after_state)) + result["usb_after"] = usb_identity(args.usb_by_id, args.usb_serial) + result["success"] = result["discovery_started"] + except (ValueError, OSError, subprocess.SubprocessError): + result["success"] = False + result["error"] = "Identity-bound discovery could not complete; raw output suppressed" + print(json.dumps(result, sort_keys=True)) + return 0 if result["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main())