Ryan Malloy
6398a5223a
UART-controlled ESP32 peripheral for automated E2E Bluetooth testing. Dual-mode (Classic BT + BLE) via Bluedroid on original ESP32. Firmware (ESP-IDF v5.x, 2511 lines C): - NDJSON protocol over UART1 (115200 baud) - System commands: ping, reset, get_info, get_status - Classic BT: GAP, SPP, all 4 SSP pairing modes - BLE: GATTS, advertising, GATT service/characteristic management - 6 device personas: headset, speaker, keyboard, sensor, phone, bare - Event reporter: thread-safe async event queue to host Python MCP server (FastMCP, 1626 lines): - Async serial client with command/response correlation - Event queue with wait_for pattern matching - Tools: connection, configure, classic, ble, persona, events - MCP resources: esp32://status, esp32://events, esp32://personas Tests: 74 unit tests passing, 5 integration test stubs (skip without hardware)
286 lines
8.9 KiB
Python
286 lines
8.9 KiB
Python
"""Unit tests for mcbluetooth_esp32.event_queue."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from mcbluetooth_esp32.event_queue import EventQueue
|
|
from mcbluetooth_esp32.protocol import Event, MsgType
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_event(name: str, data: dict | None = None, ts: int | None = None) -> Event:
|
|
return Event(type=MsgType.EVENT, event=name, data=data or {}, ts=ts)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: push / get_events basics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPushAndGetEvents:
|
|
"""push() and get_events() without any async waiting."""
|
|
|
|
async def test_push_and_retrieve(self):
|
|
q = EventQueue()
|
|
q.push(_make_event("boot"))
|
|
q.push(_make_event("connect"))
|
|
|
|
events = q.get_events()
|
|
assert len(events) == 2
|
|
assert events[0].event == "boot"
|
|
assert events[1].event == "connect"
|
|
|
|
async def test_get_events_filtered_by_name(self):
|
|
q = EventQueue()
|
|
q.push(_make_event("boot"))
|
|
q.push(_make_event("connect"))
|
|
q.push(_make_event("boot"))
|
|
|
|
events = q.get_events(event_name="boot")
|
|
assert len(events) == 2
|
|
assert all(e.event == "boot" for e in events)
|
|
|
|
async def test_get_events_with_limit(self):
|
|
q = EventQueue()
|
|
for i in range(10):
|
|
q.push(_make_event("tick", data={"i": i}))
|
|
|
|
events = q.get_events(limit=3)
|
|
assert len(events) == 3
|
|
# Should be the *most recent* three
|
|
assert events[0].data["i"] == 7
|
|
assert events[1].data["i"] == 8
|
|
assert events[2].data["i"] == 9
|
|
|
|
async def test_get_events_with_since_ts(self):
|
|
q = EventQueue()
|
|
q.push(_make_event("a", ts=100))
|
|
q.push(_make_event("b", ts=200))
|
|
q.push(_make_event("c", ts=300))
|
|
|
|
events = q.get_events(since_ts=200)
|
|
assert len(events) == 2
|
|
assert events[0].event == "b"
|
|
assert events[1].event == "c"
|
|
|
|
async def test_get_events_with_none_ts_excluded_by_since_ts(self):
|
|
q = EventQueue()
|
|
q.push(_make_event("a", ts=None))
|
|
q.push(_make_event("b", ts=500))
|
|
|
|
events = q.get_events(since_ts=100)
|
|
assert len(events) == 1
|
|
assert events[0].event == "b"
|
|
|
|
async def test_get_events_empty_queue(self):
|
|
q = EventQueue()
|
|
assert q.get_events() == []
|
|
|
|
async def test_len(self):
|
|
q = EventQueue()
|
|
assert len(q) == 0
|
|
q.push(_make_event("a"))
|
|
assert len(q) == 1
|
|
q.push(_make_event("b"))
|
|
assert len(q) == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: wait_for
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestWaitFor:
|
|
"""Async waiting for specific events."""
|
|
|
|
async def test_wait_for_resolves_on_push(self):
|
|
q = EventQueue()
|
|
|
|
async def _push_later():
|
|
await asyncio.sleep(0.05)
|
|
q.push(_make_event("pair_complete", data={"addr": "AA:BB"}))
|
|
|
|
task = asyncio.create_task(_push_later())
|
|
event = await q.wait_for(event_name="pair_complete", timeout=2.0)
|
|
await task
|
|
|
|
assert event.event == "pair_complete"
|
|
assert event.data["addr"] == "AA:BB"
|
|
|
|
async def test_wait_for_timeout(self):
|
|
q = EventQueue()
|
|
with pytest.raises(asyncio.TimeoutError):
|
|
await q.wait_for(event_name="never_arrives", timeout=0.1)
|
|
|
|
async def test_wait_for_finds_existing_event(self):
|
|
"""If a matching event is already in the history, return immediately."""
|
|
q = EventQueue()
|
|
q.push(_make_event("boot", data={"v": "1.0"}))
|
|
|
|
# Should return instantly without waiting
|
|
event = await q.wait_for(event_name="boot", timeout=0.5)
|
|
assert event.event == "boot"
|
|
assert event.data["v"] == "1.0"
|
|
|
|
async def test_wait_for_returns_most_recent_existing(self):
|
|
"""When scanning history, the most recent match is returned."""
|
|
q = EventQueue()
|
|
q.push(_make_event("boot", data={"v": "old"}))
|
|
q.push(_make_event("boot", data={"v": "new"}))
|
|
|
|
event = await q.wait_for(event_name="boot", timeout=0.5)
|
|
assert event.data["v"] == "new"
|
|
|
|
async def test_wait_for_with_custom_match(self):
|
|
q = EventQueue()
|
|
|
|
async def _push_later():
|
|
await asyncio.sleep(0.05)
|
|
q.push(_make_event("pair_complete", data={"success": False}))
|
|
q.push(_make_event("pair_complete", data={"success": True}))
|
|
|
|
task = asyncio.create_task(_push_later())
|
|
event = await q.wait_for(
|
|
event_name="pair_complete",
|
|
match=lambda e: e.data.get("success") is True,
|
|
timeout=2.0,
|
|
)
|
|
await task
|
|
|
|
assert event.data["success"] is True
|
|
|
|
async def test_wait_for_requires_at_least_one_filter(self):
|
|
q = EventQueue()
|
|
with pytest.raises(ValueError, match="at least one"):
|
|
await q.wait_for(timeout=1.0)
|
|
|
|
async def test_wait_for_with_match_only(self):
|
|
"""Using only a custom match function (no event_name)."""
|
|
q = EventQueue()
|
|
|
|
async def _push():
|
|
await asyncio.sleep(0.05)
|
|
q.push(_make_event("x", data={"val": 42}))
|
|
|
|
task = asyncio.create_task(_push())
|
|
event = await q.wait_for(
|
|
match=lambda e: e.data.get("val") == 42,
|
|
timeout=2.0,
|
|
)
|
|
await task
|
|
assert event.data["val"] == 42
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: clear
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestClear:
|
|
async def test_clear_removes_all_events(self):
|
|
q = EventQueue()
|
|
q.push(_make_event("a"))
|
|
q.push(_make_event("b"))
|
|
assert len(q) == 2
|
|
|
|
q.clear()
|
|
assert len(q) == 0
|
|
assert q.get_events() == []
|
|
|
|
async def test_clear_does_not_cancel_waiters(self):
|
|
"""Waiters registered before clear() should still fire on new events."""
|
|
q = EventQueue()
|
|
|
|
async def _push_after_clear():
|
|
await asyncio.sleep(0.05)
|
|
q.clear()
|
|
await asyncio.sleep(0.05)
|
|
q.push(_make_event("target"))
|
|
|
|
task = asyncio.create_task(_push_after_clear())
|
|
event = await q.wait_for(event_name="target", timeout=2.0)
|
|
await task
|
|
assert event.event == "target"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: max_events boundary
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMaxEvents:
|
|
async def test_max_events_trims_oldest(self):
|
|
q = EventQueue(max_events=10)
|
|
for i in range(15):
|
|
q.push(_make_event("tick", data={"i": i}))
|
|
|
|
assert len(q) == 10
|
|
events = q.get_events()
|
|
# Oldest five (0..4) should have been trimmed
|
|
assert events[0].data["i"] == 5
|
|
assert events[-1].data["i"] == 14
|
|
|
|
async def test_max_events_exact_boundary(self):
|
|
q = EventQueue(max_events=1000)
|
|
for i in range(1001):
|
|
q.push(_make_event("x", data={"i": i}))
|
|
|
|
assert len(q) == 1000
|
|
events = q.get_events(limit=1000)
|
|
assert events[0].data["i"] == 1
|
|
assert events[-1].data["i"] == 1000
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: multiple waiters
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMultipleWaiters:
|
|
async def test_different_event_waiters(self):
|
|
"""Two waiters for different events should each get their own match."""
|
|
q = EventQueue()
|
|
|
|
async def _push_events():
|
|
await asyncio.sleep(0.05)
|
|
q.push(_make_event("alpha", data={"n": 1}))
|
|
q.push(_make_event("beta", data={"n": 2}))
|
|
|
|
task = asyncio.create_task(_push_events())
|
|
|
|
evt_a, evt_b = await asyncio.gather(
|
|
q.wait_for(event_name="alpha", timeout=2.0),
|
|
q.wait_for(event_name="beta", timeout=2.0),
|
|
)
|
|
await task
|
|
|
|
assert evt_a.event == "alpha"
|
|
assert evt_a.data["n"] == 1
|
|
assert evt_b.event == "beta"
|
|
assert evt_b.data["n"] == 2
|
|
|
|
async def test_same_event_wakes_all_matching_waiters(self):
|
|
"""If multiple waiters match the same event, they all get resolved."""
|
|
q = EventQueue()
|
|
|
|
async def _push():
|
|
await asyncio.sleep(0.05)
|
|
q.push(_make_event("boom"))
|
|
|
|
task = asyncio.create_task(_push())
|
|
|
|
e1, e2 = await asyncio.gather(
|
|
q.wait_for(event_name="boom", timeout=2.0),
|
|
q.wait_for(event_name="boom", timeout=2.0),
|
|
)
|
|
await task
|
|
|
|
assert e1.event == "boom"
|
|
assert e2.event == "boom"
|