More test coverage

homeassistant/components/wyoming/tts.py

@@ -4,7 +4,7 @@
 import logging
 import wave
 
-from wyoming.audio import AudioChunk, AudioChunkConverter, AudioStop
+from wyoming.audio import AudioChunk, AudioStop
 from wyoming.client import AsyncTcpClient
 from wyoming.tts import Synthesize, SynthesizeVoice
 
@@ -147,27 +147,4 @@ async def async_get_tts_audio(self, message, language, options):
         except (OSError, WyomingError):
             return (None, None)
 
-        if options.get(tts.ATTR_AUDIO_OUTPUT) != "raw":
-            return ("wav", data)
-
-        # Raw output (convert to 16Khz, 16-bit mono)
-        with io.BytesIO(data) as wav_io:
-            wav_reader: wave.Wave_read = wave.open(wav_io, "rb")
-            raw_data = (
-                AudioChunkConverter(
-                    rate=16000,
-                    width=2,
-                    channels=1,
-                )
-                .convert(
-                    AudioChunk(
-                        audio=wav_reader.readframes(wav_reader.getnframes()),
-                        rate=wav_reader.getframerate(),
-                        width=wav_reader.getsampwidth(),
-                        channels=wav_reader.getnchannels(),
-                    )
-                )
-                .audio
-            )
-
-        return ("raw", raw_data)
+        return ("wav", data)

tests/components/assist_pipeline/test_init.py

@@ -9,7 +9,7 @@
 import pytest
 from syrupy.assertion import SnapshotAssertion
 
-from homeassistant.components import assist_pipeline, stt
+from homeassistant.components import assist_pipeline, stt, tts
 from homeassistant.components.assist_pipeline.const import (
     CONF_DEBUG_RECORDING_DIR,
     DOMAIN,
@@ -660,3 +660,42 @@ def event_callback(event):
     assert run_1 == run_1
     assert run_1 != run_2
     assert run_1 != 1234
+
+
+async def test_tts_audio_output(
+    hass: HomeAssistant,
+    mock_stt_provider: MockSttProvider,
+    init_components,
+    pipeline_data: assist_pipeline.pipeline.PipelineData,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test using tts_audio_output with wav sets options correctly."""
+
+    def event_callback(event):
+        pass
+
+    pipeline_store = pipeline_data.pipeline_store
+    pipeline_id = pipeline_store.async_get_preferred_item()
+    pipeline = assist_pipeline.pipeline.async_get_pipeline(hass, pipeline_id)
+
+    pipeline_input = assist_pipeline.pipeline.PipelineInput(
+        tts_input="This is a test.",
+        conversation_id=None,
+        device_id=None,
+        run=assist_pipeline.pipeline.PipelineRun(
+            hass,
+            context=Context(),
+            pipeline=pipeline,
+            start_stage=assist_pipeline.PipelineStage.TTS,
+            end_stage=assist_pipeline.PipelineStage.TTS,
+            event_callback=event_callback,
+            tts_audio_output="wav",
+        ),
+    )
+    await pipeline_input.validate()
+
+    # Verify TTS audio settings
+    assert pipeline_input.run.tts_options is not None
+    assert pipeline_input.run.tts_options.get(tts.ATTR_PREFERRED_FORMAT) == "wav"
+    assert pipeline_input.run.tts_options.get(tts.ATTR_PREFERRED_SAMPLE_RATE) == 16000
+    assert pipeline_input.run.tts_options.get(tts.ATTR_PREFERRED_SAMPLE_CHANNELS) == 1

tests/components/esphome/test_voice_assistant.py

@@ -371,6 +371,39 @@ async def test_send_tts(
         voice_assistant_udp_server_v2.transport.sendto.assert_called()
 
 
+async def test_send_tts_wrong_sample_rate(
+    hass: HomeAssistant,
+    voice_assistant_udp_server_v2: VoiceAssistantUDPServer,
+) -> None:
+    """Test the UDP server calls sendto to transmit audio data to device."""
+    with io.BytesIO() as wav_io:
+        with wave.open(wav_io, "wb") as wav_file:
+            wav_file.setframerate(22050)  # should be 16000
+            wav_file.setsampwidth(2)
+            wav_file.setnchannels(1)
+            wav_file.writeframes(bytes(_ONE_SECOND))
+
+        wav_bytes = wav_io.getvalue()
+
+    with patch(
+        "homeassistant.components.esphome.voice_assistant.tts.async_get_media_source_audio",
+        return_value=("wav", wav_bytes),
+    ), pytest.raises(ValueError):
+        voice_assistant_udp_server_v2.transport = Mock(spec=asyncio.DatagramTransport)
+
+        voice_assistant_udp_server_v2._event_callback(
+            PipelineEvent(
+                type=PipelineEventType.TTS_END,
+                data={
+                    "tts_output": {"media_id": _TEST_MEDIA_ID, "url": _TEST_OUTPUT_URL}
+                },
+            )
+        )
+
+        assert voice_assistant_udp_server_v2._tts_task is not None
+        await voice_assistant_udp_server_v2._tts_task  # raises ValueError
+
+
 async def test_send_tts_wrong_format(
     hass: HomeAssistant,
     voice_assistant_udp_server_v2: VoiceAssistantUDPServer,

tests/components/tts/test_init.py

@@ -6,7 +6,7 @@
 
 import pytest
 
-from homeassistant.components import tts
+from homeassistant.components import ffmpeg, tts
 from homeassistant.components.media_player import (
     ATTR_MEDIA_ANNOUNCE,
     ATTR_MEDIA_CONTENT_ID,
@@ -1759,3 +1759,12 @@ async def test_ws_list_voices(
             {"voice_id": "fran_drescher", "name": "Fran Drescher"},
         ]
     }
+
+
+async def test_async_convert_audio_error(hass: HomeAssistant) -> None:
+    """Test that ffmpeg failing during audio conversion will raise an error."""
+    assert await async_setup_component(hass, ffmpeg.DOMAIN, {})
+
+    with pytest.raises(RuntimeError):
+        # Simulate a bad WAV file
+        await tts.async_convert_audio(hass, "wav", bytes(0), "mp3")