Files
led-controller/tests/models/test_device.py
pi f8eba0ee7e feat(api): tcp driver registry, identify, preset push delivery
- Track Wi-Fi TCP clients, liveness pings, disconnect broadcast, bind errors via gather\n- Device list/get include connected; POST identify with __identify preset\n- Presets push/send delivery helpers; bump led-driver hello type

Made-with: Cursor
2026-04-06 00:21:57 +12:00

165 lines
5.2 KiB
Python

import os
import sys
from pathlib import Path
# 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()
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 == 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["type"] == "led"
assert device["transport"] == "espnow"
assert device["address"] == mac
assert device["default_pattern"] == "on"
assert device["tabs"] == ["1", "2"]
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"] == mac
print("\nTesting update device fields")
update_data = {
"name": "Updated Device",
"default_pattern": "rainbow",
"tabs": ["1", "2", "3"],
}
result = devices.update(device_id, update_data)
assert result is True
updated = devices.read(device_id)
assert updated["name"] == "Updated Device"
assert updated["default_pattern"] == "rainbow"
assert len(updated["tabs"]) == 3
print("\nTesting list devices")
device_list = devices.list()
print(f"Device list: {device_list}")
assert mac in device_list
print("\nTesting delete device")
deleted = devices.delete(device_id)
assert deleted is True
assert mac not in devices
print("\nTesting read after delete")
device = devices.read(device_id)
assert device is None
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["type"] == "led"
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"
assert (
devices.upsert_wifi_tcp_client(
"kitchen", "192.168.1.100", m1, device_type="bogus"
)
== m1
)
assert devices.read(m1)["type"] == "led"
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()