alexei.dolgolyov
f2871319cb
Some checks failed
Lint & Test / test (push) Failing after 48s
Security: tighten CORS defaults, add webhook rate limiting, fix XSS in automations, guard WebSocket JSON.parse, validate ADB address input, seal debug exception leak, URL-encode WS tokens, CSS.escape in selectors. Code quality: add Pydantic models for brightness/power endpoints, fix thread safety and name uniqueness in DeviceStore, immutable update pattern, split 6 oversized files into 16 focused modules, enable TypeScript strictNullChecks (741→102 errors), type state variables, add dom-utils helper, migrate 3 modules from inline onclick to event delegation, ProcessorDependencies dataclass. Performance: async store saves, health endpoint log level, command palette debounce, optimized entity-events comparison, fix service worker precache list. Testing: expand from 45 to 293 passing tests — add store tests (141), route tests (25), core logic tests (42), E2E flow tests (33), organize into tests/api/, tests/storage/, tests/core/, tests/e2e/. DevOps: CI test pipeline, pre-commit config, Dockerfile multi-stage build with non-root user and health check, docker-compose improvements, version bump to 0.2.0. Docs: rewrite CLAUDE.md (202→56 lines), server/CLAUDE.md (212→76), create contexts/server-operations.md, fix .js→.ts references, fix env var prefix in README, rewrite INSTALLATION.md, add CONTRIBUTING.md and .env.example.
286 lines
9.8 KiB
Python
286 lines
9.8 KiB
Python
"""Tests for DeviceStore — device CRUD, persistence, name uniqueness, thread safety."""
|
|
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from wled_controller.storage.device_store import Device, DeviceStore
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_storage(tmp_path) -> Path:
|
|
return tmp_path / "devices.json"
|
|
|
|
|
|
@pytest.fixture
|
|
def store(temp_storage) -> DeviceStore:
|
|
return DeviceStore(temp_storage)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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, temp_storage):
|
|
s1 = DeviceStore(temp_storage)
|
|
d = s1.create_device(name="Persist", url="http://p", led_count=77)
|
|
did = d.id
|
|
|
|
s2 = DeviceStore(temp_storage)
|
|
loaded = s2.get_device(did)
|
|
assert loaded.name == "Persist"
|
|
assert loaded.led_count == 77
|
|
|
|
def test_update_persists(self, temp_storage):
|
|
s1 = DeviceStore(temp_storage)
|
|
d = s1.create_device(name="Before", url="http://x", led_count=10)
|
|
s1.update_device(d.id, name="After")
|
|
|
|
s2 = DeviceStore(temp_storage)
|
|
assert s2.get_device(d.id).name == "After"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Thread safety
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeviceThreadSafety:
|
|
def test_concurrent_creates(self, tmp_path):
|
|
s = DeviceStore(tmp_path / "conc.json")
|
|
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
|