Convert TTS tests to async (#33517)

* Convert TTS tests to async

* Address comments
This commit is contained in:
Paulus Schoutsen 2020-04-02 09:55:34 -07:00 committed by GitHub
parent 1d2713b0ea
commit b10319f69e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 501 additions and 532 deletions

View File

@ -133,7 +133,7 @@ async def async_setup(hass, config):
hass, p_config, discovery_info
)
else:
provider = await hass.async_add_job(
provider = await hass.async_add_executor_job(
platform.get_engine, hass, p_config, discovery_info
)
@ -226,41 +226,17 @@ class SpeechManager:
self.time_memory = time_memory
self.base_url = base_url
def init_tts_cache_dir(cache_dir):
"""Init cache folder."""
if not os.path.isabs(cache_dir):
cache_dir = self.hass.config.path(cache_dir)
if not os.path.isdir(cache_dir):
_LOGGER.info("Create cache dir %s.", cache_dir)
os.mkdir(cache_dir)
return cache_dir
try:
self.cache_dir = await self.hass.async_add_job(
init_tts_cache_dir, cache_dir
self.cache_dir = await self.hass.async_add_executor_job(
_init_tts_cache_dir, self.hass, cache_dir
)
except OSError as err:
raise HomeAssistantError(f"Can't init cache dir {err}")
def get_cache_files():
"""Return a dict of given engine files."""
cache = {}
folder_data = os.listdir(self.cache_dir)
for file_data in folder_data:
record = _RE_VOICE_FILE.match(file_data)
if record:
key = KEY_PATTERN.format(
record.group(1),
record.group(2),
record.group(3),
record.group(4),
)
cache[key.lower()] = file_data.lower()
return cache
try:
cache_files = await self.hass.async_add_job(get_cache_files)
cache_files = await self.hass.async_add_executor_job(
_get_cache_files, self.cache_dir
)
except OSError as err:
raise HomeAssistantError(f"Can't read cache dir {err}")
@ -273,13 +249,13 @@ class SpeechManager:
def remove_files():
"""Remove files from filesystem."""
for _, filename in self.file_cache.items():
for filename in self.file_cache.values():
try:
os.remove(os.path.join(self.cache_dir, filename))
except OSError as err:
_LOGGER.warning("Can't remove cache file '%s': %s", filename, err)
await self.hass.async_add_job(remove_files)
await self.hass.async_add_executor_job(remove_files)
self.file_cache = {}
@callback
@ -312,6 +288,7 @@ class SpeechManager:
merged_options.update(options)
options = merged_options
options = options or provider.default_options
if options is not None:
invalid_opts = [
opt_name
@ -378,10 +355,10 @@ class SpeechManager:
speech.write(data)
try:
await self.hass.async_add_job(save_speech)
await self.hass.async_add_executor_job(save_speech)
self.file_cache[key] = filename
except OSError:
_LOGGER.error("Can't write %s", filename)
except OSError as err:
_LOGGER.error("Can't write %s: %s", filename, err)
async def async_file_to_mem(self, key):
"""Load voice from file cache into memory.
@ -400,7 +377,7 @@ class SpeechManager:
return speech.read()
try:
data = await self.hass.async_add_job(load_speech)
data = await self.hass.async_add_executor_job(load_speech)
except OSError:
del self.file_cache[key]
raise HomeAssistantError(f"Can't read {voice_file}")
@ -506,11 +483,36 @@ class Provider:
Return a tuple of file extension and data as bytes.
"""
return await self.hass.async_add_job(
return await self.hass.async_add_executor_job(
ft.partial(self.get_tts_audio, message, language, options=options)
)
def _init_tts_cache_dir(hass, cache_dir):
"""Init cache folder."""
if not os.path.isabs(cache_dir):
cache_dir = hass.config.path(cache_dir)
if not os.path.isdir(cache_dir):
_LOGGER.info("Create cache dir %s", cache_dir)
os.mkdir(cache_dir)
return cache_dir
def _get_cache_files(cache_dir):
"""Return a dict of given engine files."""
cache = {}
folder_data = os.listdir(cache_dir)
for file_data in folder_data:
record = _RE_VOICE_FILE.match(file_data)
if record:
key = KEY_PATTERN.format(
record.group(1), record.group(2), record.group(3), record.group(4),
)
cache[key.lower()] = file_data.lower()
return cache
class TextToSpeechUrlView(HomeAssistantView):
"""TTS view to get a url to a generated speech file."""

View File

@ -14,6 +14,8 @@ import threading
from unittest.mock import MagicMock, Mock, patch
import uuid
from aiohttp.test_utils import unused_port as get_test_instance_port # noqa
from homeassistant import auth, config_entries, core as ha, loader
from homeassistant.auth import (
auth_store,
@ -37,7 +39,6 @@ from homeassistant.const import (
EVENT_PLATFORM_DISCOVERED,
EVENT_STATE_CHANGED,
EVENT_TIME_CHANGED,
SERVER_PORT,
STATE_OFF,
STATE_ON,
)
@ -59,7 +60,6 @@ import homeassistant.util.dt as date_util
from homeassistant.util.unit_system import METRIC_SYSTEM
import homeassistant.util.yaml.loader as yaml_loader
_TEST_INSTANCE_PORT = SERVER_PORT
_LOGGER = logging.getLogger(__name__)
INSTANCES = []
CLIENT_ID = "https://example.com/app"
@ -217,18 +217,6 @@ async def async_test_home_assistant(loop):
return hass
def get_test_instance_port():
"""Return unused port for running test instance.
The socket that holds the default port does not get released when we stop
HA in a different test case. Until I have figured out what is going on,
let's run each test on a different port.
"""
global _TEST_INSTANCE_PORT
_TEST_INSTANCE_PORT += 1
return _TEST_INSTANCE_PORT
def async_mock_service(hass, domain, service, schema=None):
"""Set up a fake service & return a calls log list to this service."""
calls = []

View File

@ -1,14 +1,12 @@
"""The tests for the TTS component."""
import ctypes
import os
import shutil
from unittest.mock import PropertyMock, patch
import pytest
import requests
import yarl
from homeassistant.components.demo.tts import DemoProvider
import homeassistant.components.http as http
from homeassistant.components.media_player.const import (
ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE,
@ -17,15 +15,52 @@ from homeassistant.components.media_player.const import (
SERVICE_PLAY_MEDIA,
)
import homeassistant.components.tts as tts
from homeassistant.setup import async_setup_component, setup_component
from homeassistant.components.tts import _get_cache_files
from homeassistant.setup import async_setup_component
from tests.common import (
assert_setup_component,
get_test_home_assistant,
get_test_instance_port,
mock_service,
mock_storage,
)
from tests.common import assert_setup_component, async_mock_service
def relative_url(url):
"""Convert an absolute url to a relative one."""
return str(yarl.URL(url).relative())
@pytest.fixture
def demo_provider():
"""Demo TTS provider."""
return DemoProvider("en")
@pytest.fixture(autouse=True)
def mock_get_cache_files():
"""Mock the list TTS cache function."""
with patch(
"homeassistant.components.tts._get_cache_files", return_value={}
) as mock_cache_files:
yield mock_cache_files
@pytest.fixture(autouse=True)
def mock_init_cache_dir():
"""Mock the TTS cache dir in memory."""
with patch(
"homeassistant.components.tts._init_tts_cache_dir",
side_effect=lambda hass, cache_dir: hass.config.path(cache_dir),
) as mock_cache_dir:
yield mock_cache_dir
@pytest.fixture
def empty_cache_dir(tmp_path, mock_init_cache_dir, mock_get_cache_files):
"""Mock the TTS cache dir with empty dir."""
mock_init_cache_dir.side_effect = None
mock_init_cache_dir.return_value = str(tmp_path)
# Restore original get cache files behavior, we're working with a real dir.
mock_get_cache_files.side_effect = _get_cache_files
return tmp_path
@pytest.fixture(autouse=True)
@ -38,134 +73,111 @@ def mutagen_mock():
yield
class TestTTS:
"""Test the Google speech component."""
def setup_method(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.demo_provider = DemoProvider("en")
self.default_tts_cache = self.hass.config.path(tts.DEFAULT_CACHE_DIR)
self.mock_storage = mock_storage()
self.mock_storage.__enter__()
setup_component(
self.hass,
http.DOMAIN,
{http.DOMAIN: {http.CONF_SERVER_PORT: get_test_instance_port()}},
)
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
self.mock_storage.__exit__(None, None, None)
if os.path.isdir(self.default_tts_cache):
shutil.rmtree(self.default_tts_cache)
def test_setup_component_demo(self):
async def test_setup_component_demo(hass):
"""Set up the demo platform with defaults."""
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
assert self.hass.services.has_service(tts.DOMAIN, "demo_say")
assert self.hass.services.has_service(tts.DOMAIN, "clear_cache")
assert hass.services.has_service(tts.DOMAIN, "demo_say")
assert hass.services.has_service(tts.DOMAIN, "clear_cache")
@patch("os.mkdir", side_effect=OSError(2, "No access"))
def test_setup_component_demo_no_access_cache_folder(self, mock_mkdir):
async def test_setup_component_demo_no_access_cache_folder(hass, mock_init_cache_dir):
"""Set up the demo platform with defaults."""
config = {tts.DOMAIN: {"platform": "demo"}}
assert not setup_component(self.hass, tts.DOMAIN, config)
mock_init_cache_dir.side_effect = OSError(2, "No access")
assert not await async_setup_component(hass, tts.DOMAIN, config)
assert not self.hass.services.has_service(tts.DOMAIN, "demo_say")
assert not self.hass.services.has_service(tts.DOMAIN, "clear_cache")
assert not hass.services.has_service(tts.DOMAIN, "demo_say")
assert not hass.services.has_service(tts.DOMAIN, "clear_cache")
def test_setup_component_and_test_service(self):
async def test_setup_component_and_test_service(hass, empty_cache_dir):
"""Set up the demo platform and call service."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "There is someone at the door.",
},
blocking=True,
)
self.hass.block_till_done()
assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3".format(
self.hass.config.api.base_url
)
assert os.path.isfile(
os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3",
)
hass.config.api.base_url
)
assert (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
).is_file()
def test_setup_component_and_test_service_with_config_language(self):
async def test_setup_component_and_test_service_with_config_language(
hass, empty_cache_dir
):
"""Set up the demo platform and call service."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo", "language": "de"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "There is someone at the door.",
},
blocking=True,
)
self.hass.block_till_done()
assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3".format(
self.hass.config.api.base_url
)
assert os.path.isfile(
os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3",
)
hass.config.api.base_url
)
assert (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3"
).is_file()
def test_setup_component_and_test_service_with_wrong_conf_language(self):
async def test_setup_component_and_test_service_with_wrong_conf_language(hass):
"""Set up the demo platform and call service with wrong config."""
config = {tts.DOMAIN: {"platform": "demo", "language": "ru"}}
with assert_setup_component(0, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
def test_setup_component_and_test_service_with_service_language(self):
async def test_setup_component_and_test_service_with_service_language(
hass, empty_cache_dir
):
"""Set up the demo platform and call service."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
@ -173,33 +185,32 @@ class TestTTS:
tts.ATTR_MESSAGE: "There is someone at the door.",
tts.ATTR_LANGUAGE: "de",
},
blocking=True,
)
self.hass.block_till_done()
assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3".format(
self.hass.config.api.base_url
)
assert os.path.isfile(
os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3",
)
hass.config.api.base_url
)
assert (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3"
).is_file()
def test_setup_component_test_service_with_wrong_service_language(self):
async def test_setup_component_test_service_with_wrong_service_language(
hass, empty_cache_dir
):
"""Set up the demo platform and call service."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
@ -207,27 +218,26 @@ class TestTTS:
tts.ATTR_MESSAGE: "There is someone at the door.",
tts.ATTR_LANGUAGE: "lang",
},
blocking=True,
)
self.hass.block_till_done()
assert len(calls) == 0
assert not os.path.isfile(
os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_lang_-_demo.mp3",
)
)
assert not (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_lang_-_demo.mp3"
).is_file()
def test_setup_component_and_test_service_with_service_options(self):
async def test_setup_component_and_test_service_with_service_options(
hass, empty_cache_dir
):
"""Set up the demo platform and call service with options."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
@ -236,9 +246,8 @@ class TestTTS:
tts.ATTR_LANGUAGE: "de",
tts.ATTR_OPTIONS: {"voice": "alex"},
},
blocking=True,
)
self.hass.block_till_done()
opt_hash = ctypes.c_size_t(hash(frozenset({"voice": "alex"}))).value
assert len(calls) == 1
@ -246,31 +255,27 @@ class TestTTS:
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{}_demo.mp3".format(
self.hass.config.api.base_url, opt_hash
)
assert os.path.isfile(
os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{0}_demo.mp3".format(
opt_hash
),
)
hass.config.api.base_url, opt_hash
)
assert (
empty_cache_dir
/ f"42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{opt_hash}_demo.mp3"
).is_file()
@patch(
"homeassistant.components.demo.tts.DemoProvider.default_options",
new_callable=PropertyMock(return_value={"voice": "alex"}),
)
def test_setup_component_and_test_with_service_options_def(self, def_mock):
async def test_setup_component_and_test_with_service_options_def(hass, empty_cache_dir):
"""Set up the demo platform and call service with default options."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
with assert_setup_component(1, tts.DOMAIN), patch(
"homeassistant.components.demo.tts.DemoProvider.default_options",
new_callable=PropertyMock(return_value={"voice": "alex"}),
):
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
@ -278,9 +283,8 @@ class TestTTS:
tts.ATTR_MESSAGE: "There is someone at the door.",
tts.ATTR_LANGUAGE: "de",
},
blocking=True,
)
self.hass.block_till_done()
opt_hash = ctypes.c_size_t(hash(frozenset({"voice": "alex"}))).value
assert len(calls) == 1
@ -288,27 +292,30 @@ class TestTTS:
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{}_demo.mp3".format(
self.hass.config.api.base_url, opt_hash
hass.config.api.base_url, opt_hash
)
assert os.path.isfile(
os.path.join(
self.default_tts_cache,
empty_cache_dir,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{0}_demo.mp3".format(
opt_hash
),
)
)
def test_setup_component_and_test_service_with_service_options_wrong(self):
async def test_setup_component_and_test_service_with_service_options_wrong(
hass, empty_cache_dir
):
"""Set up the demo platform and call service with wrong options."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
@ -317,40 +324,35 @@ class TestTTS:
tts.ATTR_LANGUAGE: "de",
tts.ATTR_OPTIONS: {"speed": 1},
},
blocking=True,
)
self.hass.block_till_done()
opt_hash = ctypes.c_size_t(hash(frozenset({"speed": 1}))).value
assert len(calls) == 0
assert not os.path.isfile(
os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{0}_demo.mp3".format(
opt_hash
),
)
)
assert not (
empty_cache_dir
/ f"42f18378fd4393d18c8dd11d03fa9563c1e54491_de_{opt_hash}_demo.mp3"
).is_file()
def test_setup_component_and_test_service_with_base_url_set(self):
async def test_setup_component_and_test_service_with_base_url_set(hass):
"""Set up the demo platform with ``base_url`` set and call service."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo", "base_url": "http://fnord"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "There is someone at the door.",
},
blocking=True,
)
self.hass.block_till_done()
assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
assert (
@ -359,182 +361,182 @@ class TestTTS:
"_en_-_demo.mp3"
)
def test_setup_component_and_test_service_clear_cache(self):
async def test_setup_component_and_test_service_clear_cache(hass, empty_cache_dir):
"""Set up the demo platform and call service clear cache."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "There is someone at the door.",
},
blocking=True,
)
self.hass.block_till_done()
# To make sure the file is persisted
await hass.async_block_till_done()
assert len(calls) == 1
assert os.path.isfile(
os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3",
)
assert (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
).is_file()
await hass.services.async_call(
tts.DOMAIN, tts.SERVICE_CLEAR_CACHE, {}, blocking=True
)
self.hass.services.call(tts.DOMAIN, tts.SERVICE_CLEAR_CACHE, {})
self.hass.block_till_done()
assert not (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
).is_file()
assert not os.path.isfile(
os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3",
)
)
def test_setup_component_and_test_service_with_receive_voice(self):
async def test_setup_component_and_test_service_with_receive_voice(
hass, demo_provider, hass_client
):
"""Set up the demo platform and call service and receive voice."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.start()
client = await hass_client()
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "There is someone at the door.",
},
blocking=True,
)
self.hass.block_till_done()
assert len(calls) == 1
req = requests.get(calls[0].data[ATTR_MEDIA_CONTENT_ID])
_, demo_data = self.demo_provider.get_tts_audio("bla", "en")
req = await client.get(relative_url(calls[0].data[ATTR_MEDIA_CONTENT_ID]))
_, demo_data = demo_provider.get_tts_audio("bla", "en")
demo_data = tts.SpeechManager.write_tags(
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3",
demo_data,
self.demo_provider,
demo_provider,
"AI person is in front of your door.",
"en",
None,
)
assert req.status_code == 200
assert req.content == demo_data
assert req.status == 200
assert await req.read() == demo_data
def test_setup_component_and_test_service_with_receive_voice_german(self):
async def test_setup_component_and_test_service_with_receive_voice_german(
hass, demo_provider, hass_client
):
"""Set up the demo platform and call service and receive voice."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo", "language": "de"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.start()
client = await hass_client()
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "There is someone at the door.",
},
blocking=True,
)
self.hass.block_till_done()
assert len(calls) == 1
req = requests.get(calls[0].data[ATTR_MEDIA_CONTENT_ID])
_, demo_data = self.demo_provider.get_tts_audio("bla", "de")
req = await client.get(relative_url(calls[0].data[ATTR_MEDIA_CONTENT_ID]))
_, demo_data = demo_provider.get_tts_audio("bla", "de")
demo_data = tts.SpeechManager.write_tags(
"42f18378fd4393d18c8dd11d03fa9563c1e54491_de_-_demo.mp3",
demo_data,
self.demo_provider,
demo_provider,
"There is someone at the door.",
"de",
None,
)
assert req.status_code == 200
assert req.content == demo_data
assert req.status == 200
assert await req.read() == demo_data
def test_setup_component_and_web_view_wrong_file(self):
async def test_setup_component_and_web_view_wrong_file(hass, hass_client):
"""Set up the demo platform and receive wrong file from web."""
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.start()
client = await hass_client()
url = (
"{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
).format(self.hass.config.api.base_url)
url = "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
req = requests.get(url)
assert req.status_code == 404
req = await client.get(url)
assert req.status == 404
def test_setup_component_and_web_view_wrong_filename(self):
async def test_setup_component_and_web_view_wrong_filename(hass, hass_client):
"""Set up the demo platform and receive wrong filename from web."""
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.start()
client = await hass_client()
url = (
"{}/api/tts_proxy/265944dsk32c1b2a621be5930510bb2cd_en_-_demo.mp3"
).format(self.hass.config.api.base_url)
url = "/api/tts_proxy/265944dsk32c1b2a621be5930510bb2cd_en_-_demo.mp3"
req = requests.get(url)
assert req.status_code == 404
req = await client.get(url)
assert req.status == 404
def test_setup_component_test_without_cache(self):
async def test_setup_component_test_without_cache(hass, empty_cache_dir):
"""Set up demo platform without cache."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo", "cache": False}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "There is someone at the door.",
},
blocking=True,
)
self.hass.block_till_done()
assert len(calls) == 1
assert not os.path.isfile(
os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3",
)
)
assert not (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
).is_file()
def test_setup_component_test_with_cache_call_service_without_cache(self):
async def test_setup_component_test_with_cache_call_service_without_cache(
hass, empty_cache_dir
):
"""Set up demo platform with cache and call service without cache."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo", "cache": True}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
@ -542,108 +544,89 @@ class TestTTS:
tts.ATTR_MESSAGE: "There is someone at the door.",
tts.ATTR_CACHE: False,
},
blocking=True,
)
self.hass.block_till_done()
assert len(calls) == 1
assert not os.path.isfile(
os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3",
)
)
assert not (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
).is_file()
def test_setup_component_test_with_cache_dir(self):
async def test_setup_component_test_with_cache_dir(
hass, empty_cache_dir, demo_provider
):
"""Set up demo platform with cache and call service without cache."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
_, demo_data = self.demo_provider.get_tts_audio("bla", "en")
cache_file = os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3",
_, demo_data = demo_provider.get_tts_audio("bla", "en")
cache_file = (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
)
os.mkdir(self.default_tts_cache)
with open(cache_file, "wb") as voice_file:
voice_file.write(demo_data)
config = {tts.DOMAIN: {"platform": "demo", "cache": True}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
with patch(
"homeassistant.components.demo.tts.DemoProvider.get_tts_audio",
return_value=(None, None),
):
self.hass.services.call(
await hass.services.async_call(
tts.DOMAIN,
"demo_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "There is someone at the door.",
},
blocking=True,
)
self.hass.block_till_done()
assert len(calls) == 1
assert calls[0].data[
ATTR_MEDIA_CONTENT_ID
] == "{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3".format(
self.hass.config.api.base_url
hass.config.api.base_url
)
@patch(
"homeassistant.components.demo.tts.DemoProvider.get_tts_audio",
return_value=(None, None),
)
def test_setup_component_test_with_error_on_get_tts(self, tts_mock):
async def test_setup_component_test_with_error_on_get_tts(hass):
"""Set up demo platform with wrong get_tts_audio."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "demo"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
with assert_setup_component(1, tts.DOMAIN), patch(
"homeassistant.components.demo.tts.DemoProvider.get_tts_audio",
return_value=(None, None),
):
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.services.call(
tts.DOMAIN,
"demo_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "There is someone at the door.",
},
)
self.hass.block_till_done()
assert len(calls) == 0
def test_setup_component_load_cache_retrieve_without_mem_cache(self):
async def test_setup_component_load_cache_retrieve_without_mem_cache(
hass, demo_provider, empty_cache_dir, hass_client
):
"""Set up component and load cache and get without mem cache."""
_, demo_data = self.demo_provider.get_tts_audio("bla", "en")
cache_file = os.path.join(
self.default_tts_cache,
"42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3",
_, demo_data = demo_provider.get_tts_audio("bla", "en")
cache_file = (
empty_cache_dir / "42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
)
os.mkdir(self.default_tts_cache)
with open(cache_file, "wb") as voice_file:
voice_file.write(demo_data)
config = {tts.DOMAIN: {"platform": "demo", "cache": True}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
assert await async_setup_component(hass, tts.DOMAIN, config)
self.hass.start()
client = await hass_client()
url = (
"{}/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
).format(self.hass.config.api.base_url)
url = "/api/tts_proxy/42f18378fd4393d18c8dd11d03fa9563c1e54491_en_-_demo.mp3"
req = requests.get(url)
assert req.status_code == 200
assert req.content == demo_data
req = await client.get(url)
assert req.status == 200
assert await req.read() == demo_data
async def test_setup_component_and_web_get_url(hass, hass_client):
@ -666,10 +649,6 @@ async def test_setup_component_and_web_get_url(hass, hass_client):
)
)
tts_cache = hass.config.path(tts.DEFAULT_CACHE_DIR)
if os.path.isdir(tts_cache):
shutil.rmtree(tts_cache)
async def test_setup_component_and_web_get_url_bad_config(hass, hass_client):
"""Set up the demo platform and receive wrong file from web."""