165 lines
5.2 KiB
Python
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", zones=["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["zones"] == ["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",
|
|
"zones": ["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["zones"]) == 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()
|