starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 01:11:51 -04:00
Code Issues Packages Projects Releases Wiki Activity

Migrate tests away from tts.async_get_media_source_audio (#181700)

Browse Source 
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Paulus Schoutsen
2026-09-09 06:52:25 +02:00
committed by GitHub
co-authored by Claude Opus 5
parent 87e8b92773
commit 7de1d80022
3 changed files with 94 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/components/picotts/test_tts.py
+10 -18
View File
@@ -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"
tests/components/tts/test_init.py
+43 -57
View File
@@ -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(
tests/components/wyoming/test_tts.py
+41 -60
View File
@@ -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