40751fecb7
Lint & Test / test (push) Successful in 1m24s
- Cache per-entity colors in HALightTargetProcessor._update_lights()
- Broadcast colors_update to WS clients at target's update_rate
- WS endpoint: /api/v1/output-targets/{target_id}/ha-light/ws
- Frontend: connect WS when target runs, update swatch colors live
- Card shows colored boxes per mapped entity with entity name labels
191 lines
6.4 KiB
Python
191 lines
6.4 KiB
Python
"""Tests for OutputTargetStore — CRUD for LED and HA light targets."""
|
|
|
|
import pytest
|
|
|
|
from wled_controller.storage.output_target import OutputTarget
|
|
from wled_controller.storage.output_target_store import OutputTargetStore
|
|
from wled_controller.storage.wled_output_target import WledOutputTarget
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_db) -> OutputTargetStore:
|
|
return OutputTargetStore(tmp_db)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OutputTarget model dispatching
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOutputTargetModel:
|
|
def test_led_from_dict(self):
|
|
data = {
|
|
"id": "pt_1",
|
|
"name": "LED Target",
|
|
"target_type": "led",
|
|
"device_id": "dev_1",
|
|
"color_strip_source_id": "css_1",
|
|
"fps": 30,
|
|
"protocol": "ddp",
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
target = OutputTarget.from_dict(data)
|
|
assert isinstance(target, WledOutputTarget)
|
|
assert target.device_id == "dev_1"
|
|
|
|
def test_unknown_type_raises(self):
|
|
data = {
|
|
"id": "pt_3",
|
|
"name": "Bad",
|
|
"target_type": "nonexistent",
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
with pytest.raises(ValueError, match="Unknown target type"):
|
|
OutputTarget.from_dict(data)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# OutputTargetStore CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOutputTargetStoreCRUD:
|
|
def test_create_led_target(self, store):
|
|
t = store.create_target(
|
|
name="LED 1",
|
|
target_type="led",
|
|
device_id="dev_1",
|
|
color_strip_source_id="css_1",
|
|
fps=60,
|
|
protocol="ddp",
|
|
)
|
|
assert t.id.startswith("pt_")
|
|
assert isinstance(t, WledOutputTarget)
|
|
assert t.name == "LED 1"
|
|
assert store.count() == 1
|
|
|
|
def test_create_invalid_type(self, store):
|
|
with pytest.raises(ValueError, match="Invalid target type"):
|
|
store.create_target(name="Bad", target_type="invalid")
|
|
|
|
def test_get_all(self, store):
|
|
store.create_target("A", "led")
|
|
store.create_target("B", "led")
|
|
assert len(store.get_all_targets()) == 2
|
|
|
|
def test_get(self, store):
|
|
created = store.create_target("Get", "led")
|
|
got = store.get_target(created.id)
|
|
assert got.name == "Get"
|
|
|
|
def test_delete(self, store):
|
|
t = store.create_target("Del", "led")
|
|
store.delete_target(t.id)
|
|
assert store.count() == 0
|
|
|
|
def test_delete_not_found(self, store):
|
|
with pytest.raises(ValueError, match="not found"):
|
|
store.delete_target("nope")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Update
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOutputTargetUpdate:
|
|
def test_update_name(self, store):
|
|
t = store.create_target("Old", "led")
|
|
updated = store.update_target(t.id, name="New")
|
|
assert updated.name == "New"
|
|
|
|
def test_update_led_fields(self, store):
|
|
t = store.create_target("LED", "led", fps=30, protocol="ddp")
|
|
updated = store.update_target(t.id, fps=60, protocol="drgb")
|
|
assert isinstance(updated, WledOutputTarget)
|
|
assert updated.fps == 60
|
|
assert updated.protocol == "drgb"
|
|
|
|
def test_update_not_found(self, store):
|
|
with pytest.raises(ValueError, match="not found"):
|
|
store.update_target("nope", name="X")
|
|
|
|
def test_update_tags(self, store):
|
|
t = store.create_target("Tags", "led", tags=["old"])
|
|
updated = store.update_target(t.id, tags=["new", "tags"])
|
|
assert updated.tags == ["new", "tags"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Name uniqueness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOutputTargetNameUniqueness:
|
|
def test_duplicate_name_create(self, store):
|
|
store.create_target("Same", "led")
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
store.create_target("Same", "led")
|
|
|
|
def test_duplicate_name_update(self, store):
|
|
store.create_target("First", "led")
|
|
t2 = store.create_target("Second", "led")
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
store.update_target(t2.id, name="First")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Query helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOutputTargetQueries:
|
|
def test_get_targets_for_device(self, store):
|
|
store.create_target("T1", "led", device_id="dev_a")
|
|
store.create_target("T2", "led", device_id="dev_b")
|
|
store.create_target("T3", "led", device_id="dev_a")
|
|
|
|
results = store.get_targets_for_device("dev_a")
|
|
assert len(results) == 2
|
|
assert all(isinstance(t, WledOutputTarget) for t in results)
|
|
|
|
def test_get_targets_for_device_empty(self, store):
|
|
assert store.get_targets_for_device("nonexistent") == []
|
|
|
|
def test_get_targets_referencing_css(self, store):
|
|
store.create_target("T1", "led", color_strip_source_id="css_x")
|
|
store.create_target("T2", "led", color_strip_source_id="css_y")
|
|
names = store.get_targets_referencing_css("css_x")
|
|
assert names == ["T1"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOutputTargetPersistence:
|
|
def test_persist_and_reload(self, tmp_path):
|
|
from wled_controller.storage.database import Database
|
|
|
|
db_path = str(tmp_path / "ot_persist.db")
|
|
db = Database(db_path)
|
|
s1 = OutputTargetStore(db)
|
|
t = s1.create_target(
|
|
"Persist",
|
|
"led",
|
|
device_id="dev_1",
|
|
fps=60,
|
|
tags=["tv"],
|
|
)
|
|
tid = t.id
|
|
|
|
s2 = OutputTargetStore(db)
|
|
loaded = s2.get_target(tid)
|
|
assert loaded.name == "Persist"
|
|
assert isinstance(loaded, WledOutputTarget)
|
|
assert loaded.tags == ["tv"]
|
|
db.close()
|