alexei.dolgolyov
ce1f4847f3
Fix test_list_filters test (filter_id field name mismatch). Add tests for audio filters, template store, and source store. All 678 tests pass, ruff clean, tsc clean, esbuild clean. No dead code remaining from old source types.
289 lines
11 KiB
Python
289 lines
11 KiB
Python
"""Tests for AudioSourceStore — capture/processed source CRUD, chain resolution, cycle detection."""
|
|
|
|
import pytest
|
|
|
|
from wled_controller.storage.audio_source import CaptureAudioSource, ProcessedAudioSource
|
|
from wled_controller.storage.audio_source_store import AudioSourceStore, ResolvedAudioSource
|
|
from wled_controller.storage.database import Database
|
|
|
|
# Ensure audio filter registration for any template-related code
|
|
import wled_controller.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")
|