evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tts.say to use media source URLs (#70382)

Browse source
This commit is contained in:
Paulus Schoutsen 2022-04-26 13:49:32 -07:00 • committed by GitHub
parent e387e6d332
commit 9303e35a7d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 215 additions and 101 deletions
Download patch file Download diff file Expand all files Collapse all files

70
homeassistant/components/tts/__init__.py
View file

@ -27,6 +27,7 @@ from homeassistant.components.media_player.const import (
MEDIA_TYPE_MUSIC, MEDIA_TYPE_MUSIC,
SERVICE_PLAY_MEDIA, SERVICE_PLAY_MEDIA,
) )
from homeassistant.components.media_source import generate_media_source_id
from homeassistant.const import ( from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_ENTITY_ID,
CONF_DESCRIPTION, CONF_DESCRIPTION,
@ -141,6 +142,10 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
cache_dir = conf.get(CONF_CACHE_DIR, DEFAULT_CACHE_DIR) cache_dir = conf.get(CONF_CACHE_DIR, DEFAULT_CACHE_DIR)
time_memory = conf.get(CONF_TIME_MEMORY, DEFAULT_TIME_MEMORY) time_memory = conf.get(CONF_TIME_MEMORY, DEFAULT_TIME_MEMORY)
base_url = conf.get(CONF_BASE_URL) base_url = conf.get(CONF_BASE_URL)
if base_url is not None:
_LOGGER.warning(
"TTS base_url option is deprecated. Configure internal/external URL instead"
)
hass.data[BASE_URL_KEY] = base_url hass.data[BASE_URL_KEY] = base_url
await tts.async_init_cache(use_cache, cache_dir, time_memory, base_url) await tts.async_init_cache(use_cache, cache_dir, time_memory, base_url)
@ -198,6 +203,34 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
language = service.data.get(ATTR_LANGUAGE) language = service.data.get(ATTR_LANGUAGE)
options = service.data.get(ATTR_OPTIONS) options = service.data.get(ATTR_OPTIONS)
if tts.base_url is None or tts.base_url == get_url(hass):
tts.process_options(p_type, language, options)
params = {
"message": message,
}
if cache is not None:
params["cache"] = "true" if cache else "false"
if language is not None:
params["language"] = language
if options is not None:
params.update(options)
await hass.services.async_call(
DOMAIN_MP,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: entity_ids,
ATTR_MEDIA_CONTENT_ID: generate_media_source_id(
DOMAIN,
str(yarl.URL.build(path=p_type, query=params)),
),
ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
},
blocking=True,
context=service.context,
)
return
try: try:
url = await tts.async_get_url_path( url = await tts.async_get_url_path(
p_type, message, cache=cache, language=language, options=options p_type, message, cache=cache, language=language, options=options
@ -344,24 +377,17 @@ class SpeechManager:
PLATFORM_FORMAT.format(domain=engine, platform=DOMAIN) PLATFORM_FORMAT.format(domain=engine, platform=DOMAIN)
) )
async def async_get_url_path( @callback
def process_options(
self, self,
engine: str, engine: str,
message: str,
cache: bool | None = None,
language: str | None = None, language: str | None = None,
options: dict | None = None, options: dict | None = None,
) -> str: ) -> tuple[str, dict | None]:
"""Get URL for play message. """Validate and process options."""
This method is a coroutine.
"""
if (provider := self.providers.get(engine)) is None: if (provider := self.providers.get(engine)) is None:
raise HomeAssistantError(f"Provider {engine} not found") raise HomeAssistantError(f"Provider {engine} not found")
msg_hash = hashlib.sha1(bytes(message, "utf-8")).hexdigest()
use_cache = cache if cache is not None else self.use_cache
# Languages # Languages
language = language or provider.default_language language = language or provider.default_language
if language is None or language not in provider.supported_languages: if language is None or language not in provider.supported_languages:
@ -382,9 +408,25 @@ class SpeechManager:
] ]
if invalid_opts: if invalid_opts:
raise HomeAssistantError(f"Invalid options found: {invalid_opts}") raise HomeAssistantError(f"Invalid options found: {invalid_opts}")
options_key = _hash_options(options)
else: return language, options
options_key = "-"
async def async_get_url_path(
self,
engine: str,
message: str,
cache: bool | None = None,
language: str | None = None,
options: dict | None = None,
) -> str:
"""Get URL for play message.
This method is a coroutine.
"""
language, options = self.process_options(engine, language, options)
options_key = _hash_options(options) if options else "-"
msg_hash = hashlib.sha1(bytes(message, "utf-8")).hexdigest()
use_cache = cache if cache is not None else self.use_cache
key = KEY_PATTERN.format( key = KEY_PATTERN.format(
msg_hash, language.replace("_", "-"), options_key, engine msg_hash, language.replace("_", "-"), options_key, engine

8
homeassistant/components/tts/media_source.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations from __future__ import annotations
import mimetypes import mimetypes
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, Any
from yarl import URL from yarl import URL
@ -46,17 +46,19 @@ class TTSMediaSource(MediaSource):
raise Unresolvable("No message specified.") raise Unresolvable("No message specified.")
options = dict(parsed.query) options = dict(parsed.query)
kwargs = { kwargs: dict[str, Any] = {
"engine": parsed.name, "engine": parsed.name,
"message": options.pop("message"), "message": options.pop("message"),
"language": options.pop("language", None), "language": options.pop("language", None),
"options": options, "options": options,
} }
if "cache" in options:
kwargs["cache"] = options.pop("cache") == "true"
manager: SpeechManager = self.hass.data[DOMAIN] manager: SpeechManager = self.hass.data[DOMAIN]
try: try:
url = await manager.async_get_url_path(**kwargs) # type: ignore[arg-type] url = await manager.async_get_url_path(**kwargs)
except HomeAssistantError as err: except HomeAssistantError as err:
raise Unresolvable(str(err)) from err raise Unresolvable(str(err)) from err

21
tests/components/google_translate/test_tts.py
View file

@ -6,19 +6,29 @@ from unittest.mock import patch
from gtts import gTTSError from gtts import gTTSError
import pytest import pytest
from homeassistant.components import media_source, tts
from homeassistant.components.media_player.const import ( from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID, ATTR_MEDIA_CONTENT_ID,
DOMAIN as DOMAIN_MP, DOMAIN as DOMAIN_MP,
SERVICE_PLAY_MEDIA, SERVICE_PLAY_MEDIA,
) )
import homeassistant.components.tts as tts
from homeassistant.config import async_process_ha_core_config from homeassistant.config import async_process_ha_core_config
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import async_mock_service from tests.common import async_mock_service
from tests.components.tts.conftest import mutagen_mock # noqa: F401 from tests.components.tts.conftest import mutagen_mock # noqa: F401
async def get_media_source_url(hass, media_content_id):
"""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)
return resolved.url
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def cleanup_cache(hass): def cleanup_cache(hass):
"""Clean up TTS cache.""" """Clean up TTS cache."""
@ -67,8 +77,9 @@ async def test_service_say(hass, mock_gtts, calls):
) )
assert len(calls) == 1 assert len(calls) == 1
url = await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(mock_gtts.mock_calls) == 2 assert len(mock_gtts.mock_calls) == 2
assert calls[0].data[ATTR_MEDIA_CONTENT_ID].find(".mp3") != -1 assert url.endswith(".mp3")
assert mock_gtts.mock_calls[0][2] == { assert mock_gtts.mock_calls[0][2] == {
"text": "There is a person at the front door.", "text": "There is a person at the front door.",
@ -96,6 +107,7 @@ async def test_service_say_german_config(hass, mock_gtts, calls):
) )
assert len(calls) == 1 assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(mock_gtts.mock_calls) == 2 assert len(mock_gtts.mock_calls) == 2
assert mock_gtts.mock_calls[0][2] == { assert mock_gtts.mock_calls[0][2] == {
"text": "There is a person at the front door.", "text": "There is a person at the front door.",
@ -124,6 +136,7 @@ async def test_service_say_german_service(hass, mock_gtts, calls):
) )
assert len(calls) == 1 assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(mock_gtts.mock_calls) == 2 assert len(mock_gtts.mock_calls) == 2
assert mock_gtts.mock_calls[0][2] == { assert mock_gtts.mock_calls[0][2] == {
"text": "There is a person at the front door.", "text": "There is a person at the front door.",
@ -148,5 +161,7 @@ async def test_service_say_error(hass, mock_gtts, calls):
blocking=True, blocking=True,
) )
assert len(calls) == 0 assert len(calls) == 1
with pytest.raises(HomeAssistantError):
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(mock_gtts.mock_calls) == 2 assert len(mock_gtts.mock_calls) == 2

23
tests/components/marytts/test_tts.py
View file

@ -5,17 +5,26 @@ from unittest.mock import patch
import pytest import pytest
from homeassistant.components import media_source, tts
from homeassistant.components.media_player.const import ( from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID, ATTR_MEDIA_CONTENT_ID,
DOMAIN as DOMAIN_MP, DOMAIN as DOMAIN_MP,
SERVICE_PLAY_MEDIA, SERVICE_PLAY_MEDIA,
) )
import homeassistant.components.tts as tts
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component, async_mock_service from tests.common import assert_setup_component, async_mock_service
async def get_media_source_url(hass, media_content_id):
"""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)
return resolved.url
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def cleanup_cache(hass): def cleanup_cache(hass):
"""Prevent TTS writing.""" """Prevent TTS writing."""
@ -58,11 +67,13 @@ async def test_service_say(hass):
blocking=True, blocking=True,
) )
url = await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
mock_speak.assert_called_once() mock_speak.assert_called_once()
mock_speak.assert_called_with("HomeAssistant", {}) mock_speak.assert_called_with("HomeAssistant", {})
assert len(calls) == 1 assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_ID].find(".wav") != -1 assert url.endswith(".wav")
async def test_service_say_with_effect(hass): async def test_service_say_with_effect(hass):
@ -89,11 +100,13 @@ async def test_service_say_with_effect(hass):
blocking=True, blocking=True,
) )
url = await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
mock_speak.assert_called_once() mock_speak.assert_called_once()
mock_speak.assert_called_with("HomeAssistant", {"Volume": "amount:2.0;"}) mock_speak.assert_called_with("HomeAssistant", {"Volume": "amount:2.0;"})
assert len(calls) == 1 assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_ID].find(".wav") != -1 assert url.endswith(".wav")
async def test_service_say_http_error(hass): async def test_service_say_http_error(hass):
@ -120,5 +133,7 @@ async def test_service_say_http_error(hass):
) )
await hass.async_block_till_done() await hass.async_block_till_done()
with pytest.raises(Exception):
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
mock_speak.assert_called_once() mock_speak.assert_called_once()
assert len(calls) == 0

