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

feat(devices): wifi tcp registry, device API/UI, tests; bump led-tool

Browse Source 
Made-with: Cursor
This commit is contained in:
pi
2026-04-05 21:13:07 +12:00
parent fbae75b957
commit e6b5bf2cf1
15 changed files with 825 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
tests/conftest.py
View File

@@ -5,7 +5,9 @@ PROJECT_ROOT = Path(__file__).resolve().parents[1]
SRC_PATH = PROJECT_ROOT / "src"
LIB_PATH = PROJECT_ROOT / "lib"
for p in (str(SRC_PATH), str(LIB_PATH), str(PROJECT_ROOT)):
# Last insert(0) wins: order must be (root, lib, src) so src/models wins over
# tests/models (same package name "models" on sys.path when pytest imports tests).
for p in (str(PROJECT_ROOT), str(LIB_PATH), str(SRC_PATH)):
if p in sys.path:
sys.path.remove(p)
sys.path.insert(0, p)

115
tests/models/test_device.py
View File

@@ -1,36 +1,67 @@
from models.device import Device
import os
import sys
from pathlib import Path
def test_device():
"""Test Device model CRUD operations."""
db_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "db")
# Prefer src/models; pytest may have registered tests/models as top-level ``models``.
_src = Path(__file__).resolve().parents[2] / "src"
_sp = str(_src)
if _sp in sys.path:
sys.path.remove(_sp)
sys.path.insert(0, _sp)
_m = sys.modules.get("models")
if _m is not None:
mf = (getattr(_m, "__file__", "") or "").replace("\\", "/")
if "/tests/models" in mf:
del sys.modules["models"]
from models.device import Device
def _fresh_device():
"""New empty device DB and new Device singleton (tests only)."""
db_dir = os.path.join(
os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "db"
)
device_file = os.path.join(db_dir, "device.json")
if os.path.exists(device_file):
os.remove(device_file)
if hasattr(Device, "_instance"):
del Device._instance
return Device()
devices = Device()
def test_device():
"""Test Device model CRUD operations (id = MAC)."""
devices = _fresh_device()
mac = "aabbccddeeff"
print("Testing create device")
device_id = devices.create("Test Device", address="aa:bb:cc:dd:ee:ff", default_pattern="on", tabs=["1", "2"])
print(f"Created device with ID: {device_id}")
assert device_id is not None
assert device_id == mac
assert device_id in devices
print("\nTesting read device")
device = devices.read(device_id)
print(f"Read: {device}")
assert device is not None
assert device["id"] == mac
assert device["name"] == "Test Device"
assert device["address"] == "aabbccddeeff"
assert device["type"] == "led"
assert device["transport"] == "espnow"
assert device["address"] == mac
assert device["default_pattern"] == "on"
assert device["tabs"] == ["1", "2"]
print("\nTesting address normalization")
print("\nTesting read by colon MAC")
assert devices.read("aa:bb:cc:dd:ee:ff")["id"] == mac
print("\nTesting address normalization on update (espnow keeps MAC as address)")
devices.update(device_id, {"address": "11:22:33:44:55:66"})
updated = devices.read(device_id)
assert updated["address"] == "112233445566"
assert updated["address"] == mac
print("\nTesting update device")
print("\nTesting update device fields")
update_data = {
"name": "Updated Device",
"default_pattern": "rainbow",
@@ -46,12 +77,12 @@ def test_device():
print("\nTesting list devices")
device_list = devices.list()
print(f"Device list: {device_list}")
assert device_id in device_list
assert mac in device_list
print("\nTesting delete device")
deleted = devices.delete(device_id)
assert deleted is True
assert device_id not in devices
assert mac not in devices
print("\nTesting read after delete")
device = devices.read(device_id)
@@ -60,5 +91,65 @@ def test_device():
print("\nAll device tests passed!")
def test_upsert_wifi_tcp_client():
devices = _fresh_device()
assert devices.upsert_wifi_tcp_client("", "192.168.1.10", None) is None
assert devices.upsert_wifi_tcp_client("kitchen", "192.168.1.20", "bad") is None
m1 = "001122334455"
m2 = "001122334466"
i1 = devices.upsert_wifi_tcp_client("kitchen", "192.168.1.20", m1)
assert i1 == m1
d = devices.read(i1)
assert d["name"] == "kitchen"
assert d["transport"] == "wifi"
assert d["address"] == "192.168.1.20"
i2 = devices.upsert_wifi_tcp_client("kitchen", "192.168.1.21", m2)
assert i2 == m2
assert devices.read(m1)["address"] == "192.168.1.20"
assert devices.read(m2)["address"] == "192.168.1.21"
assert devices.read(m1)["name"] == devices.read(m2)["name"] == "kitchen"
again = devices.upsert_wifi_tcp_client("kitchen", "192.168.1.99", m1)
assert again == m1
assert devices.read(m1)["address"] == "192.168.1.99"
i3 = devices.upsert_wifi_tcp_client("hall", "10.0.0.5", "deadbeefcafe")
assert i3 == "deadbeefcafe"
assert len(devices.list()) == 3
def test_device_can_change_address():
devices = _fresh_device()
m = "feedfacec0de"
did = devices.create("mover", mac=m, address="192.168.1.1", transport="wifi")
assert did == m
devices.update(did, {"address": "10.0.0.99"})
assert devices.read(did)["address"] == "10.0.0.99"
def test_device_duplicate_names_allowed():
devices = _fresh_device()
a1 = devices.create("alpha", address="aa:bb:cc:dd:ee:ff")
a2 = devices.create("alpha", address="11:22:33:44:55:66")
assert a1 != a2
assert devices.read(a1)["name"] == devices.read(a2)["name"] == "alpha"
def test_device_duplicate_mac_rejected():
devices = _fresh_device()
devices.create("one", address="aa:bb:cc:dd:ee:ff")
try:
devices.create("two", address="aa-bb-cc-dd-ee-ff")
assert False, "expected ValueError"
except ValueError as e:
assert "already exists" in str(e).lower()
if __name__ == "__main__":
test_device()
test_upsert_wifi_tcp_client()
test_device_can_change_address()
test_device_duplicate_names_allowed()
test_device_duplicate_mac_rejected()

7
tests/tcp_test_server.py
View File

@@ -4,9 +4,14 @@ Simple TCP test server for led-controller.
Listens on the same TCP port used by led-driver WiFi transport and
every 5 seconds sends a newline-delimited JSON message with v="1".
Clients talking to the real Pi registry should send a first line JSON object
that includes device_name and mac (12 hex) so the controller can register
the device by MAC.
"""
import asyncio
import contextlib
import json
import os
import sys
@@ -204,8 +209,6 @@ async def main():
if __name__ == "__main__":
import contextlib
try:
asyncio.run(main())
except KeyboardInterrupt:

105
tests/test_endpoints_pytest.py
View File

@@ -124,6 +124,7 @@ def server(monkeypatch, tmp_path_factory):
import models.scene as models_scene # noqa: E402
import models.pattern as models_pattern # noqa: E402
import models.squence as models_sequence # noqa: E402
import models.device as models_device # noqa: E402
for cls in (
models_preset.Preset,
@@ -134,6 +135,7 @@ def server(monkeypatch, tmp_path_factory):
models_scene.Scene,
models_pattern.Pattern,
models_sequence.Sequence,
models_device.Device,
):
if hasattr(cls, "_instance"):
delattr(cls, "_instance")
@@ -167,6 +169,7 @@ def server(monkeypatch, tmp_path_factory):
"controllers.scene",
"controllers.pattern",
"controllers.settings",
"controllers.device",
):
sys.modules.pop(mod_name, None)
@@ -180,6 +183,7 @@ def server(monkeypatch, tmp_path_factory):
import controllers.scene as scene_ctl # noqa: E402
import controllers.pattern as pattern_ctl # noqa: E402
import controllers.settings as settings_ctl # noqa: E402
import controllers.device as device_ctl # noqa: E402
# Configure transport sender used by /presets/send.
from models.transport import set_sender # noqa: E402
@@ -206,6 +210,7 @@ def server(monkeypatch, tmp_path_factory):
app.mount(scene_ctl.controller, "/scenes")
app.mount(pattern_ctl.controller, "/patterns")
app.mount(settings_ctl.controller, "/settings")
app.mount(device_ctl.controller, "/devices")
@app.route("/")
def index(request):
@@ -562,6 +567,106 @@ def test_groups_sequences_scenes_palettes_patterns_endpoints(server):
resp = c.delete(f"{base_url}/palettes/{palette_id}")
assert resp.status_code == 200
# Devices (LED driver registry).
resp = c.get(f"{base_url}/devices")
assert resp.status_code == 200
assert resp.json() == {}
resp = c.post(f"{base_url}/devices", json={})
assert resp.status_code == 400
resp = c.post(
f"{base_url}/devices",
json={"name": "pytest-dev", "address": "aa:bb:cc:dd:ee:ff"},
)
assert resp.status_code == 201
dev_map = resp.json()
dev_id = next(iter(dev_map.keys()))
assert dev_id == "aabbccddeeff"
assert dev_map[dev_id]["name"] == "pytest-dev"
assert dev_map[dev_id]["id"] == dev_id
assert dev_map[dev_id]["type"] == "led"
assert dev_map[dev_id]["transport"] == "espnow"
assert dev_map[dev_id]["address"] == "aabbccddeeff"
resp = c.get(f"{base_url}/devices/{dev_id}")
assert resp.status_code == 200
assert resp.json()["name"] == "pytest-dev"
assert resp.json()["type"] == "led"
resp = c.post(
f"{base_url}/devices",
json={
"name": "pytest-wifi",
"type": "led",
"transport": "wifi",
"address": "192.168.50.10",
"mac": "102030405060",
},
)
assert resp.status_code == 201
wid = "102030405060"
assert wid in resp.json()
assert resp.json()[wid]["transport"] == "wifi"
assert resp.json()[wid]["address"] == "192.168.50.10"
resp = c.post(
f"{base_url}/devices",
json={
"name": "pytest-wifi",
"transport": "wifi",
"address": "192.168.50.11",
"mac": "102030405061",
},
)
assert resp.status_code == 201
wid2 = "102030405061"
assert wid2 in resp.json()
assert resp.json()[wid2]["name"] == "pytest-wifi"
resp = c.post(
f"{base_url}/devices",
json={
"name": "pytest-wifi-dupmac",
"transport": "wifi",
"address": "192.168.50.99",
"mac": "102030405060",
},
)
assert resp.status_code == 409
resp = c.post(
f"{base_url}/devices",
json={"name": "no-mac-wifi", "transport": "wifi", "address": "192.168.50.12"},
)
assert resp.status_code == 400
resp = c.post(
f"{base_url}/devices",
json={"name": "bad-tr", "transport": "serial"},
)
assert resp.status_code == 400
resp = c.put(f"{base_url}/devices/{dev_id}", json={"name": " "})
assert resp.status_code == 400
resp = c.put(f"{base_url}/devices/{dev_id}", json={"name": "renamed"})
assert resp.status_code == 200
assert resp.json()["name"] == "renamed"
resp = c.put(f"{base_url}/devices/{wid}", json={"name": "renamed"})
assert resp.status_code == 200
assert resp.json()["name"] == "renamed"
resp = c.delete(f"{base_url}/devices/{wid2}")
assert resp.status_code == 200
resp = c.delete(f"{base_url}/devices/{wid}")
assert resp.status_code == 200
resp = c.delete(f"{base_url}/devices/{dev_id}")
assert resp.status_code == 200
# Patterns.
resp = c.get(f"{base_url}/patterns/definitions")
assert resp.status_code == 200