"""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")