alexei.dolgolyov
2da5c047f9
Some checks failed
Lint & Test / test (push) Failing after 1m33s
All store tests were passing file paths instead of Database objects after the JSON-to-SQLite migration. Updated fixtures to create temp Database instances, rewrote backup e2e tests for binary .db format, and fixed config tests for the simplified StorageConfig.
262 lines
8.7 KiB
Python
262 lines
8.7 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_db) -> ValueSourceStore:
|
|
return ValueSourceStore(tmp_db)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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):
|
|
from wled_controller.storage.database import Database
|
|
db = Database(tmp_path / "vs_persist.db")
|
|
s1 = ValueSourceStore(db)
|
|
src = s1.create_source("Persist", "static", value=0.42)
|
|
sid = src.id
|
|
|
|
s2 = ValueSourceStore(db)
|
|
loaded = s2.get_source(sid)
|
|
assert loaded.name == "Persist"
|
|
assert isinstance(loaded, StaticValueSource)
|
|
assert loaded.value == 0.42
|
|
db.close()
|