98 lines
2.7 KiB
Python
98 lines
2.7 KiB
Python
# tests/test_microphone_monitor.py
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from whisper_local.microphone._poll import PollMonitor
|
|
|
|
|
|
def _fake_devices(names: list[str]) -> list[dict]:
|
|
return [{"name": n, "max_input_channels": 1} for n in names]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_device_added_fires_when_device_appears():
|
|
monitor = PollMonitor(configured_device=None, interval=0.05)
|
|
event = asyncio.Event()
|
|
added: list[str] = []
|
|
|
|
async def on_added(name: str) -> None:
|
|
added.append(name)
|
|
event.set()
|
|
|
|
monitor.on_device_added = on_added
|
|
call_count = 0
|
|
|
|
def fake_query():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 1:
|
|
return _fake_devices(["Mic A"])
|
|
return _fake_devices(["Mic A", "Mic B"])
|
|
|
|
with patch("sounddevice.query_devices", side_effect=fake_query):
|
|
await monitor.start()
|
|
await asyncio.wait_for(event.wait(), timeout=1.0)
|
|
monitor.stop()
|
|
|
|
assert added == ["Mic B"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_device_removed_fires_when_device_disappears():
|
|
monitor = PollMonitor(configured_device=None, interval=0.05)
|
|
event = asyncio.Event()
|
|
removed: list[str] = []
|
|
|
|
async def on_removed(name: str) -> None:
|
|
removed.append(name)
|
|
event.set()
|
|
|
|
monitor.on_device_removed = on_removed
|
|
call_count = 0
|
|
|
|
def fake_query():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 1:
|
|
return _fake_devices(["Mic A", "Mic B"])
|
|
return _fake_devices(["Mic A"])
|
|
|
|
with patch("sounddevice.query_devices", side_effect=fake_query):
|
|
await monitor.start()
|
|
await asyncio.wait_for(event.wait(), timeout=1.0)
|
|
monitor.stop()
|
|
|
|
assert removed == ["Mic B"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_configured_missing_fires_immediately_at_start():
|
|
monitor = PollMonitor(configured_device="Headset USB", interval=99.0)
|
|
missing_called = asyncio.Event()
|
|
|
|
async def on_missing() -> None:
|
|
missing_called.set()
|
|
|
|
monitor.on_configured_missing = on_missing
|
|
|
|
with patch("sounddevice.query_devices", return_value=_fake_devices(["Mic A"])):
|
|
await monitor.start()
|
|
|
|
assert missing_called.is_set()
|
|
monitor.stop()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_configured_missing_does_not_fire_when_device_present():
|
|
monitor = PollMonitor(configured_device="Headset USB", interval=99.0)
|
|
missing_mock = AsyncMock()
|
|
monitor.on_configured_missing = missing_mock
|
|
|
|
with patch("sounddevice.query_devices", return_value=_fake_devices(["Headset USB", "Mic A"])):
|
|
await monitor.start()
|
|
|
|
missing_mock.assert_not_called()
|
|
monitor.stop()
|