1140 lines
37 KiB
Python
1140 lines
37 KiB
Python
"""The tests for the TTS component."""
|
|
import asyncio
|
|
from http import HTTPStatus
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components import media_source, tts
|
|
from homeassistant.components.media_player import (
|
|
ATTR_MEDIA_ANNOUNCE,
|
|
ATTR_MEDIA_CONTENT_ID,
|
|
ATTR_MEDIA_CONTENT_TYPE,
|
|
DOMAIN as DOMAIN_MP,
|
|
SERVICE_PLAY_MEDIA,
|
|
MediaType,
|
|
)
|
|
from homeassistant.components.media_source import Unresolvable
|
|
from homeassistant.components.tts.legacy import _valid_base_url
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.setup import async_setup_component
|
|
from homeassistant.util.network import normalize_url
|
|
|
|
from .common import MockProvider, MockTTS
|
|
|
|
from tests.common import (
|
|
MockModule,
|
|
assert_setup_component,
|
|
async_mock_service,
|
|
mock_integration,
|
|
mock_platform,
|
|
)
|
|
from tests.typing import ClientSessionGenerator, WebSocketGenerator
|
|
|
|
ORIG_WRITE_TAGS = tts.SpeechManager.write_tags
|
|
|
|
|
|
async def get_media_source_url(hass: HomeAssistant, media_content_id: str) -> str:
|
|
"""Get the media source url."""
|
|
if media_source.DOMAIN not in hass.config.components:
|
|
assert await async_setup_component(hass, media_source.DOMAIN, {})
|
|
|
|
resolved = await media_source.async_resolve_media(hass, media_content_id, None)
|
|
return resolved.url
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_provider() -> MockProvider:
|
|
"""Test TTS provider."""
|
|
return MockProvider("en_US")
|
|
|
|
|
|
@pytest.fixture
|
|
async def setup_tts(hass: HomeAssistant, mock_tts: None) -> None:
|
|
"""Mock TTS."""
|
|
assert await async_setup_component(hass, tts.DOMAIN, {"tts": {"platform": "test"}})
|
|
|
|
|
|
async def test_setup_component(hass: HomeAssistant, setup_tts) -> None:
|
|
"""Set up a TTS platform with defaults."""
|
|
assert hass.services.has_service(tts.DOMAIN, "test_say")
|
|
assert hass.services.has_service(tts.DOMAIN, "clear_cache")
|
|
assert f"{tts.DOMAIN}.test" in hass.config.components
|
|
|
|
|
|
async def test_setup_component_no_access_cache_folder(
|
|
hass: HomeAssistant, mock_init_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform with defaults."""
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
mock_init_cache_dir.side_effect = OSError(2, "No access")
|
|
assert not await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
assert not hass.services.has_service(tts.DOMAIN, "test_say")
|
|
assert not hass.services.has_service(tts.DOMAIN, "clear_cache")
|
|
|
|
|
|
async def test_setup_component_and_test_service(
|
|
hass: HomeAssistant, empty_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and call service."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
},
|
|
blocking=True,
|
|
)
|
|
|
|
assert len(calls) == 1
|
|
assert calls[0].data[ATTR_MEDIA_ANNOUNCE] is True
|
|
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MediaType.MUSIC
|
|
assert (
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
== "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_and_test_service_with_config_language(
|
|
hass: HomeAssistant, empty_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and call service."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
# Language de is matched with de_DE
|
|
config = {tts.DOMAIN: {"platform": "test", "language": "de"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 1
|
|
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MediaType.MUSIC
|
|
assert (
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
== "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_-_test.mp3"
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_-_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_and_test_service_with_config_language_special(
|
|
hass: HomeAssistant, empty_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and call service with extend language."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test", "language": "en_US"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 1
|
|
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MediaType.MUSIC
|
|
assert (
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
== "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_and_test_service_with_wrong_conf_language(
|
|
hass: HomeAssistant, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and call service with wrong config."""
|
|
config = {tts.DOMAIN: {"platform": "test", "language": "ru"}}
|
|
|
|
with assert_setup_component(0, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
|
|
async def test_setup_component_and_test_service_with_service_language(
|
|
hass: HomeAssistant, empty_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and call service."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
# Language de is matched to de_DE
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
tts.ATTR_LANGUAGE: "de",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 1
|
|
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MediaType.MUSIC
|
|
assert (
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
== "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_-_test.mp3"
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_-_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_test_service_with_wrong_service_language(
|
|
hass: HomeAssistant, empty_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and call service."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
with pytest.raises(HomeAssistantError):
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
tts.ATTR_LANGUAGE: "lang",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 0
|
|
assert not (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_lang_-_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_and_test_service_with_service_options(
|
|
hass: HomeAssistant, empty_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and call service with options."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
# Language de is matched with de_DE
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
tts.ATTR_LANGUAGE: "de",
|
|
tts.ATTR_OPTIONS: {"voice": "alex", "age": 5},
|
|
},
|
|
blocking=True,
|
|
)
|
|
opt_hash = tts._hash_options({"voice": "alex", "age": 5})
|
|
|
|
assert len(calls) == 1
|
|
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MediaType.MUSIC
|
|
assert (
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
== f"/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_{opt_hash}_test.mp3"
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert (
|
|
empty_cache_dir
|
|
/ f"42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_{opt_hash}_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_and_test_with_service_options_def(
|
|
hass: HomeAssistant, empty_cache_dir
|
|
) -> None:
|
|
"""Set up a TTS platform and call service with default options."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
class MockProviderWithDefaults(MockProvider):
|
|
"""Mock provider with default options."""
|
|
|
|
@property
|
|
def default_options(self):
|
|
return {"voice": "alex"}
|
|
|
|
mock_integration(hass, MockModule(domain="test"))
|
|
mock_platform(hass, "test.tts", MockTTS(MockProviderWithDefaults))
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
# Language de is matched with de_DE
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
tts.ATTR_LANGUAGE: "de",
|
|
},
|
|
blocking=True,
|
|
)
|
|
opt_hash = tts._hash_options({"voice": "alex"})
|
|
|
|
assert len(calls) == 1
|
|
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MediaType.MUSIC
|
|
assert (
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
== f"/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_{opt_hash}_test.mp3"
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert (
|
|
empty_cache_dir
|
|
/ f"42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_{opt_hash}_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_and_test_with_service_options_def_2(
|
|
hass: HomeAssistant, empty_cache_dir
|
|
) -> None:
|
|
"""Set up a TTS platform and call service with default options.
|
|
|
|
This tests merging default and user provided options.
|
|
"""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
class MockProviderWithDefaults(MockProvider):
|
|
"""Mock provider with default options."""
|
|
|
|
@property
|
|
def default_options(self):
|
|
return {"voice": "alex"}
|
|
|
|
mock_integration(hass, MockModule(domain="test"))
|
|
mock_platform(hass, "test.tts", MockTTS(MockProviderWithDefaults))
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
# Language de is matched with de_DE
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
tts.ATTR_LANGUAGE: "de",
|
|
tts.ATTR_OPTIONS: {"age": 5},
|
|
},
|
|
blocking=True,
|
|
)
|
|
opt_hash = tts._hash_options({"voice": "alex", "age": 5})
|
|
|
|
assert len(calls) == 1
|
|
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MediaType.MUSIC
|
|
assert (
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
== f"/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_{opt_hash}_test.mp3"
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert (
|
|
empty_cache_dir
|
|
/ f"42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_{opt_hash}_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_and_test_service_with_service_options_wrong(
|
|
hass: HomeAssistant, empty_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and call service with wrong options."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
# Language de is matched with de_DE
|
|
with pytest.raises(HomeAssistantError):
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
tts.ATTR_LANGUAGE: "de",
|
|
tts.ATTR_OPTIONS: {"speed": 1},
|
|
},
|
|
blocking=True,
|
|
)
|
|
opt_hash = tts._hash_options({"speed": 1})
|
|
|
|
assert len(calls) == 0
|
|
await hass.async_block_till_done()
|
|
assert not (
|
|
empty_cache_dir
|
|
/ f"42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_{opt_hash}_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_and_test_service_with_base_url_set(
|
|
hass: HomeAssistant, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform with ``base_url`` set and call service."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test", "base_url": "http://fnord"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 1
|
|
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MediaType.MUSIC
|
|
assert (
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
== "http://fnord"
|
|
"/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491"
|
|
"_en-us_-_test.mp3"
|
|
)
|
|
|
|
|
|
async def test_setup_component_and_test_service_clear_cache(
|
|
hass: HomeAssistant, empty_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and call service clear cache."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
},
|
|
blocking=True,
|
|
)
|
|
# To make sure the file is persisted
|
|
assert len(calls) == 1
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
await hass.async_block_till_done()
|
|
assert (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
).is_file()
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN, tts.SERVICE_CLEAR_CACHE, {}, blocking=True
|
|
)
|
|
|
|
assert not (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_and_test_service_with_receive_voice(
|
|
hass: HomeAssistant,
|
|
mock_provider: MockProvider,
|
|
hass_client: ClientSessionGenerator,
|
|
mock_tts,
|
|
) -> None:
|
|
"""Set up a TTS platform and call service and receive voice."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
message = "There is someone at the door."
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: message,
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 1
|
|
|
|
url = await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
client = await hass_client()
|
|
req = await client.get(url)
|
|
# Language en is matched with en_US
|
|
_, tts_data = mock_provider.get_tts_audio("bla", "en")
|
|
assert tts_data is not None
|
|
tts_data = tts.SpeechManager.write_tags(
|
|
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3",
|
|
tts_data,
|
|
mock_provider,
|
|
message,
|
|
"en",
|
|
None,
|
|
)
|
|
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
|
|
|
|
|
|
async def test_setup_component_and_test_service_with_receive_voice_german(
|
|
hass: HomeAssistant,
|
|
mock_provider: MockProvider,
|
|
hass_client: ClientSessionGenerator,
|
|
mock_tts,
|
|
) -> None:
|
|
"""Set up a TTS platform and call service and receive voice."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
# Language de is matched with de_DE
|
|
config = {tts.DOMAIN: {"platform": "test", "language": "de"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 1
|
|
url = await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
client = await hass_client()
|
|
req = await client.get(url)
|
|
_, tts_data = mock_provider.get_tts_audio("bla", "de")
|
|
assert tts_data is not None
|
|
tts_data = tts.SpeechManager.write_tags(
|
|
"42f18378fd4393d18c8dd11d03fa9563c1e54491_de-de_-_test.mp3",
|
|
tts_data,
|
|
mock_provider,
|
|
"There is someone at the door.",
|
|
"de",
|
|
None,
|
|
)
|
|
assert req.status == HTTPStatus.OK
|
|
assert await req.read() == tts_data
|
|
|
|
|
|
async def test_setup_component_and_web_view_wrong_file(
|
|
hass: HomeAssistant, hass_client: ClientSessionGenerator, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and receive wrong file from web."""
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
client = await hass_client()
|
|
|
|
url = "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
|
|
req = await client.get(url)
|
|
assert req.status == HTTPStatus.NOT_FOUND
|
|
|
|
|
|
async def test_setup_component_and_web_view_wrong_filename(
|
|
hass: HomeAssistant, hass_client: ClientSessionGenerator, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and receive wrong filename from web."""
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
client = await hass_client()
|
|
|
|
url = "/api/tts_proxy/265944dsk32c1b2a621be5930510bb2cd_en-us_-_test.mp3"
|
|
|
|
req = await client.get(url)
|
|
assert req.status == HTTPStatus.NOT_FOUND
|
|
|
|
|
|
async def test_setup_component_test_without_cache(
|
|
hass: HomeAssistant, empty_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform without cache."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test", "cache": False}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 1
|
|
await hass.async_block_till_done()
|
|
assert not (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_test_with_cache_call_service_without_cache(
|
|
hass: HomeAssistant, empty_cache_dir, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform with cache and call service without cache."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test", "cache": True}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
tts.ATTR_CACHE: False,
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 1
|
|
await hass.async_block_till_done()
|
|
assert not (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
).is_file()
|
|
|
|
|
|
async def test_setup_component_test_with_cache_dir(
|
|
hass: HomeAssistant, empty_cache_dir, mock_provider: MockProvider
|
|
) -> None:
|
|
"""Set up a TTS platform with cache and call service without cache."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
# Language en is matched with en_US
|
|
_, tts_data = mock_provider.get_tts_audio("bla", "en")
|
|
assert tts_data is not None
|
|
cache_file = (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
)
|
|
|
|
with open(cache_file, "wb") as voice_file:
|
|
voice_file.write(tts_data)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test", "cache": True}}
|
|
|
|
class MockProviderBoom(MockProvider):
|
|
"""Mock provider that blows up."""
|
|
|
|
def get_tts_audio(
|
|
self, message: str, language: str, options: dict[str, Any] | None = None
|
|
) -> tts.TtsAudioType:
|
|
"""Load TTS dat."""
|
|
# This should not be called, data should be fetched from cache
|
|
raise Exception("Boom!") # pylint: disable=broad-exception-raised
|
|
|
|
mock_integration(hass, MockModule(domain="test"))
|
|
mock_platform(hass, "test.tts", MockTTS(MockProviderBoom))
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 1
|
|
assert (
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
== "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
|
|
async def test_setup_component_test_with_error_on_get_tts(hass: HomeAssistant) -> None:
|
|
"""Set up a TTS platform with wrong get_tts_audio."""
|
|
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
class MockProviderEmpty(MockProvider):
|
|
"""Mock provider with empty get_tts_audio."""
|
|
|
|
def get_tts_audio(
|
|
self, message: str, language: str, options: dict[str, Any] | None = None
|
|
) -> tts.TtsAudioType:
|
|
"""Load TTS dat."""
|
|
return (None, None)
|
|
|
|
mock_integration(hass, MockModule(domain="test"))
|
|
mock_platform(hass, "test.tts", MockTTS(MockProviderEmpty))
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
await hass.services.async_call(
|
|
tts.DOMAIN,
|
|
"test_say",
|
|
{
|
|
"entity_id": "media_player.something",
|
|
tts.ATTR_MESSAGE: "There is someone at the door.",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert len(calls) == 1
|
|
with pytest.raises(Unresolvable):
|
|
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
|
|
|
|
|
|
async def test_setup_component_load_cache_retrieve_without_mem_cache(
|
|
hass: HomeAssistant,
|
|
mock_provider: MockProvider,
|
|
empty_cache_dir,
|
|
hass_client: ClientSessionGenerator,
|
|
mock_tts,
|
|
) -> None:
|
|
"""Set up component and load cache and get without mem cache."""
|
|
# Language en is matched with en_US
|
|
_, tts_data = mock_provider.get_tts_audio("bla", "en")
|
|
assert tts_data is not None
|
|
cache_file = (
|
|
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
)
|
|
|
|
with open(cache_file, "wb") as voice_file:
|
|
voice_file.write(tts_data)
|
|
|
|
config = {tts.DOMAIN: {"platform": "test", "cache": True}}
|
|
|
|
with assert_setup_component(1, tts.DOMAIN):
|
|
assert await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
client = await hass_client()
|
|
|
|
url = "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3"
|
|
|
|
req = await client.get(url)
|
|
assert req.status == HTTPStatus.OK
|
|
assert await req.read() == tts_data
|
|
|
|
|
|
async def test_setup_component_and_web_get_url(
|
|
hass: HomeAssistant, hass_client: ClientSessionGenerator, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and receive file from web."""
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
client = await hass_client()
|
|
|
|
url = "/api/tts_get_url"
|
|
data = {"platform": "test", "message": "There is someone at the door."}
|
|
|
|
req = await client.post(url, json=data)
|
|
assert req.status == HTTPStatus.OK
|
|
response = await req.json()
|
|
assert response == {
|
|
"url": "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3",
|
|
"path": "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.mp3",
|
|
}
|
|
|
|
|
|
async def test_setup_component_and_web_get_url_bad_config(
|
|
hass: HomeAssistant, hass_client: ClientSessionGenerator, mock_tts
|
|
) -> None:
|
|
"""Set up a TTS platform and receive wrong file from web."""
|
|
config = {tts.DOMAIN: {"platform": "test"}}
|
|
|
|
await async_setup_component(hass, tts.DOMAIN, config)
|
|
|
|
client = await hass_client()
|
|
|
|
url = "/api/tts_get_url"
|
|
data = {"message": "There is someone at the door."}
|
|
|
|
req = await client.post(url, json=data)
|
|
assert req.status == HTTPStatus.BAD_REQUEST
|
|
|
|
|
|
async def test_tags_with_wave(hass: HomeAssistant, mock_provider: MockProvider) -> None:
|
|
"""Set up a TTS platform and call service and receive voice."""
|
|
|
|
# below data represents an empty wav file
|
|
tts_data = bytes.fromhex(
|
|
"52 49 46 46 24 00 00 00 57 41 56 45 66 6d 74 20 10 00 00 00 01 00 02 00"
|
|
+ "22 56 00 00 88 58 01 00 04 00 10 00 64 61 74 61 00 00 00 00"
|
|
)
|
|
|
|
tagged_data = ORIG_WRITE_TAGS(
|
|
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_test.wav",
|
|
tts_data,
|
|
mock_provider,
|
|
"AI person is in front of your door.",
|
|
"en",
|
|
None,
|
|
)
|
|
|
|
assert tagged_data != tts_data
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value",
|
|
(
|
|
"http://example.local:8123",
|
|
"http://example.local",
|
|
"http://example.local:80",
|
|
"https://example.com",
|
|
"https://example.com:443",
|
|
"https://example.com:8123",
|
|
),
|
|
)
|
|
def test_valid_base_url(value) -> None:
|
|
"""Test we validate base urls."""
|
|
assert _valid_base_url(value) == normalize_url(value)
|
|
# Test we strip trailing `/`
|
|
assert _valid_base_url(value + "/") == normalize_url(value)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value",
|
|
(
|
|
"http://example.local:8123/sub-path",
|
|
"http://example.local/sub-path",
|
|
"https://example.com/sub-path",
|
|
"https://example.com:8123/sub-path",
|
|
"mailto:some@email",
|
|
"http:example.com",
|
|
"http:/example.com",
|
|
"http//example.com",
|
|
"example.com",
|
|
),
|
|
)
|
|
def test_invalid_base_url(value) -> None:
|
|
"""Test we catch bad base urls."""
|
|
with pytest.raises(vol.Invalid):
|
|
_valid_base_url(value)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("engine", "language", "options", "cache", "result_engine", "result_query"),
|
|
(
|
|
(None, None, None, None, "test", ""),
|
|
(None, "de", None, None, "test", "language=de"),
|
|
(None, "de", {"voice": "henk"}, None, "test", "language=de&voice=henk"),
|
|
(None, "de", None, True, "test", "cache=true&language=de"),
|
|
),
|
|
)
|
|
async def test_generate_media_source_id(
|
|
hass: HomeAssistant,
|
|
setup_tts,
|
|
engine,
|
|
language,
|
|
options,
|
|
cache,
|
|
result_engine,
|
|
result_query,
|
|
) -> None:
|
|
"""Test generating a media source ID."""
|
|
media_source_id = tts.generate_media_source_id(
|
|
hass, "msg", engine, language, options, cache
|
|
)
|
|
|
|
assert media_source_id.startswith("media-source://tts/")
|
|
_, _, engine_query = media_source_id.rpartition("/")
|
|
engine, _, query = engine_query.partition("?")
|
|
assert engine == result_engine
|
|
assert query.startswith("message=msg")
|
|
assert query[12:] == result_query
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("engine", "language", "options"),
|
|
(
|
|
("not-loaded-engine", None, None),
|
|
(None, "unsupported-language", None),
|
|
(None, None, {"option": "not-supported"}),
|
|
),
|
|
)
|
|
async def test_generate_media_source_id_invalid_options(
|
|
hass: HomeAssistant, setup_tts, engine, language, options
|
|
) -> None:
|
|
"""Test generating a media source ID."""
|
|
with pytest.raises(HomeAssistantError):
|
|
tts.generate_media_source_id(hass, "msg", engine, language, options, None)
|
|
|
|
|
|
def test_resolve_engine(hass: HomeAssistant, setup_tts) -> None:
|
|
"""Test resolving engine."""
|
|
assert tts.async_resolve_engine(hass, None) == "test"
|
|
assert tts.async_resolve_engine(hass, "test") == "test"
|
|
assert tts.async_resolve_engine(hass, "non-existing") is None
|
|
|
|
with patch.dict(hass.data[tts.DOMAIN].providers, {}, clear=True):
|
|
assert tts.async_resolve_engine(hass, "test") is None
|
|
|
|
with patch.dict(hass.data[tts.DOMAIN].providers, {"cloud": object()}):
|
|
assert tts.async_resolve_engine(hass, None) == "cloud"
|
|
|
|
|
|
async def test_support_options(hass: HomeAssistant, setup_tts) -> None:
|
|
"""Test supporting options."""
|
|
# Language en is matched with en_US
|
|
assert await tts.async_support_options(hass, "test", "en") is True
|
|
assert await tts.async_support_options(hass, "test", "nl") is False
|
|
assert (
|
|
await tts.async_support_options(hass, "test", "en", {"invalid_option": "yo"})
|
|
is False
|
|
)
|
|
|
|
|
|
async def test_fetching_in_async(
|
|
hass: HomeAssistant, hass_client: ClientSessionGenerator
|
|
) -> None:
|
|
"""Test async fetching of data."""
|
|
tts_audio: asyncio.Future[bytes] = asyncio.Future()
|
|
|
|
class ProviderWithAsyncFetching(MockProvider):
|
|
"""Provider that supports audio output option."""
|
|
|
|
@property
|
|
def supported_options(self) -> list[str]:
|
|
"""Return list of supported options like voice, emotions."""
|
|
return [tts.ATTR_AUDIO_OUTPUT]
|
|
|
|
@property
|
|
def default_options(self) -> dict[str, str]:
|
|
"""Return a dict including the default options."""
|
|
return {tts.ATTR_AUDIO_OUTPUT: "mp3"}
|
|
|
|
async def async_get_tts_audio(
|
|
self, message: str, language: str, options: dict[str, Any] | None = None
|
|
) -> tts.TtsAudioType:
|
|
return ("mp3", await tts_audio)
|
|
|
|
mock_integration(hass, MockModule(domain="test"))
|
|
mock_platform(hass, "test.tts", MockTTS(ProviderWithAsyncFetching))
|
|
assert await async_setup_component(hass, tts.DOMAIN, {"tts": {"platform": "test"}})
|
|
|
|
# Test async_get_media_source_audio
|
|
media_source_id = tts.generate_media_source_id(
|
|
hass, "test message", "test", "en", None, None
|
|
)
|
|
|
|
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)
|
|
)
|
|
|
|
url = await get_media_source_url(hass, media_source_id)
|
|
client = await hass_client()
|
|
client_get_task = hass.async_create_task(client.get(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)
|
|
assert len(done) == 0
|
|
assert len(pending) == 3
|
|
|
|
tts_audio.set_result(b"test")
|
|
|
|
assert await task == ("mp3", b"test")
|
|
assert await task2 == ("mp3", 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", 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)
|
|
|
|
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",
|
|
)
|
|
|
|
|
|
async def test_ws_list_engines(
|
|
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, setup_tts
|
|
) -> None:
|
|
"""Test streaming audio and getting response."""
|
|
client = await hass_ws_client()
|
|
|
|
await client.send_json_auto_id({"type": "tts/engine/list"})
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["success"]
|
|
assert msg["result"] == {
|
|
"providers": [
|
|
{
|
|
"engine_id": "test",
|
|
"supported_languages": ["de_CH", "de_DE", "en_GB", "en_US"],
|
|
}
|
|
]
|
|
}
|
|
|
|
await client.send_json_auto_id({"type": "tts/engine/list", "language": "smurfish"})
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["success"]
|
|
assert msg["result"] == {
|
|
"providers": [{"engine_id": "test", "supported_languages": []}]
|
|
}
|
|
|
|
await client.send_json_auto_id({"type": "tts/engine/list", "language": "en"})
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["success"]
|
|
assert msg["result"] == {
|
|
"providers": [{"engine_id": "test", "supported_languages": ["en_US", "en_GB"]}]
|
|
}
|
|
|
|
await client.send_json_auto_id({"type": "tts/engine/list", "language": "en-UK"})
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["success"]
|
|
assert msg["result"] == {
|
|
"providers": [{"engine_id": "test", "supported_languages": ["en_GB", "en_US"]}]
|
|
}
|
|
|
|
await client.send_json_auto_id({"type": "tts/engine/list", "language": "de"})
|
|
msg = await client.receive_json()
|
|
assert msg["type"] == "result"
|
|
assert msg["success"]
|
|
assert msg["result"] == {
|
|
"providers": [{"engine_id": "test", "supported_languages": ["de_DE", "de_CH"]}]
|
|
}
|
|
|
|
await client.send_json_auto_id(
|
|
{"type": "tts/engine/list", "language": "de", "country": "ch"}
|
|
)
|
|
msg = await client.receive_json()
|
|
assert msg["type"] == "result"
|
|
assert msg["success"]
|
|
assert msg["result"] == {
|
|
"providers": [{"engine_id": "test", "supported_languages": ["de_CH", "de_DE"]}]
|
|
}
|
|
|
|
|
|
async def test_ws_list_voices(
|
|
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, setup_tts
|
|
) -> None:
|
|
"""Test streaming audio and getting response."""
|
|
client = await hass_ws_client()
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "tts/engine/voices",
|
|
"engine_id": "smurf_tts",
|
|
"language": "smurfish",
|
|
}
|
|
)
|
|
|
|
msg = await client.receive_json()
|
|
assert not msg["success"]
|
|
assert msg["error"] == {
|
|
"code": "not_found",
|
|
"message": "tts engine smurf_tts not found",
|
|
}
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "tts/engine/voices",
|
|
"engine_id": "test",
|
|
"language": "smurfish",
|
|
}
|
|
)
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["success"]
|
|
assert msg["result"] == {"voices": None}
|
|
|
|
await client.send_json_auto_id(
|
|
{
|
|
"type": "tts/engine/voices",
|
|
"engine_id": "test",
|
|
"language": "en-US",
|
|
}
|
|
)
|
|
|
|
msg = await client.receive_json()
|
|
assert msg["success"]
|
|
assert msg["result"] == {"voices": ["James Earl Jones", "Fran Drescher"]}
|