From 7de1d800226d383a947266b507346dfaba6c3666 Mon Sep 17 00:00:00 2001 From: Paulus Schoutsen Date: Wed, 9 Sep 2026 00:52:25 -0400 Subject: [PATCH] Migrate tests away from tts.async_get_media_source_audio (#181700) Co-authored-by: Claude Opus 5 --- tests/components/picotts/test_tts.py | 28 +++----- tests/components/tts/test_init.py | 100 ++++++++++++-------------- tests/components/wyoming/test_tts.py | 101 +++++++++++---------------- 3 files changed, 94 insertions(+), 135 deletions(-) diff --git a/tests/components/picotts/test_tts.py b/tests/components/picotts/test_tts.py index 1df39807d159..35b889c2253d 100644 --- a/tests/components/picotts/test_tts.py +++ b/tests/components/picotts/test_tts.py @@ -23,6 +23,13 @@ from tests.components.tts.common import retrieve_media from tests.typing import ClientSessionGenerator +async def get_tts_audio(hass: HomeAssistant) -> bytes: + """Get TTS audio from the Pico TTS entity.""" + stream = tts.async_create_stream(hass, "tts.pico_tts_en_us", "en-US") + stream.async_set_message("Hello world") + return b"".join([chunk async for chunk in stream.async_stream_result()]) + + def get_empty_wav() -> bytes: """Get bytes for empty WAV file.""" with io.BytesIO() as wav_io: @@ -142,12 +149,7 @@ async def test_get_tts_audio_subprocess_error( ), pytest.raises(HomeAssistantError) as exc_info, ): - await tts.async_get_media_source_audio( - hass, - tts.generate_media_source_id( - hass, "Hello world", "tts.pico_tts_en_us", "en-US" - ), - ) + await get_tts_audio(hass) assert exc_info.value.translation_key == "returncode_error" assert exc_info.value.translation_placeholders == {"returncode": "1"} @@ -165,12 +167,7 @@ async def test_get_tts_audio_timeout( ), pytest.raises(HomeAssistantError) as exc_info, ): - await tts.async_get_media_source_audio( - hass, - tts.generate_media_source_id( - hass, "Hello world", "tts.pico_tts_en_us", "en-US" - ), - ) + await get_tts_audio(hass) assert exc_info.value.translation_key == "timeout_error" @@ -190,11 +187,6 @@ async def test_get_tts_audio_file_read_error( ), pytest.raises(HomeAssistantError) as exc_info, ): - await tts.async_get_media_source_audio( - hass, - tts.generate_media_source_id( - hass, "Hello world", "tts.pico_tts_en_us", "en-US" - ), - ) + await get_tts_audio(hass) assert exc_info.value.translation_key == "file_read_error" diff --git a/tests/components/tts/test_init.py b/tests/components/tts/test_init.py index f9ae16712372..fba4caa26e65 100644 --- a/tests/components/tts/test_init.py +++ b/tests/components/tts/test_init.py @@ -50,6 +50,11 @@ from tests.typing import ClientSessionGenerator, WebSocketGenerator ORIG_WRITE_TAGS = tts.SpeechManager.write_tags +async def get_stream_data(stream: tts.ResultStream) -> bytes: + """Get all data of a result stream.""" + return b"".join([chunk async for chunk in stream.async_stream_result()]) + + async def test_config_entry_unload( hass: HomeAssistant, hass_client: ClientSessionGenerator, @@ -835,11 +840,10 @@ async def test_service_receive_voice( assert req.status == HTTPStatus.OK assert await req.read() == tts_data - extension, data = await tts.async_get_media_source_audio( - hass, calls[0].data[ATTR_MEDIA_CONTENT_ID] - ) - assert extension == "mp3" - assert tts_data == data + stream = tts.async_get_stream(hass, url.rsplit("/", 1)[-1]) + assert stream is not None + assert stream.extension == "mp3" + assert tts_data == b"".join([chunk async for chunk in stream.async_stream_result()]) @pytest.mark.parametrize( @@ -1441,25 +1445,21 @@ async def test_legacy_fetching_in_async( await mock_setup(hass, ProviderWithAsyncFetching(DEFAULT_LANG)) - # Test async_get_media_source_audio - media_source_id = tts.generate_media_source_id( - hass, - "test message", - "test", - "en_US", - cache=None, - ) + def create_stream(message: str) -> tts.ResultStream: + stream = tts.async_create_stream(hass, "test", "en_US") + stream.async_set_message(message) + return stream - task = hass.async_create_task( - tts.async_get_media_source_audio(hass, media_source_id) - ) - task2 = hass.async_create_task( - tts.async_get_media_source_audio(hass, media_source_id) - ) + # Streams for the same message share a single fetch + stream = create_stream("test message") + stream2 = create_stream("test message") + stream3 = create_stream("test message") + + task = hass.async_create_task(get_stream_data(stream)) + task2 = hass.async_create_task(get_stream_data(stream2)) - url = await get_media_source_url(hass, media_source_id) client = await hass_client() - client_get_task = hass.async_create_task(client.get(url)) + client_get_task = hass.async_create_task(client.get(stream3.url)) # Make sure that tasks are waiting for our future to resolve done, pending = await asyncio.wait((task, task2, client_get_task), timeout=0.1) @@ -1468,28 +1468,23 @@ async def test_legacy_fetching_in_async( tts_audio.set_result(b"test") - assert await task == ("mp3", b"test") - assert await task2 == ("mp3", b"test") + assert stream.extension == "mp3" + assert await task == b"test" + assert await task2 == b"test" req = await client_get_task assert req.status == HTTPStatus.OK assert await req.read() == b"test" # Test error is not cached - media_source_id = tts.generate_media_source_id( - hass, "test message 2", "test", "en_US", None, None - ) tts_audio = asyncio.Future() tts_audio.set_exception(HomeAssistantError("test error")) with pytest.raises(HomeAssistantError): - assert await tts.async_get_media_source_audio(hass, media_source_id) + await get_stream_data(create_stream("test message 2")) tts_audio = asyncio.Future() tts_audio.set_result(b"test 2") - assert await tts.async_get_media_source_audio(hass, media_source_id) == ( - "mp3", - b"test 2", - ) + assert await get_stream_data(create_stream("test message 2")) == b"test 2" async def test_fetching_in_async( @@ -1508,25 +1503,21 @@ async def test_fetching_in_async( await mock_config_entry_setup(hass, EntityWithAsyncFetching(DEFAULT_LANG)) - # Test async_get_media_source_audio - media_source_id = tts.generate_media_source_id( - hass, - "test message", - "tts.test", - "en_US", - cache=None, - ) + def create_stream(message: str) -> tts.ResultStream: + stream = tts.async_create_stream(hass, "tts.test", "en_US") + stream.async_set_message(message) + return stream - task = hass.async_create_task( - tts.async_get_media_source_audio(hass, media_source_id) - ) - task2 = hass.async_create_task( - tts.async_get_media_source_audio(hass, media_source_id) - ) + # Streams for the same message share a single fetch + stream = create_stream("test message") + stream2 = create_stream("test message") + stream3 = create_stream("test message") + + task = hass.async_create_task(get_stream_data(stream)) + task2 = hass.async_create_task(get_stream_data(stream2)) - url = await get_media_source_url(hass, media_source_id) client = await hass_client() - client_get_task = hass.async_create_task(client.get(url)) + client_get_task = hass.async_create_task(client.get(stream3.url)) # Make sure that tasks are waiting for our future to resolve done, pending = await asyncio.wait((task, task2, client_get_task), timeout=0.1) @@ -1535,28 +1526,23 @@ async def test_fetching_in_async( tts_audio.set_result(b"test") - assert await task == ("mp3", b"test") - assert await task2 == ("mp3", b"test") + assert stream.extension == "mp3" + assert await task == b"test" + assert await task2 == b"test" req = await client_get_task assert req.status == HTTPStatus.OK assert await req.read() == b"test" # Test error is not cached - media_source_id = tts.generate_media_source_id( - hass, "test message 2", "tts.test", "en_US", None, None - ) tts_audio = asyncio.Future() tts_audio.set_exception(HomeAssistantError("test error")) with pytest.raises(HomeAssistantError): - assert await tts.async_get_media_source_audio(hass, media_source_id) + await get_stream_data(create_stream("test message 2")) tts_audio = asyncio.Future() tts_audio.set_result(b"test 2") - assert await tts.async_get_media_source_audio(hass, media_source_id) == ( - "mp3", - b"test 2", - ) + assert await get_stream_data(create_stream("test message 2")) == b"test 2" @pytest.mark.parametrize( diff --git a/tests/components/wyoming/test_tts.py b/tests/components/wyoming/test_tts.py index 6a42b8190dc1..e66ab85ff555 100644 --- a/tests/components/wyoming/test_tts.py +++ b/tests/components/wyoming/test_tts.py @@ -132,18 +132,13 @@ async def test_get_tts_audio( "homeassistant.components.wyoming.tts.AsyncTcpClient", MockAsyncTcpClient(audio_events), ) as mock_client: - extension, data = await tts.async_get_media_source_audio( - hass, - tts.generate_media_source_id( - hass, - "Hello world", - "tts.test_tts", - "en-US", - options={tts.ATTR_PREFERRED_FORMAT: "wav"}, - ), + stream = tts.async_create_stream( + hass, "tts.test_tts", "en-US", options={tts.ATTR_PREFERRED_FORMAT: "wav"} ) + stream.async_set_message("Hello world") + data = b"".join([chunk async for chunk in stream.async_stream_result()]) - assert extension == "wav" + assert stream.extension == "wav" assert data is not None with io.BytesIO(data) as wav_io, wave.open(wav_io, "rb") as wav_file: assert wav_file.getframerate() == 16000 @@ -173,22 +168,20 @@ async def test_get_tts_audio_different_formats( "homeassistant.components.wyoming.tts.AsyncTcpClient", MockAsyncTcpClient(audio_events), ) as mock_client: - extension, data = await tts.async_get_media_source_audio( + stream = tts.async_create_stream( hass, - tts.generate_media_source_id( - hass, - "Hello world", - "tts.test_tts", - "en-US", - options={ - tts.ATTR_PREFERRED_FORMAT: "wav", - tts.ATTR_PREFERRED_SAMPLE_RATE: 48000, - tts.ATTR_PREFERRED_SAMPLE_CHANNELS: 2, - }, - ), + "tts.test_tts", + "en-US", + options={ + tts.ATTR_PREFERRED_FORMAT: "wav", + tts.ATTR_PREFERRED_SAMPLE_RATE: 48000, + tts.ATTR_PREFERRED_SAMPLE_CHANNELS: 2, + }, ) + stream.async_set_message("Hello world") + data = b"".join([chunk async for chunk in stream.async_stream_result()]) - assert extension == "wav" + assert stream.extension == "wav" assert data is not None with io.BytesIO(data) as wav_io, wave.open(wav_io, "rb") as wav_file: assert wav_file.getframerate() == 48000 @@ -208,17 +201,11 @@ async def test_get_tts_audio_different_formats( "homeassistant.components.wyoming.tts.AsyncTcpClient", MockAsyncTcpClient(audio_events), ) as mock_client: - extension, data = await tts.async_get_media_source_audio( - hass, - tts.generate_media_source_id( - hass, - "Hello world", - "tts.test_tts", - "en-US", - ), - ) + stream = tts.async_create_stream(hass, "tts.test_tts", "en-US") + stream.async_set_message("Hello world") + data = b"".join([chunk async for chunk in stream.async_stream_result()]) - assert extension == "mp3" + assert stream.extension == "mp3" assert b"ID3" in data assert mock_client.written == snapshot @@ -251,22 +238,18 @@ async def test_get_tts_audio_audio_oserror( mock_client = MockAsyncTcpClient(audio_events) + stream = tts.async_create_stream(hass, "tts.test_tts", "en-US") with ( patch( "homeassistant.components.wyoming.tts.AsyncTcpClient", mock_client, ), patch.object(mock_client, "read_event", side_effect=OSError("Boom!")), - pytest.raises( - HomeAssistantError, - ), ): - await tts.async_get_media_source_audio( - hass, - tts.generate_media_source_id( - hass, "Hello world", "tts.test_tts", hass.config.language - ), - ) + stream.async_set_message("Hello world") + with pytest.raises(HomeAssistantError): + async for _chunk in stream.async_stream_result(): + pass @pytest.mark.usefixtures("init_wyoming_tts") @@ -285,17 +268,15 @@ async def test_get_tts_audio_error_event( hass: HomeAssistant, error_code: str | None, expected_message: str ) -> None: """Test that an error event from the service is reported.""" - with ( - patch( - "homeassistant.components.wyoming.tts.AsyncTcpClient", - MockAsyncTcpClient([Error(text="Boom!", code=error_code).event()]), - ), - pytest.raises(HomeAssistantError, match=re.escape(expected_message)), + stream = tts.async_create_stream(hass, "tts.test_tts", "en-US") + with patch( + "homeassistant.components.wyoming.tts.AsyncTcpClient", + MockAsyncTcpClient([Error(text="Boom!", code=error_code).event()]), ): - await tts.async_get_media_source_audio( - hass, - tts.generate_media_source_id(hass, "Hello world", "tts.test_tts", "en-US"), - ) + stream.async_set_message("Hello world") + with pytest.raises(HomeAssistantError, match=re.escape(expected_message)): + async for _chunk in stream.async_stream_result(): + pass @pytest.mark.usefixtures("init_wyoming_streaming_tts") @@ -339,16 +320,16 @@ async def test_voice_speaker( "homeassistant.components.wyoming.tts.AsyncTcpClient", MockAsyncTcpClient(audio_events), ) as mock_client: - await tts.async_get_media_source_audio( + stream = tts.async_create_stream( hass, - tts.generate_media_source_id( - hass, - "Hello world", - "tts.test_tts", - "en-US", - options={tts.ATTR_VOICE: "voice1", wyoming.ATTR_SPEAKER: "speaker1"}, - ), + "tts.test_tts", + "en-US", + options={tts.ATTR_VOICE: "voice1", wyoming.ATTR_SPEAKER: "speaker1"}, ) + stream.async_set_message("Hello world") + async for _chunk in stream.async_stream_result(): + pass + assert mock_client.written == snapshot