jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-11-13 13:00:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add audio_settings for pipeline from ESPHome device (#100894)

Browse Source 
* Add audio_settings for pipeline from ESPHome device

* ruff fixes

* Bump aioesphomeapi 17.0.0

* Mypy

* Fix tests

---------

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
Jesse Hills
2023-09-27 10:27:26 +13:00
committed by GitHub
parent f899e5159b
commit 4c21aa18db
6 changed files with 50 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
tests/components/esphome/test_voice_assistant.py
View File

@@ -10,7 +10,6 @@ import pytest
from homeassistant.components.assist_pipeline import (
PipelineEvent,
PipelineEventType,
PipelineNotFound,
PipelineStage,
)
from homeassistant.components.assist_pipeline.error import WakeWordDetectionError
@@ -370,6 +369,8 @@ async def test_wake_word(
with patch(
"homeassistant.components.esphome.voice_assistant.async_pipeline_from_audio_stream",
new=async_pipeline_from_audio_stream,
), patch(
"asyncio.Event.wait" # TTS wait event
):
voice_assistant_udp_server_v2.transport = Mock()
@@ -377,7 +378,6 @@ async def test_wake_word(
device_id="mock-device-id",
conversation_id=None,
flags=2,
pipeline_timeout=1,
)
@@ -410,38 +410,4 @@ async def test_wake_word_exception(
device_id="mock-device-id",
conversation_id=None,
flags=2,
pipeline_timeout=1,
)
async def test_pipeline_timeout(
hass: HomeAssistant,
voice_assistant_udp_server_v2: VoiceAssistantUDPServer,
) -> None:
"""Test that the pipeline is set to start with Wake word."""
async def async_pipeline_from_audio_stream(*args, **kwargs):
raise PipelineNotFound("not-found", "Pipeline not found")
with patch(
"homeassistant.components.esphome.voice_assistant.async_pipeline_from_audio_stream",
new=async_pipeline_from_audio_stream,
):
voice_assistant_udp_server_v2.transport = Mock()
def handle_event(
event_type: VoiceAssistantEventType, data: dict[str, str] | None
) -> None:
if event_type == VoiceAssistantEventType.VOICE_ASSISTANT_ERROR:
assert data is not None
assert data["code"] == "pipeline not found"
assert data["message"] == "Selected pipeline not found"
voice_assistant_udp_server_v2.handle_event = handle_event
await voice_assistant_udp_server_v2.run_pipeline(
device_id="mock-device-id",
conversation_id=None,
flags=2,
pipeline_timeout=1,
)