102
tests/components/tts/test_init.py
View file

@ -4,9 +4,8 @@ from unittest.mock import PropertyMock, patch
import pytest import pytest
import voluptuous as vol import voluptuous as vol
import yarl
from homeassistant.components import tts from homeassistant.components import media_source, tts
from homeassistant.components.demo.tts import DemoProvider from homeassistant.components.demo.tts import DemoProvider
from homeassistant.components.media_player.const import ( from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID, ATTR_MEDIA_CONTENT_ID,
@ -16,6 +15,7 @@ from homeassistant.components.media_player.const import (
SERVICE_PLAY_MEDIA, SERVICE_PLAY_MEDIA,
) )
from homeassistant.config import async_process_ha_core_config from homeassistant.config import async_process_ha_core_config
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from homeassistant.util.network import normalize_url from homeassistant.util.network import normalize_url
@ -24,9 +24,13 @@ from tests.common import assert_setup_component, async_mock_service
ORIG_WRITE_TAGS = tts.SpeechManager.write_tags ORIG_WRITE_TAGS = tts.SpeechManager.write_tags
def relative_url(url): async def get_media_source_url(hass, media_content_id):
"""Convert an absolute url to a relative one.""" """Get the media source url."""
return str(yarl.URL(url).relative()) 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)
return resolved.url
@pytest.fixture @pytest.fixture
@ -89,8 +93,8 @@ async def test_setup_component_and_test_service(hass, empty_cache_dir):
assert len(calls) == 1 assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert ( assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID] await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
== "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3" == "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
@ -121,8 +125,8 @@ async def test_setup_component_and_test_service_with_config_language(
assert len(calls) == 1 assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert ( assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID] await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
== "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3" == "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3"
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
@ -156,8 +160,8 @@ async def test_setup_component_and_test_service_with_config_language_special(
assert len(calls) == 1 assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert ( assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID] await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
== "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_demo.mp3" == "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en-us_-_demo.mp3"
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
@ -197,8 +201,8 @@ async def test_setup_component_and_test_service_with_service_language(
assert len(calls) == 1 assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert ( assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID] await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
== "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3" == "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3"
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
@ -217,18 +221,18 @@ async def test_setup_component_test_service_with_wrong_service_language(
with assert_setup_component(1, tts.DOMAIN): with assert_setup_component(1, tts.DOMAIN):
assert await async_setup_component(hass, tts.DOMAIN, config) assert await async_setup_component(hass, tts.DOMAIN, config)
await hass.services.async_call( with pytest.raises(HomeAssistantError):
tts.DOMAIN, await hass.services.async_call(
"demo_say", tts.DOMAIN,
{ "demo_say",
"entity_id": "media_player.something", {
tts.ATTR_MESSAGE: "There is someone at the door.", "entity_id": "media_player.something",
tts.ATTR_LANGUAGE: "lang", tts.ATTR_MESSAGE: "There is someone at the door.",
}, tts.ATTR_LANGUAGE: "lang",
blocking=True, },
) blocking=True,
)
assert len(calls) == 0 assert len(calls) == 0
await hass.async_block_till_done()
assert not ( assert not (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_lang_-_demo.mp3" empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_lang_-_demo.mp3"
).is_file() ).is_file()
@ -261,8 +265,8 @@ async def test_setup_component_and_test_service_with_service_options(
assert len(calls) == 1 assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert ( assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID] await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
== f"http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{opt_hash}_demo.mp3" == f"/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{opt_hash}_demo.mp3"
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
@ -298,8 +302,8 @@ async def test_setup_component_and_test_with_service_options_def(hass, empty_cac
assert len(calls) == 1 assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert ( assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID] await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
== f"http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{opt_hash}_demo.mp3" == f"/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{opt_hash}_demo.mp3"
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
@ -319,17 +323,18 @@ async def test_setup_component_and_test_service_with_service_options_wrong(
with assert_setup_component(1, tts.DOMAIN): with assert_setup_component(1, tts.DOMAIN):
assert await async_setup_component(hass, tts.DOMAIN, config) assert await async_setup_component(hass, tts.DOMAIN, config)
await hass.services.async_call( with pytest.raises(HomeAssistantError):
tts.DOMAIN, await hass.services.async_call(
"demo_say", tts.DOMAIN,
{ "demo_say",
"entity_id": "media_player.something", {
tts.ATTR_MESSAGE: "There is someone at the door.", "entity_id": "media_player.something",
tts.ATTR_LANGUAGE: "de", tts.ATTR_MESSAGE: "There is someone at the door.",
tts.ATTR_OPTIONS: {"speed": 1}, tts.ATTR_LANGUAGE: "de",
}, tts.ATTR_OPTIONS: {"speed": 1},
blocking=True, },
) blocking=True,
)
opt_hash = tts._hash_options({"speed": 1}) opt_hash = tts._hash_options({"speed": 1})
assert len(calls) == 0 assert len(calls) == 0
@ -386,8 +391,8 @@ async def test_setup_component_and_test_service_clear_cache(hass, empty_cache_di
blocking=True, blocking=True,
) )
# To make sure the file is persisted # To make sure the file is persisted
await hass.async_block_till_done()
assert len(calls) == 1 assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3" empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
@ -397,7 +402,6 @@ async def test_setup_component_and_test_service_clear_cache(hass, empty_cache_di
tts.DOMAIN, tts.SERVICE_CLEAR_CACHE, {}, blocking=True tts.DOMAIN, tts.SERVICE_CLEAR_CACHE, {}, blocking=True
) )
await hass.async_block_till_done()
assert not ( assert not (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3" empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
).is_file() ).is_file()
@ -414,8 +418,6 @@ async def test_setup_component_and_test_service_with_receive_voice(
with assert_setup_component(1, tts.DOMAIN): with assert_setup_component(1, tts.DOMAIN):
assert await async_setup_component(hass, tts.DOMAIN, config) assert await async_setup_component(hass, tts.DOMAIN, config)
client = await hass_client()
await hass.services.async_call( await hass.services.async_call(
tts.DOMAIN, tts.DOMAIN,
"demo_say", "demo_say",
@ -427,7 +429,9 @@ async def test_setup_component_and_test_service_with_receive_voice(
) )
assert len(calls) == 1 assert len(calls) == 1
req = await client.get(relative_url(calls[0].data[ATTR_MEDIA_CONTENT_ID])) url = await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
client = await hass_client()
req = await client.get(url)
_, demo_data = demo_provider.get_tts_audio("bla", "en") _, demo_data = demo_provider.get_tts_audio("bla", "en")
demo_data = tts.SpeechManager.write_tags( demo_data = tts.SpeechManager.write_tags(
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3", "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3",
@ -452,8 +456,6 @@ async def test_setup_component_and_test_service_with_receive_voice_german(
with assert_setup_component(1, tts.DOMAIN): with assert_setup_component(1, tts.DOMAIN):
assert await async_setup_component(hass, tts.DOMAIN, config) assert await async_setup_component(hass, tts.DOMAIN, config)
client = await hass_client()
await hass.services.async_call( await hass.services.async_call(
tts.DOMAIN, tts.DOMAIN,
"demo_say", "demo_say",
@ -464,7 +466,9 @@ async def test_setup_component_and_test_service_with_receive_voice_german(
blocking=True, blocking=True,
) )
assert len(calls) == 1 assert len(calls) == 1
req = await client.get(relative_url(calls[0].data[ATTR_MEDIA_CONTENT_ID])) url = await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
client = await hass_client()
req = await client.get(url)
_, demo_data = demo_provider.get_tts_audio("bla", "de") _, demo_data = demo_provider.get_tts_audio("bla", "de")
demo_data = tts.SpeechManager.write_tags( demo_data = tts.SpeechManager.write_tags(
"42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3", "42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3",
@ -595,8 +599,8 @@ async def test_setup_component_test_with_cache_dir(
) )
assert len(calls) == 1 assert len(calls) == 1
assert ( assert (
calls[0].data[ATTR_MEDIA_CONTENT_ID] await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
== "http://example.local:8123/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3" == "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
) )

26
tests/components/voicerss/test_tts.py
View file

@ -6,12 +6,13 @@ import shutil
import pytest import pytest
from homeassistant.components import media_source, tts
from homeassistant.components.media_player.const import ( from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID, ATTR_MEDIA_CONTENT_ID,
DOMAIN as DOMAIN_MP, DOMAIN as DOMAIN_MP,
SERVICE_PLAY_MEDIA, SERVICE_PLAY_MEDIA,
) )
import homeassistant.components.tts as tts from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component, async_mock_service from tests.common import assert_setup_component, async_mock_service
@ -27,6 +28,15 @@ FORM_DATA = {
} }
async def get_media_source_url(hass, media_content_id):
"""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)
return resolved.url
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def cleanup_cache(hass): def cleanup_cache(hass):
"""Prevent TTS writing.""" """Prevent TTS writing."""
@ -77,9 +87,10 @@ async def test_service_say(hass, aioclient_mock):
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(calls) == 1
url = await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert url.endswith(".mp3")
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == FORM_DATA assert aioclient_mock.mock_calls[0][2] == FORM_DATA
assert calls[0].data[ATTR_MEDIA_CONTENT_ID].find(".mp3") != -1
async def test_service_say_german_config(hass, aioclient_mock): async def test_service_say_german_config(hass, aioclient_mock):
@ -112,6 +123,7 @@ async def test_service_say_german_config(hass, aioclient_mock):
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == form_data assert aioclient_mock.mock_calls[0][2] == form_data
@ -141,6 +153,7 @@ async def test_service_say_german_service(hass, aioclient_mock):
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 1 assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == form_data assert aioclient_mock.mock_calls[0][2] == form_data
@ -167,7 +180,8 @@ async def test_service_say_error(hass, aioclient_mock):
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 with pytest.raises(HomeAssistantError):
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == FORM_DATA assert aioclient_mock.mock_calls[0][2] == FORM_DATA
@ -194,7 +208,8 @@ async def test_service_say_timeout(hass, aioclient_mock):
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 with pytest.raises(HomeAssistantError):
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == FORM_DATA assert aioclient_mock.mock_calls[0][2] == FORM_DATA
@ -226,6 +241,7 @@ async def test_service_say_error_msg(hass, aioclient_mock):
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 with pytest.raises(media_source.Unresolvable):
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == FORM_DATA assert aioclient_mock.mock_calls[0][2] == FORM_DATA

66
tests/components/yandextts/test_tts.py
View file

@ -6,11 +6,12 @@ import shutil
import pytest import pytest
from homeassistant.components import media_source, tts
from homeassistant.components.media_player.const import ( from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID,
DOMAIN as DOMAIN_MP, DOMAIN as DOMAIN_MP,
SERVICE_PLAY_MEDIA, SERVICE_PLAY_MEDIA,
) )
import homeassistant.components.tts as tts
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from tests.common import assert_setup_component, async_mock_service from tests.common import assert_setup_component, async_mock_service
@ -21,6 +22,15 @@ from tests.components.tts.conftest import ( # noqa: F401, pylint: disable=unuse
URL = "https://tts.voicetech.yandex.net/generate?" URL = "https://tts.voicetech.yandex.net/generate?"
async def get_media_source_url(hass, media_content_id):
"""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)
return resolved.url
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def cleanup_cache(hass): def cleanup_cache(hass):
"""Prevent TTS writing.""" """Prevent TTS writing."""
@ -73,11 +83,11 @@ async def test_service_say(hass, aioclient_mock):
tts.DOMAIN, tts.DOMAIN,
"yandextts_say", "yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"}, {"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
blocking=True,
) )
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1 assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1
async def test_service_say_russian_config(hass, aioclient_mock): async def test_service_say_russian_config(hass, aioclient_mock):
@ -111,11 +121,12 @@ async def test_service_say_russian_config(hass, aioclient_mock):
tts.DOMAIN, tts.DOMAIN,
"yandextts_say", "yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"}, {"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
blocking=True,
) )
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1 assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1
async def test_service_say_russian_service(hass, aioclient_mock): async def test_service_say_russian_service(hass, aioclient_mock):
@ -147,11 +158,11 @@ async def test_service_say_russian_service(hass, aioclient_mock):
tts.ATTR_MESSAGE: "HomeAssistant", tts.ATTR_MESSAGE: "HomeAssistant",
tts.ATTR_LANGUAGE: "ru-RU", tts.ATTR_LANGUAGE: "ru-RU",
}, },
blocking=True,
) )
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1 assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1
async def test_service_say_timeout(hass, aioclient_mock): async def test_service_say_timeout(hass, aioclient_mock):
@ -184,10 +195,13 @@ async def test_service_say_timeout(hass, aioclient_mock):
tts.DOMAIN, tts.DOMAIN,
"yandextts_say", "yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"}, {"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
blocking=True,
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert len(calls) == 0 assert len(calls) == 1
with pytest.raises(media_source.Unresolvable):
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
@ -221,10 +235,12 @@ async def test_service_say_http_error(hass, aioclient_mock):
tts.DOMAIN, tts.DOMAIN,
"yandextts_say", "yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"}, {"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
blocking=True,
) )
await hass.async_block_till_done()
assert len(calls) == 0 assert len(calls) == 1
with pytest.raises(media_source.Unresolvable):
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
async def test_service_say_specified_speaker(hass, aioclient_mock): async def test_service_say_specified_speaker(hass, aioclient_mock):
@ -258,11 +274,11 @@ async def test_service_say_specified_speaker(hass, aioclient_mock):
tts.DOMAIN, tts.DOMAIN,
"yandextts_say", "yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"}, {"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
blocking=True,
) )
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1 assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1
async def test_service_say_specified_emotion(hass, aioclient_mock): async def test_service_say_specified_emotion(hass, aioclient_mock):
@ -296,11 +312,12 @@ async def test_service_say_specified_emotion(hass, aioclient_mock):
tts.DOMAIN, tts.DOMAIN,
"yandextts_say", "yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"}, {"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
blocking=True,
) )
await hass.async_block_till_done() assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1
async def test_service_say_specified_low_speed(hass, aioclient_mock): async def test_service_say_specified_low_speed(hass, aioclient_mock):
@ -330,11 +347,12 @@ async def test_service_say_specified_low_speed(hass, aioclient_mock):
tts.DOMAIN, tts.DOMAIN,
"yandextts_say", "yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"}, {"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
blocking=True,
) )
await hass.async_block_till_done() assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1
async def test_service_say_specified_speed(hass, aioclient_mock): async def test_service_say_specified_speed(hass, aioclient_mock):
@ -362,11 +380,12 @@ async def test_service_say_specified_speed(hass, aioclient_mock):
tts.DOMAIN, tts.DOMAIN,
"yandextts_say", "yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"}, {"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
blocking=True,
) )
await hass.async_block_till_done() assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1
async def test_service_say_specified_options(hass, aioclient_mock): async def test_service_say_specified_options(hass, aioclient_mock):
@ -397,8 +416,9 @@ async def test_service_say_specified_options(hass, aioclient_mock):
tts.ATTR_MESSAGE: "HomeAssistant", tts.ATTR_MESSAGE: "HomeAssistant",
"options": {"emotion": "evil", "speed": 2}, "options": {"emotion": "evil", "speed": 2},
}, },
blocking=True,
) )
await hass.async_block_till_done() assert len(calls) == 1
await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1