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

[esphome] 100% voice assistant test coverage (#118334)

Browse Source
This commit is contained in:
Jesse Hills 2024-05-29 13:14:47 +12:00 committed by GitHub
parent 097ca3a0ae
commit 035e21ddbb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 165 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
homeassistant/components/esphome/voice_assistant.py
View File

@ -250,12 +250,12 @@ class VoiceAssistantPipeline:
await self._tts_done.wait()
_LOGGER.debug("Pipeline finished")
except PipelineNotFound:
except PipelineNotFound as e:
self.handle_event(
VoiceAssistantEventType.VOICE_ASSISTANT_ERROR,
{
"code": "pipeline not found",
"message": "Selected pipeline not found",
"code": e.code,
"message": e.message,
},
)
_LOGGER.warning("Pipeline not found")

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

@ -24,6 +24,7 @@ from homeassistant.components.assist_pipeline import (
PipelineStage,
)
from homeassistant.components.assist_pipeline.error import (
PipelineNotFound,
WakeWordDetectionAborted,
WakeWordDetectionError,
)
@ -353,6 +354,87 @@ async def test_udp_server_after_stopped(
await voice_assistant_udp_pipeline_v1.start_server()
async def test_events_converted_correctly(
hass: HomeAssistant,
voice_assistant_api_pipeline: VoiceAssistantAPIPipeline,
) -> None:
"""Test the pipeline events produce the correct data to send to the device."""
with patch(
"homeassistant.components.esphome.voice_assistant.VoiceAssistantPipeline._send_tts",
):
voice_assistant_api_pipeline._event_callback(
PipelineEvent(
type=PipelineEventType.STT_START,
data={},
)
)
voice_assistant_api_pipeline.handle_event.assert_called_with(
VoiceAssistantEventType.VOICE_ASSISTANT_STT_START, None
)
voice_assistant_api_pipeline._event_callback(
PipelineEvent(
type=PipelineEventType.STT_END,
data={"stt_output": {"text": "text"}},
)
)
voice_assistant_api_pipeline.handle_event.assert_called_with(
VoiceAssistantEventType.VOICE_ASSISTANT_STT_END, {"text": "text"}
)
voice_assistant_api_pipeline._event_callback(
PipelineEvent(
type=PipelineEventType.INTENT_START,
data={},
)
)
voice_assistant_api_pipeline.handle_event.assert_called_with(
VoiceAssistantEventType.VOICE_ASSISTANT_INTENT_START, None
)
voice_assistant_api_pipeline._event_callback(
PipelineEvent(
type=PipelineEventType.INTENT_END,
data={
"intent_output": {
"conversation_id": "conversation-id",
}
},
)
)
voice_assistant_api_pipeline.handle_event.assert_called_with(
VoiceAssistantEventType.VOICE_ASSISTANT_INTENT_END,
{"conversation_id": "conversation-id"},
)
voice_assistant_api_pipeline._event_callback(
PipelineEvent(
type=PipelineEventType.TTS_START,
data={"tts_input": "text"},
)
)
voice_assistant_api_pipeline.handle_event.assert_called_with(
VoiceAssistantEventType.VOICE_ASSISTANT_TTS_START, {"text": "text"}
)
voice_assistant_api_pipeline._event_callback(
PipelineEvent(
type=PipelineEventType.TTS_END,
data={"tts_output": {"url": "url", "media_id": "media-id"}},
)
)
voice_assistant_api_pipeline.handle_event.assert_called_with(
VoiceAssistantEventType.VOICE_ASSISTANT_TTS_END, {"url": "url"}
)
async def test_unknown_event_type(
hass: HomeAssistant,
voice_assistant_api_pipeline: VoiceAssistantAPIPipeline,
@ -780,3 +862,83 @@ async def test_timer_events(
3723,
True,
)
async def test_unknown_timer_event(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: Callable[
[APIClient, list[EntityInfo], list[UserService], list[EntityState]],
Awaitable[MockESPHomeDevice],
],
) -> None:
"""Test that unknown (new) timer event types do not result in api calls."""
mock_device: MockESPHomeDevice = await mock_esphome_device(
mock_client=mock_client,
entity_info=[],
user_service=[],
states=[],
device_info={
"voice_assistant_feature_flags": VoiceAssistantFeature.VOICE_ASSISTANT
| VoiceAssistantFeature.TIMERS
},
)
dev_reg = dr.async_get(hass)
dev = dev_reg.async_get_device(
connections={(dr.CONNECTION_NETWORK_MAC, mock_device.entry.unique_id)}
)
with patch(
"homeassistant.components.esphome.voice_assistant._TIMER_EVENT_TYPES.from_hass",
side_effect=KeyError,
):
await intent_helper.async_handle(
hass,
"test",
intent_helper.INTENT_START_TIMER,
{
"name": {"value": "test timer"},
"hours": {"value": 1},
"minutes": {"value": 2},
"seconds": {"value": 3},
},
device_id=dev.id,
)
mock_client.send_voice_assistant_timer_event.assert_not_called()
async def test_invalid_pipeline_id(
hass: HomeAssistant,
voice_assistant_api_pipeline: VoiceAssistantAPIPipeline,
) -> None:
"""Test that the pipeline is set to start with Wake word."""
invalid_pipeline_id = "invalid-pipeline-id"
async def async_pipeline_from_audio_stream(*args, **kwargs):
raise PipelineNotFound(
"pipeline_not_found", f"Pipeline {invalid_pipeline_id} not found"
)
with patch(
"homeassistant.components.esphome.voice_assistant.async_pipeline_from_audio_stream",
new=async_pipeline_from_audio_stream,
):
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"] == f"Pipeline {invalid_pipeline_id} not found"
voice_assistant_api_pipeline.handle_event = handle_event
await voice_assistant_api_pipeline.run_pipeline(
device_id="mock-device-id",
conversation_id=None,
flags=2,
)