alexei.dolgolyov
888f8fd16e
ruff --select UP007,UP045 --fix converted ~1760 sites across the backend: `Optional[T]` → `T | None`, `Union[X, Y]` → `X | Y`. The remaining module-level alias targets that ruff conservatively skips (BindableFloatInput, ColorList, DeviceConfig) were converted by hand earlier in the pass. black -formatted the result so the wider unions fit cleanly under the 100-char line budget. pyproject.toml now sets [tool.ruff.lint] extend-select = ["UP007", "UP045"] so future legacy imports fire CI on every push. The pre-commit ruff hook was bumped from v0.8.0 -> v0.15.12 to recognise UP045 (split off from UP007 in v0.13).
303 lines
10 KiB
Python
303 lines
10 KiB
Python
"""Tests for DeviceStore — device CRUD, persistence, name uniqueness, thread safety."""
|
|
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
import pytest
|
|
|
|
from ledgrab.storage.device_store import Device, DeviceStore
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_db(tmp_path):
|
|
from ledgrab.storage.database import Database
|
|
|
|
db = Database(tmp_path / "test.db")
|
|
yield db
|
|
db.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_db) -> DeviceStore:
|
|
return DeviceStore(tmp_db)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Device model
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeviceModel:
|
|
def test_creation_defaults(self):
|
|
d = Device(device_id="d1", name="D", url="http://1.2.3.4", led_count=100)
|
|
assert d.id == "d1"
|
|
assert d.enabled is True
|
|
assert d.device_type == "wled"
|
|
assert d.software_brightness == 255
|
|
assert d.rgbw is False
|
|
assert d.zone_mode == "combined"
|
|
assert d.tags == []
|
|
|
|
def test_creation_with_all_fields(self):
|
|
d = Device(
|
|
device_id="d2",
|
|
name="Full",
|
|
url="http://1.2.3.4",
|
|
led_count=300,
|
|
enabled=False,
|
|
device_type="adalight",
|
|
baud_rate=115200,
|
|
software_brightness=128,
|
|
auto_shutdown=True,
|
|
send_latency_ms=10,
|
|
rgbw=True,
|
|
zone_mode="individual",
|
|
tags=["living", "tv"],
|
|
dmx_protocol="sacn",
|
|
dmx_start_universe=1,
|
|
dmx_start_channel=5,
|
|
)
|
|
assert d.enabled is False
|
|
assert d.baud_rate == 115200
|
|
assert d.rgbw is True
|
|
assert d.tags == ["living", "tv"]
|
|
assert d.dmx_protocol == "sacn"
|
|
|
|
def test_to_dict_round_trip(self):
|
|
original = Device(
|
|
device_id="rt1",
|
|
name="RoundTrip",
|
|
url="http://10.0.0.1",
|
|
led_count=60,
|
|
rgbw=True,
|
|
tags=["test"],
|
|
)
|
|
data = original.to_dict()
|
|
restored = Device.from_dict(data)
|
|
|
|
assert restored.id == original.id
|
|
assert restored.name == original.name
|
|
assert restored.url == original.url
|
|
assert restored.led_count == original.led_count
|
|
assert restored.rgbw == original.rgbw
|
|
assert restored.tags == original.tags
|
|
|
|
def test_to_dict_omits_defaults(self):
|
|
"""Fields at their default value should be omitted from to_dict for compactness."""
|
|
d = Device(device_id="d", name="D", url="http://x", led_count=10)
|
|
data = d.to_dict()
|
|
assert "baud_rate" not in data
|
|
assert "rgbw" not in data
|
|
assert "tags" not in data
|
|
|
|
def test_to_dict_includes_non_defaults(self):
|
|
d = Device(
|
|
device_id="d",
|
|
name="D",
|
|
url="http://x",
|
|
led_count=10,
|
|
rgbw=True,
|
|
tags=["a"],
|
|
software_brightness=100,
|
|
)
|
|
data = d.to_dict()
|
|
assert data["rgbw"] is True
|
|
assert data["tags"] == ["a"]
|
|
assert data["software_brightness"] == 100
|
|
|
|
def test_from_dict_missing_optional_fields(self):
|
|
"""from_dict should handle minimal data gracefully."""
|
|
data = {"id": "m1", "name": "Minimal", "url": "http://x", "led_count": 10}
|
|
d = Device.from_dict(data)
|
|
assert d.enabled is True
|
|
assert d.device_type == "wled"
|
|
assert d.tags == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DeviceStore CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeviceStoreCRUD:
|
|
def test_init_empty(self, store):
|
|
assert store.count() == 0
|
|
|
|
def test_create_device(self, store):
|
|
d = store.create_device(name="Test", url="http://1.2.3.4", led_count=100)
|
|
assert d.id.startswith("device_")
|
|
assert d.name == "Test"
|
|
assert store.count() == 1
|
|
|
|
def test_create_device_with_options(self, store):
|
|
d = store.create_device(
|
|
name="Full",
|
|
url="http://1.2.3.4",
|
|
led_count=200,
|
|
device_type="adalight",
|
|
baud_rate=115200,
|
|
auto_shutdown=True,
|
|
rgbw=True,
|
|
tags=["bedroom"],
|
|
)
|
|
assert d.device_type == "adalight"
|
|
assert d.baud_rate == 115200
|
|
assert d.auto_shutdown is True
|
|
assert d.rgbw is True
|
|
assert d.tags == ["bedroom"]
|
|
|
|
def test_create_mock_device_url(self, store):
|
|
d = store.create_device(
|
|
name="MockDev", url="http://whatever", led_count=10, device_type="mock"
|
|
)
|
|
assert d.url.startswith("mock://")
|
|
|
|
def test_get_device(self, store):
|
|
created = store.create_device(name="Get", url="http://x", led_count=50)
|
|
got = store.get_device(created.id)
|
|
assert got.name == "Get"
|
|
assert got.led_count == 50
|
|
|
|
def test_get_device_not_found(self, store):
|
|
with pytest.raises(ValueError, match="not found"):
|
|
store.get_device("no_such_id")
|
|
|
|
def test_get_all_devices(self, store):
|
|
store.create_device("A", "http://a", 10)
|
|
store.create_device("B", "http://b", 20)
|
|
all_devices = store.get_all_devices()
|
|
assert len(all_devices) == 2
|
|
names = {d.name for d in all_devices}
|
|
assert names == {"A", "B"}
|
|
|
|
def test_update_device(self, store):
|
|
d = store.create_device(name="Old", url="http://x", led_count=100)
|
|
updated = store.update_device(d.id, name="New", led_count=200)
|
|
assert updated.name == "New"
|
|
assert updated.led_count == 200
|
|
assert updated.id == d.id
|
|
|
|
def test_update_device_ignores_none(self, store):
|
|
d = store.create_device(name="Keep", url="http://x", led_count=100)
|
|
updated = store.update_device(d.id, name=None, led_count=200)
|
|
assert updated.name == "Keep"
|
|
assert updated.led_count == 200
|
|
|
|
def test_update_device_ignores_unknown_fields(self, store):
|
|
d = store.create_device(name="Unk", url="http://x", led_count=100)
|
|
updated = store.update_device(d.id, bogus_field="ignored")
|
|
assert updated.name == "Unk"
|
|
|
|
def test_update_device_not_found(self, store):
|
|
with pytest.raises(ValueError, match="not found"):
|
|
store.update_device("missing", name="X")
|
|
|
|
def test_delete_device(self, store):
|
|
d = store.create_device(name="Del", url="http://x", led_count=50)
|
|
store.delete_device(d.id)
|
|
assert store.count() == 0
|
|
with pytest.raises(ValueError, match="not found"):
|
|
store.get_device(d.id)
|
|
|
|
def test_delete_device_not_found(self, store):
|
|
with pytest.raises(ValueError, match="not found"):
|
|
store.delete_device("missing")
|
|
|
|
def test_device_exists(self, store):
|
|
d = store.create_device(name="E", url="http://x", led_count=10)
|
|
assert store.device_exists(d.id) is True
|
|
assert store.device_exists("nope") is False
|
|
|
|
def test_clear(self, store):
|
|
store.create_device("A", "http://a", 10)
|
|
store.create_device("B", "http://b", 20)
|
|
store.clear()
|
|
assert store.count() == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Name uniqueness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeviceNameUniqueness:
|
|
def test_duplicate_name_on_create(self, store):
|
|
store.create_device(name="Same", url="http://a", led_count=10)
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
store.create_device(name="Same", url="http://b", led_count=10)
|
|
|
|
def test_duplicate_name_on_update(self, store):
|
|
store.create_device(name="First", url="http://a", led_count=10)
|
|
d2 = store.create_device(name="Second", url="http://b", led_count=10)
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
store.update_device(d2.id, name="First")
|
|
|
|
def test_rename_to_own_name_ok(self, store):
|
|
d = store.create_device(name="Self", url="http://a", led_count=10)
|
|
updated = store.update_device(d.id, name="Self", led_count=99)
|
|
assert updated.led_count == 99
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDevicePersistence:
|
|
def test_persistence_across_instances(self, tmp_path):
|
|
from ledgrab.storage.database import Database
|
|
|
|
db = Database(tmp_path / "persist.db")
|
|
s1 = DeviceStore(db)
|
|
d = s1.create_device(name="Persist", url="http://p", led_count=77)
|
|
did = d.id
|
|
|
|
s2 = DeviceStore(db)
|
|
loaded = s2.get_device(did)
|
|
assert loaded.name == "Persist"
|
|
assert loaded.led_count == 77
|
|
db.close()
|
|
|
|
def test_update_persists(self, tmp_path):
|
|
from ledgrab.storage.database import Database
|
|
|
|
db = Database(tmp_path / "persist2.db")
|
|
s1 = DeviceStore(db)
|
|
d = s1.create_device(name="Before", url="http://x", led_count=10)
|
|
s1.update_device(d.id, name="After")
|
|
|
|
s2 = DeviceStore(db)
|
|
assert s2.get_device(d.id).name == "After"
|
|
db.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Thread safety
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeviceThreadSafety:
|
|
def test_concurrent_creates(self, tmp_path):
|
|
from ledgrab.storage.database import Database
|
|
|
|
db = Database(tmp_path / "conc.db")
|
|
s = DeviceStore(db)
|
|
errors = []
|
|
|
|
def _create(i):
|
|
try:
|
|
s.create_device(name=f"Dev {i}", url=f"http://{i}", led_count=10)
|
|
except Exception as e:
|
|
errors.append(e)
|
|
|
|
with ThreadPoolExecutor(max_workers=8) as pool:
|
|
futures = [pool.submit(_create, i) for i in range(25)]
|
|
for f in as_completed(futures):
|
|
f.result()
|
|
|
|
assert len(errors) == 0
|
|
assert s.count() == 25
|