alexei.dolgolyov
02cd9d519c
Lint & Test / test (push) Successful in 1m56s
- Rename Python package: wled_controller -> ledgrab - Rename env var prefix: WLED_ -> LEDGRAB_ (with auto-migration for old vars) - Rename localStorage key: wled_api_key -> ledgrab_api_key (with migration) - Rename HA integration domain: wled_screen_controller -> ledgrab - Update all imports, build scripts, Docker, installer, config, docs - Remove HA integration (moved to ledgrab-haos-integration repo) - Remove hacs.json (belongs in HA repo now) - Add startup warning for users with old WLED_ env vars - All tests pass (715/715), ruff clean, tsc clean, frontend builds
289 lines
11 KiB
Python
289 lines
11 KiB
Python
"""Tests for AudioSourceStore — capture/processed source CRUD, chain resolution, cycle detection."""
|
|
|
|
import pytest
|
|
|
|
from ledgrab.storage.audio_source import CaptureAudioSource, ProcessedAudioSource
|
|
from ledgrab.storage.audio_source_store import AudioSourceStore, ResolvedAudioSource
|
|
from ledgrab.storage.database import Database
|
|
|
|
# Ensure audio filter registration for any template-related code
|
|
import ledgrab.core.audio.filters # noqa: F401
|
|
|
|
|
|
@pytest.fixture
|
|
def audio_store(tmp_path):
|
|
"""Provide an AudioSourceStore backed by a temp database."""
|
|
db = Database(tmp_path / "test.db")
|
|
store = AudioSourceStore(db)
|
|
yield store
|
|
db.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CaptureAudioSource CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCaptureSource:
|
|
def test_create_capture(self, audio_store):
|
|
s = audio_store.create_source(
|
|
name="System Audio",
|
|
source_type="capture",
|
|
device_index=0,
|
|
is_loopback=True,
|
|
)
|
|
assert s.id.startswith("as_")
|
|
assert isinstance(s, CaptureAudioSource)
|
|
assert s.source_type == "capture"
|
|
assert s.device_index == 0
|
|
assert s.is_loopback is True
|
|
|
|
def test_create_capture_defaults(self, audio_store):
|
|
s = audio_store.create_source(name="Default", source_type="capture")
|
|
assert isinstance(s, CaptureAudioSource)
|
|
assert s.device_index == -1
|
|
assert s.is_loopback is True
|
|
|
|
def test_update_capture_device(self, audio_store):
|
|
s = audio_store.create_source(name="Mic", source_type="capture", device_index=0)
|
|
updated = audio_store.update_source(s.id, device_index=3)
|
|
assert isinstance(updated, CaptureAudioSource)
|
|
assert updated.device_index == 3
|
|
|
|
def test_delete_capture(self, audio_store):
|
|
s = audio_store.create_source(name="ToDelete", source_type="capture")
|
|
audio_store.delete_source(s.id)
|
|
with pytest.raises(ValueError):
|
|
audio_store.get_source(s.id)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ProcessedAudioSource CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProcessedSource:
|
|
def test_create_processed(self, audio_store):
|
|
parent = audio_store.create_source(name="Parent", source_type="capture")
|
|
s = audio_store.create_source(
|
|
name="Processed",
|
|
source_type="processed",
|
|
audio_source_id=parent.id,
|
|
audio_processing_template_id="apt_test_001",
|
|
)
|
|
assert isinstance(s, ProcessedAudioSource)
|
|
assert s.audio_source_id == parent.id
|
|
assert s.audio_processing_template_id == "apt_test_001"
|
|
|
|
def test_create_processed_missing_parent_raises(self, audio_store):
|
|
with pytest.raises(ValueError, match="Parent audio source not found"):
|
|
audio_store.create_source(
|
|
name="Orphan",
|
|
source_type="processed",
|
|
audio_source_id="as_nonexistent",
|
|
audio_processing_template_id="apt_test_001",
|
|
)
|
|
|
|
def test_create_processed_no_source_id_raises(self, audio_store):
|
|
with pytest.raises(ValueError, match="audio_source_id"):
|
|
audio_store.create_source(
|
|
name="Bad",
|
|
source_type="processed",
|
|
audio_processing_template_id="apt_test_001",
|
|
)
|
|
|
|
def test_create_processed_no_template_raises(self, audio_store):
|
|
parent = audio_store.create_source(name="P", source_type="capture")
|
|
with pytest.raises(ValueError, match="audio_processing_template_id"):
|
|
audio_store.create_source(
|
|
name="Bad",
|
|
source_type="processed",
|
|
audio_source_id=parent.id,
|
|
)
|
|
|
|
def test_invalid_source_type_raises(self, audio_store):
|
|
with pytest.raises(ValueError, match="Invalid source type"):
|
|
audio_store.create_source(name="Bad", source_type="unknown")
|
|
|
|
def test_delete_parent_with_child_raises(self, audio_store):
|
|
parent = audio_store.create_source(name="Parent", source_type="capture")
|
|
audio_store.create_source(
|
|
name="Child",
|
|
source_type="processed",
|
|
audio_source_id=parent.id,
|
|
audio_processing_template_id="apt_test",
|
|
)
|
|
with pytest.raises(ValueError, match="referenced by"):
|
|
audio_store.delete_source(parent.id)
|
|
|
|
def test_delete_child_then_parent(self, audio_store):
|
|
parent = audio_store.create_source(name="Parent", source_type="capture")
|
|
child = audio_store.create_source(
|
|
name="Child",
|
|
source_type="processed",
|
|
audio_source_id=parent.id,
|
|
audio_processing_template_id="apt_test",
|
|
)
|
|
audio_store.delete_source(child.id)
|
|
audio_store.delete_source(parent.id)
|
|
assert len(audio_store.get_all_sources()) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Chain resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestChainResolution:
|
|
def test_resolve_capture_source(self, audio_store):
|
|
s = audio_store.create_source(
|
|
name="Mic",
|
|
source_type="capture",
|
|
device_index=2,
|
|
is_loopback=False,
|
|
audio_template_id="atpl_001",
|
|
)
|
|
resolved = audio_store.resolve_audio_source(s.id)
|
|
assert isinstance(resolved, ResolvedAudioSource)
|
|
assert resolved.device_index == 2
|
|
assert resolved.is_loopback is False
|
|
assert resolved.audio_template_id == "atpl_001"
|
|
assert resolved.audio_processing_template_ids == []
|
|
|
|
def test_resolve_processed_chain(self, audio_store):
|
|
capture = audio_store.create_source(name="Capture", source_type="capture", device_index=0)
|
|
proc = audio_store.create_source(
|
|
name="Processed",
|
|
source_type="processed",
|
|
audio_source_id=capture.id,
|
|
audio_processing_template_id="apt_tpl_A",
|
|
)
|
|
resolved = audio_store.resolve_audio_source(proc.id)
|
|
assert resolved.device_index == 0
|
|
assert resolved.audio_processing_template_ids == ["apt_tpl_A"]
|
|
|
|
def test_resolve_deep_chain(self, audio_store):
|
|
"""A -> B -> C (capture). Template IDs collected outermost first."""
|
|
capture = audio_store.create_source(name="C", source_type="capture", device_index=1)
|
|
b = audio_store.create_source(
|
|
name="B",
|
|
source_type="processed",
|
|
audio_source_id=capture.id,
|
|
audio_processing_template_id="apt_B",
|
|
)
|
|
a = audio_store.create_source(
|
|
name="A",
|
|
source_type="processed",
|
|
audio_source_id=b.id,
|
|
audio_processing_template_id="apt_A",
|
|
)
|
|
resolved = audio_store.resolve_audio_source(a.id)
|
|
assert resolved.device_index == 1
|
|
assert resolved.audio_processing_template_ids == ["apt_A", "apt_B"]
|
|
|
|
def test_resolve_nonexistent_raises(self, audio_store):
|
|
with pytest.raises(ValueError):
|
|
audio_store.resolve_audio_source("as_nonexistent")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cycle detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCycleDetection:
|
|
def test_update_to_self_raises(self, audio_store):
|
|
capture = audio_store.create_source(name="Cap", source_type="capture")
|
|
proc = audio_store.create_source(
|
|
name="Proc",
|
|
source_type="processed",
|
|
audio_source_id=capture.id,
|
|
audio_processing_template_id="apt_t",
|
|
)
|
|
with pytest.raises(ValueError, match="circular"):
|
|
audio_store.update_source(proc.id, audio_source_id=proc.id)
|
|
|
|
def test_cycle_in_chain_raises(self, audio_store):
|
|
capture = audio_store.create_source(name="C", source_type="capture")
|
|
a = audio_store.create_source(
|
|
name="A",
|
|
source_type="processed",
|
|
audio_source_id=capture.id,
|
|
audio_processing_template_id="apt_t",
|
|
)
|
|
b = audio_store.create_source(
|
|
name="B",
|
|
source_type="processed",
|
|
audio_source_id=a.id,
|
|
audio_processing_template_id="apt_t",
|
|
)
|
|
# Try to make A point to B, creating A -> B -> A cycle
|
|
with pytest.raises(ValueError, match="circular"):
|
|
audio_store.update_source(a.id, audio_source_id=b.id)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reference query helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestReferenceHelpers:
|
|
def test_get_sources_referencing_template(self, audio_store):
|
|
capture = audio_store.create_source(name="Cap", source_type="capture")
|
|
audio_store.create_source(
|
|
name="P1",
|
|
source_type="processed",
|
|
audio_source_id=capture.id,
|
|
audio_processing_template_id="apt_shared",
|
|
)
|
|
audio_store.create_source(
|
|
name="P2",
|
|
source_type="processed",
|
|
audio_source_id=capture.id,
|
|
audio_processing_template_id="apt_shared",
|
|
)
|
|
audio_store.create_source(
|
|
name="P3",
|
|
source_type="processed",
|
|
audio_source_id=capture.id,
|
|
audio_processing_template_id="apt_other",
|
|
)
|
|
refs = audio_store.get_sources_referencing_template("apt_shared")
|
|
assert len(refs) == 2
|
|
names = {r.name for r in refs}
|
|
assert names == {"P1", "P2"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPersistence:
|
|
def test_sources_survive_reload(self, tmp_path):
|
|
db = Database(tmp_path / "persist.db")
|
|
store = AudioSourceStore(db)
|
|
s = store.create_source(name="Persisted", source_type="capture", device_index=5)
|
|
sid = s.id
|
|
db.close()
|
|
|
|
db2 = Database(tmp_path / "persist.db")
|
|
store2 = AudioSourceStore(db2)
|
|
reloaded = store2.get_source(sid)
|
|
assert reloaded.name == "Persisted"
|
|
assert isinstance(reloaded, CaptureAudioSource)
|
|
assert reloaded.device_index == 5
|
|
db2.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Name uniqueness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNameUniqueness:
|
|
def test_duplicate_name_raises(self, audio_store):
|
|
audio_store.create_source(name="MySource", source_type="capture")
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
audio_store.create_source(name="MySource", source_type="capture")
|