mirror of
https://github.com/mikecarper/MeshCore.git
synced 2026-10-20 19:41:59 +00:00
127 lines
4.8 KiB
Python
127 lines
4.8 KiB
Python
"""Exercise diagnostic dispatch bracketing and the exported event decoder."""
|
|
from importlib.util import module_from_spec, spec_from_file_location
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def load(name, path):
|
|
spec = spec_from_file_location(name, ROOT / path)
|
|
module = module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
patcher = load("ble_patcher", "scripts/nrf52_ble_cccd_fix.py")
|
|
capture = load("ble_capture", "scripts/capture_nrf52_ble_debug.py")
|
|
|
|
|
|
class BleDebugToolsTest(unittest.TestCase):
|
|
def test_dispatch_brackets_real_callback_and_measures_duration(self):
|
|
source = '''#include "bluefruit.h"
|
|
void AdafruitBluefruit::_ble_handler(ble_evt_t* evt)
|
|
{
|
|
if (_event_cb) _event_cb(evt);
|
|
}
|
|
'''
|
|
patched = patcher.patched_trace_source(source)
|
|
self.assertEqual(patcher.patched_trace_source(patched), patched)
|
|
with self.assertRaises(RuntimeError):
|
|
patcher.patched_trace_source("changed SDK dispatcher")
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
work = Path(directory)
|
|
(work / "Arduino.h").write_text(
|
|
"#include <stdint.h>\n#include <stddef.h>\n"
|
|
"extern uint32_t fake_us;\n"
|
|
"inline uint32_t millis(){return 1234;}\n"
|
|
"inline uint32_t micros(){return fake_us;}\n")
|
|
(work / "bluefruit.h").write_text('''
|
|
#pragma once
|
|
#include <stdint.h>
|
|
struct ble_evt_t {
|
|
struct {uint16_t evt_id;} header;
|
|
struct {struct {uint16_t conn_handle;} common_evt;} evt;
|
|
};
|
|
class AdafruitBluefruit {
|
|
public:
|
|
void (*_event_cb)(ble_evt_t*) = nullptr;
|
|
void _ble_handler(ble_evt_t*);
|
|
};
|
|
''')
|
|
(work / "dispatch.cpp").write_text(patched + '''
|
|
#include <cassert>
|
|
#include <helpers/nrf52/BleDebugTrace.h>
|
|
uint32_t fake_us = 100;
|
|
#if MESH_NRF52_BLE_TRACE
|
|
extern "C" { MeshBleTraceBuffer mesh_ble_trace; }
|
|
void meshBleTraceState(char*,size_t) {}
|
|
#endif
|
|
bool callback_ran = false;
|
|
void callback(ble_evt_t* evt) {
|
|
callback_ran = true;
|
|
#if MESH_NRF52_BLE_TRACE
|
|
assert(mesh_ble_trace.dispatch_event.load() == evt->header.evt_id);
|
|
meshBleTrace(0xf003, evt->evt.common_evt.conn_handle);
|
|
#endif
|
|
fake_us += 50;
|
|
}
|
|
int main() {
|
|
AdafruitBluefruit bluefruit;
|
|
bluefruit._event_cb = callback;
|
|
ble_evt_t evt = {{0x10},{{7}}};
|
|
bluefruit._ble_handler(&evt);
|
|
assert(callback_ran);
|
|
#if MESH_NRF52_BLE_TRACE
|
|
MeshBleTraceRecord r;
|
|
assert(meshBleTraceRead(1,r) && r.event == 0xe010);
|
|
assert(meshBleTraceRead(2,r) && r.event == 0xf003);
|
|
assert(meshBleTraceRead(3,r) && r.event == 0xe110 && r.detail == 50);
|
|
assert(mesh_ble_trace.dispatch_event.load() == 0);
|
|
assert(mesh_ble_trace.max_dispatch_us.load() == 50);
|
|
#endif
|
|
}
|
|
''')
|
|
for enabled in (0, 1):
|
|
binary = work / ("dispatch-" + str(enabled))
|
|
subprocess.run(["c++", "-std=c++17",
|
|
"-DMESH_NRF52_BLE_TRACE=" + str(enabled), "-I" + str(work),
|
|
"-I" + str(ROOT / "src"), str(work / "dispatch.cpp"),
|
|
"-o", str(binary)], check=True, capture_output=True)
|
|
subprocess.run([str(binary)], check=True, capture_output=True)
|
|
|
|
def test_reasons_security_subscription_and_timing_decode(self):
|
|
local = capture.decode_record(1, 10, 0xF002, 0, 22, 0)
|
|
self.assertEqual(local["reason"], "local_host_terminated")
|
|
timeout = capture.decode_record(2, 20, 0x11, 0, 8, 0)
|
|
self.assertEqual(timeout["reason"], "connection_timeout")
|
|
connected = capture.decode_record(3, 30, 0x10, 0, 12 | (24 << 16), 4 | (200 << 16))
|
|
self.assertEqual(connected["supervision_timeout_ms"], 2000)
|
|
self.assertEqual(connected["min_interval_ms"], 15)
|
|
secure = capture.decode_record(4, 40, 0x19, 0, 0, 0x302)
|
|
self.assertTrue(secure["bonded"] and secure["secure_connections"])
|
|
self.assertEqual(secure["error_source"], 2)
|
|
cccd = capture.decode_record(5, 50, 0x50, 0, 7 | (2 << 16), 1 | 0x100 | (1 << 16))
|
|
self.assertTrue(cccd["is_subscription"])
|
|
self.assertEqual(cccd["subscription_bits"], 1)
|
|
self.assertNotIn("payload", cccd)
|
|
self.assertEqual(capture.values("> next=9,capacity=256,dropped=0,format=2")["format"], 2)
|
|
|
|
def test_terminal_ignores_delayed_banner_before_command_echo(self):
|
|
class Port:
|
|
in_waiting = 0
|
|
def __init__(self):
|
|
self.chunks = iter([b"\r\nWELCOME private-banner\r\n> ",
|
|
b"ver\r\nCompanion debug\r\n> "])
|
|
def read(self, _):
|
|
return next(self.chunks)
|
|
reply = capture.Terminal(Port()).read_prompt(expected="ver")
|
|
self.assertNotIn("private-banner", reply)
|
|
self.assertIn("Companion debug", reply)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|