technicalkiwi/led-controller
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

docs(espnow): update docs and tests for p2p merge

Browse Source 
Align API, architecture, and help with devices envelope transport,
bridge wifi/serial settings, and MAC-keyed device registry. Fix
endpoint tests for envelope identify payloads; remove obsolete p2p.py.
Bump led-tool for --serial-usb bridge provisioning.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Jimmy
2026-06-06 21:10:06 +12:00
parent d682753e42
commit cfdd6de291
11 changed files with 190 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
tests/README.md
View File

@@ -7,13 +7,20 @@ Tests for the LED Controller project live under **`tests/`** (pytest + legacy sc
| Path | Role |
|------|------|
| `test_endpoints.py` | HTTP endpoint checks (**`LED_CONTROLLER_RUN_DEVICE_ENDPOINT_TESTS=1`**); **`test_zones`** / **`test_zone_edit_workflow`** hit **`/zones`** |
| `test_endpoints_pytest.py` | Pytest-style endpoint coverage |
| `test_endpoints_pytest.py` | Pytest-style endpoint coverage (devices envelope transport mock) |
| `test_bridge_ws_client.py` | Bridge WebSocket client reconnect / send behaviour |
| `test_bridge_envelope.py` | Devices envelope build/split/delivery |
| `test_bridge_serial_frame.py` | Pi↔bridge USB serial framing |
| `test_bridge_wifi_connect.py` | Saved bridge profile connect (serial path) |
| `test_espnow_wire.py`, `test_espnow_ping.py` | Binary wire codec and ping registration |
| `test_binary_envelope.py` | v2 binary envelope encode/decode |
| `test_browser.py` | Selenium UI flows (set **`LED_CONTROLLER_RUN_BROWSER_TESTS=1`** to run; uses **`test_zones_ui`** and legacy **`tabsManager`** JS aliases) |
| `test_pattern_ota_send.py` | Pattern OTA / Wi-Fi send helpers |
| `test_pi_wifi_scan.py` | nmcli SSID scan helpers |
| `tcp_test_server.py`, `async_tcp_server.py` | TCP test doubles for driver protocol |
| `udp_server.py` | UDP discovery / hello test listener (port **8766**) |
| `bridge_broadcast_test.py` | Manual bridge WebSocket broadcast script |
| `ws.py` | WebSocket client checks |
| `p2p.py` | ESP-NOW–related helpers / experiments |
| `web.py` | Local dev static server (not the main app) |
| `conftest.py` | Pytest fixtures |
| `models/` | Model unit tests (`run_all.py`, `test_zone.py`, …) |

105
tests/p2p.py
View File

@@ -1,105 +0,0 @@
#!/usr/bin/env python3
# MicroPython script to test LED bar patterns over ESP-NOW (no WebSocket)
import json
import uasyncio as asyncio
# Import P2P from src/p2p.py
# Note: When running on device, ensure src/p2p.py is in the path
try:
from p2p import P2P
except ImportError:
# Fallback: import from src directory
import sys
sys.path.insert(0, 'src')
from p2p import P2P
async def main():
p2p = P2P()
# Test cases following msg.json format:
# {"g": {"df": {...}, "group_name": {...}}, "sv": true, "st": 0}
# Note: led-bar device must have matching group in settings["groups"]
tests = [
# Example 1: Default format with df defaults and dj group (matches msg.json)
{
"g": {
"df": {
"pt": "on",
"cl": ["#ff0000"],
"br": 200,
"n1": 10,
"n2": 10,
"n3": 10,
"n4": 10,
"n5": 10,
"n6": 10,
"dl": 100
},
"dj": {
"pt": "blink",
"cl": ["#00ff00"],
"dl": 500
}
},
"sv": True,
"st": 0
},
# Example 2: Different group with df defaults
{
"g": {
"df": {
"pt": "on",
"br": 150,
"dl": 100
},
"group1": {
"pt": "rainbow",
"dl": 50
}
},
"sv": False
},
# Example 3: Multiple groups
{
"g": {
"df": {
"br": 200,
"dl": 100
},
"group1": {
"pt": "on",
"cl": ["#0000ff"]
},
"group2": {
"pt": "blink",
"cl": ["#ff00ff"],
"dl": 300
}
},
"sv": True,
"st": 1
},
# Example 4: Single group without df
{
"g": {
"dj": {
"pt": "off"
}
},
"sv": False
}
]
for i, test in enumerate(tests, 1):
print(f"\n{'='*50}")
print(f"Test {i}/{len(tests)}")
print(f"Sending: {json.dumps(test, indent=2)}")
await p2p.send(json.dumps(test))
await asyncio.sleep_ms(2000)
print(f"\n{'='*50}")
print("All tests completed")
if __name__ == "__main__":
asyncio.run(main())

2
tests/test_bridge_ws_client.py
View File

@@ -20,7 +20,7 @@ def test_send_returns_false_when_not_connected():
async def _run():
client = BridgeWsClient("ws://127.0.0.1/ws", reconnect_delay_s=0.01)
async def _no_wait(_timeout=30.0):
async def _no_wait(timeout=30.0):
return False
client.wait_connected = _no_wait # type: ignore[method-assign]

63
tests/test_endpoints_pytest.py
View File

@@ -29,15 +29,56 @@ from microdot.websocket import with_websocket # noqa: E402
class DummyBridge:
def __init__(self):
self.sent: list[tuple[str, Optional[str]]] = []
self.sent: list[tuple[Any, Optional[str]]] = []
async def send(self, data: Any, addr: Optional[str] = None):
if isinstance(data, (bytes, bytearray)):
if isinstance(data, dict):
from util.bridge_envelope import ( # noqa: E402
BROADCAST_MAC,
build_devices_envelope,
format_mac_key,
is_broadcast_mac,
normalize_mac_key,
)
from util.v1_wire import compact_envelope # noqa: E402
if data.get("v") == "1" and ("devices" in data or "dv" in data):
data = compact_envelope(data)
elif addr is not None:
s = str(addr).strip().lower()
if is_broadcast_mac(s):
mac_key = BROADCAST_MAC
else:
h = normalize_mac_key(s)
mac_key = format_mac_key(h) if h else None
if mac_key:
body = {k: v for k, v in data.items() if k != "v"}
data = build_devices_envelope({mac_key: body})
else:
data = json.dumps(data, separators=(",", ":"))
else:
data = json.dumps(data, separators=(",", ":"))
elif isinstance(data, (bytes, bytearray)):
data = bytes(data).decode(errors="ignore")
self.sent.append((data, addr))
return True
def _bridge_sent_envelope(bridge: DummyBridge, index: int) -> Dict[str, Any]:
data, _addr = bridge.sent[index]
if isinstance(data, dict):
return data
return json.loads(data)
def _device_body_from_envelope(envelope: Dict[str, Any], mac: str) -> Dict[str, Any]:
from util.bridge_envelope import format_mac_key, normalize_mac_key # noqa: E402
devs = envelope.get("dv") or envelope.get("devices") or {}
key = format_mac_key(normalize_mac_key(mac))
return devs[key]
def _json(resp: requests.Response) -> Dict[str, Any]:
# Many endpoints already set Content-Type; but be tolerant for now.
return resp.json() # pragma: no cover
@@ -672,17 +713,19 @@ def test_groups_sequences_scenes_palettes_patterns_endpoints(server):
assert resp.status_code == 200
assert resp.json().get("message")
assert len(bridge.sent) >= 1
first = json.loads(bridge.sent[0][0])
assert "presets" in first and "select" in first
assert first["presets"]["__identify"]["p"] == "blink"
assert first["presets"]["__identify"]["d"] == 50
assert first["select"] == ["__identify"]
first = _bridge_sent_envelope(bridge, 0)
assert first["v"] == "1"
first_body = _device_body_from_envelope(first, dev_id)
assert first_body["p"]["__identify"]["p"] == "blink"
assert first_body["p"]["__identify"]["d"] == 50
assert first_body["s"] == ["__identify"]
deadline = time.monotonic() + 2.0
while len(bridge.sent) < 2 and time.monotonic() < deadline:
time.sleep(0.02)
assert len(bridge.sent) >= 2
second = json.loads(bridge.sent[1][0])
assert second.get("select") == ["off"]
second = _bridge_sent_envelope(bridge, 1)
second_body = _device_body_from_envelope(second, dev_id)
assert second_body["s"] == ["off"]
resp = c.post(
f"{base_url}/devices",
@@ -702,7 +745,7 @@ def test_groups_sequences_scenes_palettes_patterns_endpoints(server):
resp = c.get(f"{base_url}/devices/{wid}")
assert resp.status_code == 200
assert resp.json().get("connected") is False
assert resp.json().get("connected") is None
resp = c.post(
f"{base_url}/devices",