68 lines
1.8 KiB
Python
68 lines
1.8 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"]
|