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.
260 lines
8.6 KiB
Python
260 lines
8.6 KiB
Python
"""Tests for ValueSourceStore — CRUD for all source types."""
|
|
|
|
import pytest
|
|
|
|
from wled_controller.storage.value_source import (
|
|
AdaptiveValueSource,
|
|
AnimatedValueSource,
|
|
AudioValueSource,
|
|
DaylightValueSource,
|
|
StaticValueSource,
|
|
ValueSource,
|
|
)
|
|
from wled_controller.storage.value_source_store import ValueSourceStore
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path) -> ValueSourceStore:
|
|
return ValueSourceStore(str(tmp_path / "value_sources.json"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ValueSource model round-trips
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValueSourceModels:
|
|
def test_static_round_trip(self):
|
|
data = {
|
|
"id": "vs_1",
|
|
"name": "Static",
|
|
"source_type": "static",
|
|
"value": 0.75,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, StaticValueSource)
|
|
assert src.value == 0.75
|
|
|
|
restored = ValueSource.from_dict(src.to_dict())
|
|
assert restored.value == 0.75
|
|
|
|
def test_animated_round_trip(self):
|
|
data = {
|
|
"id": "vs_2",
|
|
"name": "Wave",
|
|
"source_type": "animated",
|
|
"waveform": "triangle",
|
|
"speed": 30.0,
|
|
"min_value": 0.2,
|
|
"max_value": 0.8,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, AnimatedValueSource)
|
|
assert src.waveform == "triangle"
|
|
assert src.speed == 30.0
|
|
|
|
def test_audio_round_trip(self):
|
|
data = {
|
|
"id": "vs_3",
|
|
"name": "Audio",
|
|
"source_type": "audio",
|
|
"audio_source_id": "as_1",
|
|
"mode": "peak",
|
|
"sensitivity": 2.0,
|
|
"smoothing": 0.5,
|
|
"auto_gain": True,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, AudioValueSource)
|
|
assert src.mode == "peak"
|
|
assert src.auto_gain is True
|
|
|
|
def test_adaptive_time_round_trip(self):
|
|
data = {
|
|
"id": "vs_4",
|
|
"name": "Time",
|
|
"source_type": "adaptive_time",
|
|
"schedule": [
|
|
{"time": "00:00", "value": 0.1},
|
|
{"time": "12:00", "value": 1.0},
|
|
],
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, AdaptiveValueSource)
|
|
assert len(src.schedule) == 2
|
|
|
|
def test_adaptive_scene_round_trip(self):
|
|
data = {
|
|
"id": "vs_5",
|
|
"name": "Scene",
|
|
"source_type": "adaptive_scene",
|
|
"picture_source_id": "ps_1",
|
|
"scene_behavior": "match",
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, AdaptiveValueSource)
|
|
assert src.scene_behavior == "match"
|
|
|
|
def test_daylight_round_trip(self):
|
|
data = {
|
|
"id": "vs_6",
|
|
"name": "Daylight",
|
|
"source_type": "daylight",
|
|
"speed": 2.0,
|
|
"use_real_time": True,
|
|
"latitude": 55.0,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, DaylightValueSource)
|
|
assert src.use_real_time is True
|
|
assert src.latitude == 55.0
|
|
|
|
def test_unknown_type_defaults_to_static(self):
|
|
data = {
|
|
"id": "vs_u",
|
|
"name": "Unknown",
|
|
"source_type": "unknown_future",
|
|
"value": 0.5,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, StaticValueSource)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ValueSourceStore CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValueSourceStoreCRUD:
|
|
def test_create_static(self, store):
|
|
s = store.create_source(name="S1", source_type="static", value=0.5)
|
|
assert s.id.startswith("vs_")
|
|
assert isinstance(s, StaticValueSource)
|
|
assert s.value == 0.5
|
|
assert store.count() == 1
|
|
|
|
def test_create_animated(self, store):
|
|
s = store.create_source(
|
|
name="A1", source_type="animated",
|
|
waveform="sawtooth", speed=20.0,
|
|
)
|
|
assert isinstance(s, AnimatedValueSource)
|
|
assert s.waveform == "sawtooth"
|
|
assert s.speed == 20.0
|
|
|
|
def test_create_audio(self, store):
|
|
s = store.create_source(
|
|
name="Au1", source_type="audio",
|
|
audio_source_id="as_1", mode="beat",
|
|
)
|
|
assert isinstance(s, AudioValueSource)
|
|
assert s.mode == "beat"
|
|
|
|
def test_create_adaptive_time(self, store):
|
|
schedule = [
|
|
{"time": "08:00", "value": 0.5},
|
|
{"time": "20:00", "value": 1.0},
|
|
]
|
|
s = store.create_source(
|
|
name="AT", source_type="adaptive_time", schedule=schedule,
|
|
)
|
|
assert isinstance(s, AdaptiveValueSource)
|
|
assert len(s.schedule) == 2
|
|
|
|
def test_create_adaptive_time_insufficient_schedule(self, store):
|
|
with pytest.raises(ValueError, match="at least 2 points"):
|
|
store.create_source(
|
|
name="Bad", source_type="adaptive_time",
|
|
schedule=[{"time": "12:00", "value": 0.5}],
|
|
)
|
|
|
|
def test_create_daylight(self, store):
|
|
s = store.create_source(
|
|
name="DL", source_type="daylight",
|
|
speed=2.0, use_real_time=True, latitude=48.0,
|
|
)
|
|
assert isinstance(s, DaylightValueSource)
|
|
assert s.use_real_time is True
|
|
|
|
def test_create_invalid_type(self, store):
|
|
with pytest.raises(ValueError, match="Invalid source type"):
|
|
store.create_source(name="Bad", source_type="invalid")
|
|
|
|
def test_get_all(self, store):
|
|
store.create_source("A", "static")
|
|
store.create_source("B", "static")
|
|
assert len(store.get_all_sources()) == 2
|
|
|
|
def test_get(self, store):
|
|
created = store.create_source("Get", "static", value=0.3)
|
|
got = store.get_source(created.id)
|
|
assert got.name == "Get"
|
|
|
|
def test_delete(self, store):
|
|
s = store.create_source("Del", "static")
|
|
store.delete_source(s.id)
|
|
assert store.count() == 0
|
|
|
|
def test_update_static(self, store):
|
|
s = store.create_source("Stat", "static", value=0.5)
|
|
updated = store.update_source(s.id, value=0.9)
|
|
assert isinstance(updated, StaticValueSource)
|
|
assert updated.value == 0.9
|
|
|
|
def test_update_name(self, store):
|
|
s = store.create_source("Old", "static")
|
|
updated = store.update_source(s.id, name="New")
|
|
assert updated.name == "New"
|
|
|
|
def test_update_animated_fields(self, store):
|
|
s = store.create_source("Anim", "animated", waveform="sine", speed=10.0)
|
|
updated = store.update_source(s.id, waveform="square", speed=30.0)
|
|
assert isinstance(updated, AnimatedValueSource)
|
|
assert updated.waveform == "square"
|
|
assert updated.speed == 30.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Name uniqueness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValueSourceNameUniqueness:
|
|
def test_duplicate_name(self, store):
|
|
store.create_source("Same", "static")
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
store.create_source("Same", "animated")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValueSourcePersistence:
|
|
def test_persist_and_reload(self, tmp_path):
|
|
path = str(tmp_path / "vs_persist.json")
|
|
s1 = ValueSourceStore(path)
|
|
src = s1.create_source("Persist", "static", value=0.42)
|
|
sid = src.id
|
|
|
|
s2 = ValueSourceStore(path)
|
|
loaded = s2.get_source(sid)
|
|
assert loaded.name == "Persist"
|
|
assert isinstance(loaded, StaticValueSource)
|
|
assert loaded.value == 0.42
|