mirror of
https://github.com/home-assistant/core.git
synced 2026-02-28 22:06:25 +00:00
6e4c78686e
* Run pipeline from audio stream function * Fix tests --------- Co-authored-by: Michael Hansen <mike@rhasspy.org>
43 lines
1.1 KiB
Python
43 lines
1.1 KiB
Python
"""Test Voice Assistant init."""
|
|
|
|
from syrupy.assertion import SnapshotAssertion
|
|
|
|
from homeassistant.components import stt, voice_assistant
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
|
|
async def test_pipeline_from_audio_stream(
|
|
hass: HomeAssistant, mock_stt_provider, snapshot: SnapshotAssertion
|
|
) -> None:
|
|
"""Test creating a pipeline from an audio stream."""
|
|
|
|
events = []
|
|
|
|
async def audio_data():
|
|
yield b"part1"
|
|
yield b"part2"
|
|
yield b""
|
|
|
|
await voice_assistant.async_pipeline_from_audio_stream(
|
|
hass,
|
|
events.append,
|
|
stt.SpeechMetadata(
|
|
language="",
|
|
format=stt.AudioFormats.WAV,
|
|
codec=stt.AudioCodecs.PCM,
|
|
bit_rate=stt.AudioBitRates.BITRATE_16,
|
|
sample_rate=stt.AudioSampleRates.SAMPLERATE_16000,
|
|
channel=stt.AudioChannels.CHANNEL_MONO,
|
|
),
|
|
audio_data(),
|
|
)
|
|
|
|
processed = []
|
|
for event in events:
|
|
as_dict = event.as_dict()
|
|
as_dict.pop("timestamp")
|
|
processed.append(as_dict)
|
|
|
|
assert processed == snapshot
|
|
assert mock_stt_provider.received == [b"part1", b"part2"]
|