jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 11:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Convert remaining TTS tests to async (#64505)

Browse Source 
* Convert remaining TTS tests to async

* Add block till done after setting up component
This commit is contained in:
Paulus Schoutsen 2022-01-20 08:58:19 -08:00 committed by GitHub
parent 8fda3ae4cb
commit ddf548cd27
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 611 additions and 639 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

203
tests/components/marytts/test_tts.py
View File

@ -1,143 +1,124 @@
"""The tests for the MaryTTS speech platform.""" """The tests for the MaryTTS speech platform."""
import asyncio
import os import os
import shutil import shutil
from unittest.mock import patch from unittest.mock import patch
import pytest
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 import homeassistant.components.tts as tts
from homeassistant.config import async_process_ha_core_config from homeassistant.setup import async_setup_component
from homeassistant.setup import setup_component
from tests.common import assert_setup_component, get_test_home_assistant, mock_service from tests.common import assert_setup_component, async_mock_service
class TestTTSMaryTTSPlatform: @pytest.fixture(autouse=True)
"""Test the speech component.""" def cleanup_cache(hass):
"""Prevent TTS writing."""
yield
default_tts = hass.config.path(tts.DEFAULT_CACHE_DIR)
if os.path.isdir(default_tts):
shutil.rmtree(default_tts)
def setup_method(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
asyncio.run_coroutine_threadsafe( async def test_setup_component(hass):
async_process_ha_core_config( """Test setup component."""
self.hass, {"internal_url": "http://example.local:8123"} config = {tts.DOMAIN: {"platform": "marytts"}}
),
self.hass.loop, with assert_setup_component(1, tts.DOMAIN):
await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
async def test_service_say(hass):
"""Test service call say."""
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "marytts"}}
with assert_setup_component(1, tts.DOMAIN):
await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
with patch(
"homeassistant.components.marytts.tts.MaryTTS.speak",
return_value=b"audio",
) as mock_speak:
await hass.services.async_call(
tts.DOMAIN,
"marytts_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "HomeAssistant",
},
blocking=True,
) )
self.host = "localhost" mock_speak.assert_called_once()
self.port = 59125 mock_speak.assert_called_with("HomeAssistant", {})
self.params = {
"INPUT_TEXT": "HomeAssistant",
"INPUT_TYPE": "TEXT",
"OUTPUT_TYPE": "AUDIO",
"LOCALE": "en_US",
"AUDIO": "WAVE_FILE",
"VOICE": "cmu-slt-hsmm",
}
def teardown_method(self): assert len(calls) == 1
"""Stop everything that was started.""" assert calls[0].data[ATTR_MEDIA_CONTENT_ID].find(".wav") != -1
default_tts = self.hass.config.path(tts.DEFAULT_CACHE_DIR)
if os.path.isdir(default_tts):
shutil.rmtree(default_tts)
self.hass.stop()
def test_setup_component(self): async def test_service_say_with_effect(hass):
"""Test setup component.""" """Test service call say with effects."""
config = {tts.DOMAIN: {"platform": "marytts"}} calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
with assert_setup_component(1, tts.DOMAIN): config = {tts.DOMAIN: {"platform": "marytts", "effect": {"Volume": "amount:2.0;"}}}
setup_component(self.hass, tts.DOMAIN, config)
def test_service_say(self): with assert_setup_component(1, tts.DOMAIN):
"""Test service call say.""" await async_setup_component(hass, tts.DOMAIN, config)
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA) await hass.async_block_till_done()
config = {tts.DOMAIN: {"platform": "marytts"}} with patch(
"homeassistant.components.marytts.tts.MaryTTS.speak",
return_value=b"audio",
) as mock_speak:
await hass.services.async_call(
tts.DOMAIN,
"marytts_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "HomeAssistant",
},
blocking=True,
)
with assert_setup_component(1, tts.DOMAIN): mock_speak.assert_called_once()
setup_component(self.hass, tts.DOMAIN, config) mock_speak.assert_called_with("HomeAssistant", {"Volume": "amount:2.0;"})
with patch( assert len(calls) == 1
"homeassistant.components.marytts.tts.MaryTTS.speak", assert calls[0].data[ATTR_MEDIA_CONTENT_ID].find(".wav") != -1
return_value=b"audio",
) as mock_speak:
self.hass.services.call(
tts.DOMAIN,
"marytts_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "HomeAssistant",
},
)
self.hass.block_till_done()
mock_speak.assert_called_once()
mock_speak.assert_called_with("HomeAssistant", {})
assert len(calls) == 1 async def test_service_say_http_error(hass):
assert calls[0].data[ATTR_MEDIA_CONTENT_ID].find(".wav") != -1 """Test service call say."""
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
def test_service_say_with_effect(self): config = {tts.DOMAIN: {"platform": "marytts"}}
"""Test service call say with effects."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = { with assert_setup_component(1, tts.DOMAIN):
tts.DOMAIN: {"platform": "marytts", "effect": {"Volume": "amount:2.0;"}} await async_setup_component(hass, tts.DOMAIN, config)
} await hass.async_block_till_done()
with assert_setup_component(1, tts.DOMAIN): with patch(
setup_component(self.hass, tts.DOMAIN, config) "homeassistant.components.marytts.tts.MaryTTS.speak",
side_effect=Exception(),
) as mock_speak:
await hass.services.async_call(
tts.DOMAIN,
"marytts_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "HomeAssistant",
},
)
await hass.async_block_till_done()
with patch( mock_speak.assert_called_once()
"homeassistant.components.marytts.tts.MaryTTS.speak", assert len(calls) == 0
return_value=b"audio",
) as mock_speak:
self.hass.services.call(
tts.DOMAIN,
"marytts_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "HomeAssistant",
},
)
self.hass.block_till_done()
mock_speak.assert_called_once()
mock_speak.assert_called_with("HomeAssistant", {"Volume": "amount:2.0;"})
assert len(calls) == 1
assert calls[0].data[ATTR_MEDIA_CONTENT_ID].find(".wav") != -1
def test_service_say_http_error(self):
"""Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "marytts"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
with patch(
"homeassistant.components.marytts.tts.MaryTTS.speak",
side_effect=Exception(),
) as mock_speak:
self.hass.services.call(
tts.DOMAIN,
"marytts_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "HomeAssistant",
},
)
self.hass.block_till_done()
mock_speak.assert_called_once()
assert len(calls) == 0

363
tests/components/voicerss/test_tts.py
View File

@ -4,231 +4,228 @@ from http import HTTPStatus
import os import os
import shutil import shutil
import pytest
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 import homeassistant.components.tts as tts
from homeassistant.config import async_process_ha_core_config from homeassistant.setup import async_setup_component
from homeassistant.setup import setup_component
from tests.common import assert_setup_component, get_test_home_assistant, mock_service from tests.common import assert_setup_component, async_mock_service
from tests.components.tts.test_init import mutagen_mock # noqa: F401 from tests.components.tts.test_init import mutagen_mock # noqa: F401
URL = "https://api.voicerss.org/"
FORM_DATA = {
"key": "1234567xx",
"hl": "en-us",
"c": "MP3",
"f": "8khz_8bit_mono",
"src": "I person is on front of your door.",
}
class TestTTSVoiceRSSPlatform:
"""Test the voicerss speech component."""
def setup_method(self): @pytest.fixture(autouse=True)
"""Set up things to be run when tests are started.""" def cleanup_cache(hass):
self.hass = get_test_home_assistant() """Prevent TTS writing."""
yield
default_tts = hass.config.path(tts.DEFAULT_CACHE_DIR)
if os.path.isdir(default_tts):
shutil.rmtree(default_tts)
asyncio.run_coroutine_threadsafe(
async_process_ha_core_config(
self.hass, {"internal_url": "http://example.local:8123"}
),
self.hass.loop,
)
self.url = "https://api.voicerss.org/" async def test_setup_component(hass):
self.form_data = { """Test setup component."""
"key": "1234567xx", config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}}
"hl": "en-us",
"c": "MP3", with assert_setup_component(1, tts.DOMAIN):
"f": "8khz_8bit_mono", await async_setup_component(hass, tts.DOMAIN, config)
"src": "I person is on front of your door.", await hass.async_block_till_done()
async def test_setup_component_without_api_key(hass):
"""Test setup component without api key."""
config = {tts.DOMAIN: {"platform": "voicerss"}}
with assert_setup_component(0, tts.DOMAIN):
await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
async def test_service_say(hass, aioclient_mock):
"""Test service call say."""
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
aioclient_mock.post(URL, data=FORM_DATA, status=HTTPStatus.OK, content=b"test")
config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN):
await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
await hass.services.async_call(
tts.DOMAIN,
"voicerss_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.",
},
)
await hass.async_block_till_done()
assert len(calls) == 1
assert len(aioclient_mock.mock_calls) == 1
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):
"""Test service call say with german code in the config."""
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
form_data = {**FORM_DATA, "hl": "de-de"}
aioclient_mock.post(URL, data=form_data, status=HTTPStatus.OK, content=b"test")
config = {
tts.DOMAIN: {
"platform": "voicerss",
"api_key": "1234567xx",
"language": "de-de",
} }
}
def teardown_method(self): with assert_setup_component(1, tts.DOMAIN):
"""Stop everything that was started.""" await async_setup_component(hass, tts.DOMAIN, config)
default_tts = self.hass.config.path(tts.DEFAULT_CACHE_DIR) await hass.async_block_till_done()
if os.path.isdir(default_tts):
shutil.rmtree(default_tts)
self.hass.stop() await hass.services.async_call(
tts.DOMAIN,
"voicerss_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.",
},
)
await hass.async_block_till_done()
def test_setup_component(self): assert len(calls) == 1
"""Test setup component.""" assert len(aioclient_mock.mock_calls) == 1
config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}} assert aioclient_mock.mock_calls[0][2] == form_data
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
def test_setup_component_without_api_key(self): async def test_service_say_german_service(hass, aioclient_mock):
"""Test setup component without api key.""" """Test service call say with german code in the service."""
config = {tts.DOMAIN: {"platform": "voicerss"}} calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
with assert_setup_component(0, tts.DOMAIN): form_data = {**FORM_DATA, "hl": "de-de"}
setup_component(self.hass, tts.DOMAIN, config) aioclient_mock.post(URL, data=form_data, status=HTTPStatus.OK, content=b"test")
def test_service_say(self, aioclient_mock): config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}}
"""Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
aioclient_mock.post( with assert_setup_component(1, tts.DOMAIN):
self.url, data=self.form_data, status=HTTPStatus.OK, content=b"test" await async_setup_component(hass, tts.DOMAIN, config)
) await hass.async_block_till_done()
config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}} await hass.services.async_call(
tts.DOMAIN,
"voicerss_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.",
tts.ATTR_LANGUAGE: "de-de",
},
)
await hass.async_block_till_done()
with assert_setup_component(1, tts.DOMAIN): assert len(calls) == 1
setup_component(self.hass, tts.DOMAIN, config) assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == form_data
self.hass.services.call(
tts.DOMAIN,
"voicerss_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.",
},
)
self.hass.block_till_done()
assert len(calls) == 1 async def test_service_say_error(hass, aioclient_mock):
assert len(aioclient_mock.mock_calls) == 1 """Test service call say with http response 400."""
assert aioclient_mock.mock_calls[0][2] == self.form_data calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
assert calls[0].data[ATTR_MEDIA_CONTENT_ID].find(".mp3") != -1
def test_service_say_german_config(self, aioclient_mock): aioclient_mock.post(URL, data=FORM_DATA, status=400, content=b"test")
"""Test service call say with german code in the config."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
self.form_data["hl"] = "de-de" config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}}
aioclient_mock.post(
self.url, data=self.form_data, status=HTTPStatus.OK, content=b"test"
)
config = { with assert_setup_component(1, tts.DOMAIN):
tts.DOMAIN: { await async_setup_component(hass, tts.DOMAIN, config)
"platform": "voicerss", await hass.async_block_till_done()
"api_key": "1234567xx",
"language": "de-de",
}
}
with assert_setup_component(1, tts.DOMAIN): await hass.services.async_call(
setup_component(self.hass, tts.DOMAIN, config) tts.DOMAIN,
"voicerss_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.",
},
)
await hass.async_block_till_done()
self.hass.services.call( assert len(calls) == 0
tts.DOMAIN, assert len(aioclient_mock.mock_calls) == 1
"voicerss_say", assert aioclient_mock.mock_calls[0][2] == FORM_DATA
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.",
},
)
self.hass.block_till_done()
assert len(calls) == 1
assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == self.form_data
def test_service_say_german_service(self, aioclient_mock): async def test_service_say_timeout(hass, aioclient_mock):
"""Test service call say with german code in the service.""" """Test service call say with http timeout."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA) calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
self.form_data["hl"] = "de-de" aioclient_mock.post(URL, data=FORM_DATA, exc=asyncio.TimeoutError())
aioclient_mock.post(
self.url, data=self.form_data, status=HTTPStatus.OK, content=b"test"
)
config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}} config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN): with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config) await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
self.hass.services.call( await hass.services.async_call(
tts.DOMAIN, tts.DOMAIN,
"voicerss_say", "voicerss_say",
{ {
"entity_id": "media_player.something", "entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.", tts.ATTR_MESSAGE: "I person is on front of your door.",
tts.ATTR_LANGUAGE: "de-de", },
}, )
) await hass.async_block_till_done()
self.hass.block_till_done()
assert len(calls) == 1 assert len(calls) == 0
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == self.form_data assert aioclient_mock.mock_calls[0][2] == FORM_DATA
def test_service_say_error(self, aioclient_mock):
"""Test service call say with http response 400."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
aioclient_mock.post(self.url, data=self.form_data, status=400, content=b"test") async def test_service_say_error_msg(hass, aioclient_mock):
"""Test service call say with http error api message."""
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}} aioclient_mock.post(
URL,
data=FORM_DATA,
status=HTTPStatus.OK,
content=b"The subscription does not support SSML!",
)
with assert_setup_component(1, tts.DOMAIN): config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}}
setup_component(self.hass, tts.DOMAIN, config)
self.hass.services.call( with assert_setup_component(1, tts.DOMAIN):
tts.DOMAIN, await async_setup_component(hass, tts.DOMAIN, config)
"voicerss_say", await hass.async_block_till_done()
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.",
},
)
self.hass.block_till_done()
assert len(calls) == 0 await hass.services.async_call(
assert len(aioclient_mock.mock_calls) == 1 tts.DOMAIN,
assert aioclient_mock.mock_calls[0][2] == self.form_data "voicerss_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.",
},
)
await hass.async_block_till_done()
def test_service_say_timeout(self, aioclient_mock): assert len(calls) == 0
"""Test service call say with http timeout.""" assert len(aioclient_mock.mock_calls) == 1
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA) assert aioclient_mock.mock_calls[0][2] == FORM_DATA
aioclient_mock.post(self.url, data=self.form_data, exc=asyncio.TimeoutError())
config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
self.hass.services.call(
tts.DOMAIN,
"voicerss_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.",
},
)
self.hass.block_till_done()
assert len(calls) == 0
assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == self.form_data
def test_service_say_error_msg(self, aioclient_mock):
"""Test service call say with http error api message."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
aioclient_mock.post(
self.url,
data=self.form_data,
status=HTTPStatus.OK,
content=b"The subscription does not support SSML!",
)
config = {tts.DOMAIN: {"platform": "voicerss", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
self.hass.services.call(
tts.DOMAIN,
"voicerss_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "I person is on front of your door.",
},
)
self.hass.block_till_done()
assert len(calls) == 0
assert len(aioclient_mock.mock_calls) == 1
assert aioclient_mock.mock_calls[0][2] == self.form_data

684
tests/components/yandextts/test_tts.py
View File

@ -4,407 +4,401 @@ from http import HTTPStatus
import os import os
import shutil import shutil
import pytest
from homeassistant.components.media_player.const import ( from homeassistant.components.media_player.const import (
DOMAIN as DOMAIN_MP, DOMAIN as DOMAIN_MP,
SERVICE_PLAY_MEDIA, SERVICE_PLAY_MEDIA,
) )
import homeassistant.components.tts as tts import homeassistant.components.tts as tts
from homeassistant.config import async_process_ha_core_config from homeassistant.setup import async_setup_component
from homeassistant.setup import setup_component
from tests.common import assert_setup_component, get_test_home_assistant, mock_service from tests.common import assert_setup_component, async_mock_service
from tests.components.tts.test_init import ( # noqa: F401, pylint: disable=unused-import from tests.components.tts.test_init import ( # noqa: F401, pylint: disable=unused-import
mutagen_mock, mutagen_mock,
) )
URL = "https://tts.voicetech.yandex.net/generate?"
class TestTTSYandexPlatform:
"""Test the speech component."""
def setup_method(self): @pytest.fixture(autouse=True)
"""Set up things to be run when tests are started.""" def cleanup_cache(hass):
self.hass = get_test_home_assistant() """Prevent TTS writing."""
self._base_url = "https://tts.voicetech.yandex.net/generate?" yield
default_tts = hass.config.path(tts.DEFAULT_CACHE_DIR)
if os.path.isdir(default_tts):
shutil.rmtree(default_tts)
asyncio.run_coroutine_threadsafe(
async_process_ha_core_config(
self.hass, {"internal_url": "http://example.local:8123"}
),
self.hass.loop,
)
def teardown_method(self): async def test_setup_component(hass):
"""Stop everything that was started.""" """Test setup component."""
default_tts = self.hass.config.path(tts.DEFAULT_CACHE_DIR) config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
if os.path.isdir(default_tts):
shutil.rmtree(default_tts)
self.hass.stop() with assert_setup_component(1, tts.DOMAIN):
await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
def test_setup_component(self):
"""Test setup component."""
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN): async def test_setup_component_without_api_key(hass):
setup_component(self.hass, tts.DOMAIN, config) """Test setup component without api key."""
config = {tts.DOMAIN: {"platform": "yandextts"}}
def test_setup_component_without_api_key(self): with assert_setup_component(0, tts.DOMAIN):
"""Test setup component without api key.""" await async_setup_component(hass, tts.DOMAIN, config)
config = {tts.DOMAIN: {"platform": "yandextts"}} await hass.async_block_till_done()
with assert_setup_component(0, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
def test_service_say(self, aioclient_mock): async def test_service_say(hass, aioclient_mock):
"""Test service call say.""" """Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA) calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = { url_param = {
"text": "HomeAssistant", "text": "HomeAssistant",
"lang": "en-US", "lang": "en-US",
"key": "1234567xx", "key": "1234567xx",
"speaker": "zahar", "speaker": "zahar",
"format": "mp3", "format": "mp3",
"emotion": "neutral", "emotion": "neutral",
"speed": 1, "speed": 1,
}
aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN):
await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
await hass.services.async_call(
tts.DOMAIN,
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1
async def test_service_say_russian_config(hass, aioclient_mock):
"""Test service call say."""
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = {
"text": "HomeAssistant",
"lang": "ru-RU",
"key": "1234567xx",
"speaker": "zahar",
"format": "mp3",
"emotion": "neutral",
"speed": 1,
}
aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)
config = {
tts.DOMAIN: {
"platform": "yandextts",
"api_key": "1234567xx",
"language": "ru-RU",
} }
aioclient_mock.get( }
self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
)
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}} with assert_setup_component(1, tts.DOMAIN):
await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
with assert_setup_component(1, tts.DOMAIN): await hass.services.async_call(
setup_component(self.hass, tts.DOMAIN, config) tts.DOMAIN,
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
await hass.async_block_till_done()
self.hass.services.call( assert len(aioclient_mock.mock_calls) == 1
tts.DOMAIN, assert len(calls) == 1
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1
def test_service_say_russian_config(self, aioclient_mock): async def test_service_say_russian_service(hass, aioclient_mock):
"""Test service call say.""" """Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA) calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = { url_param = {
"text": "HomeAssistant", "text": "HomeAssistant",
"lang": "ru-RU", "lang": "ru-RU",
"key": "1234567xx", "key": "1234567xx",
"speaker": "zahar", "speaker": "zahar",
"format": "mp3", "format": "mp3",
"emotion": "neutral", "emotion": "neutral",
"speed": 1, "speed": 1,
}
aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN):
await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
await hass.services.async_call(
tts.DOMAIN,
"yandextts_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "HomeAssistant",
tts.ATTR_LANGUAGE: "ru-RU",
},
)
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1
async def test_service_say_timeout(hass, aioclient_mock):
"""Test service call say."""
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = {
"text": "HomeAssistant",
"lang": "en-US",
"key": "1234567xx",
"speaker": "zahar",
"format": "mp3",
"emotion": "neutral",
"speed": 1,
}
aioclient_mock.get(
URL,
status=HTTPStatus.OK,
exc=asyncio.TimeoutError(),
params=url_param,
)
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN):
await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
await hass.services.async_call(
tts.DOMAIN,
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
await hass.async_block_till_done()
assert len(calls) == 0
assert len(aioclient_mock.mock_calls) == 1
async def test_service_say_http_error(hass, aioclient_mock):
"""Test service call say."""
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = {
"text": "HomeAssistant",
"lang": "en-US",
"key": "1234567xx",
"speaker": "zahar",
"format": "mp3",
"emotion": "neutral",
"speed": 1,
}
aioclient_mock.get(
URL,
status=HTTPStatus.FORBIDDEN,
content=b"test",
params=url_param,
)
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN):
await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
await hass.services.async_call(
tts.DOMAIN,
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
await hass.async_block_till_done()
assert len(calls) == 0
async def test_service_say_specified_speaker(hass, aioclient_mock):
"""Test service call say."""
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = {
"text": "HomeAssistant",
"lang": "en-US",
"key": "1234567xx",
"speaker": "alyss",
"format": "mp3",
"emotion": "neutral",
"speed": 1,
}
aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)
config = {
tts.DOMAIN: {
"platform": "yandextts",
"api_key": "1234567xx",
"voice": "alyss",
} }
aioclient_mock.get( }
self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
)
config = { with assert_setup_component(1, tts.DOMAIN):
tts.DOMAIN: { await async_setup_component(hass, tts.DOMAIN, config)
"platform": "yandextts", await hass.async_block_till_done()
"api_key": "1234567xx",
"language": "ru-RU",
}
}
with assert_setup_component(1, tts.DOMAIN): await hass.services.async_call(
setup_component(self.hass, tts.DOMAIN, config) tts.DOMAIN,
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
await hass.async_block_till_done()
self.hass.services.call( assert len(aioclient_mock.mock_calls) == 1
tts.DOMAIN, assert len(calls) == 1
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1
def test_service_say_russian_service(self, aioclient_mock): async def test_service_say_specified_emotion(hass, aioclient_mock):
"""Test service call say.""" """Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA) calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = { url_param = {
"text": "HomeAssistant", "text": "HomeAssistant",
"lang": "ru-RU", "lang": "en-US",
"key": "1234567xx", "key": "1234567xx",
"speaker": "zahar", "speaker": "zahar",
"format": "mp3", "format": "mp3",
"emotion": "neutral", "emotion": "evil",
"speed": 1, "speed": 1,
} }
aioclient_mock.get( aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)
self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
)
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}} config = {
tts.DOMAIN: {
with assert_setup_component(1, tts.DOMAIN): "platform": "yandextts",
setup_component(self.hass, tts.DOMAIN, config) "api_key": "1234567xx",
self.hass.services.call(
tts.DOMAIN,
"yandextts_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "HomeAssistant",
tts.ATTR_LANGUAGE: "ru-RU",
},
)
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1
def test_service_say_timeout(self, aioclient_mock):
"""Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = {
"text": "HomeAssistant",
"lang": "en-US",
"key": "1234567xx",
"speaker": "zahar",
"format": "mp3",
"emotion": "neutral",
"speed": 1,
}
aioclient_mock.get(
self._base_url,
status=HTTPStatus.OK,
exc=asyncio.TimeoutError(),
params=url_param,
)
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
self.hass.services.call(
tts.DOMAIN,
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
self.hass.block_till_done()
assert len(calls) == 0
assert len(aioclient_mock.mock_calls) == 1
def test_service_say_http_error(self, aioclient_mock):
"""Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = {
"text": "HomeAssistant",
"lang": "en-US",
"key": "1234567xx",
"speaker": "zahar",
"format": "mp3",
"emotion": "neutral",
"speed": 1,
}
aioclient_mock.get(
self._base_url,
status=HTTPStatus.FORBIDDEN,
content=b"test",
params=url_param,
)
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
self.hass.services.call(
tts.DOMAIN,
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
self.hass.block_till_done()
assert len(calls) == 0
def test_service_say_specified_speaker(self, aioclient_mock):
"""Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = {
"text": "HomeAssistant",
"lang": "en-US",
"key": "1234567xx",
"speaker": "alyss",
"format": "mp3",
"emotion": "neutral",
"speed": 1,
}
aioclient_mock.get(
self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
)
config = {
tts.DOMAIN: {
"platform": "yandextts",
"api_key": "1234567xx",
"voice": "alyss",
}
}
with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config)
self.hass.services.call(
tts.DOMAIN,
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1
def test_service_say_specified_emotion(self, aioclient_mock):
"""Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = {
"text": "HomeAssistant",
"lang": "en-US",
"key": "1234567xx",
"speaker": "zahar",
"format": "mp3",
"emotion": "evil", "emotion": "evil",
"speed": 1,
} }
aioclient_mock.get( }
self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
)
config = { with assert_setup_component(1, tts.DOMAIN):
tts.DOMAIN: { await async_setup_component(hass, tts.DOMAIN, config)
"platform": "yandextts", await hass.async_block_till_done()
"api_key": "1234567xx",
"emotion": "evil",
}
}
with assert_setup_component(1, tts.DOMAIN): await hass.services.async_call(
setup_component(self.hass, tts.DOMAIN, config) tts.DOMAIN,
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
await hass.async_block_till_done()
self.hass.services.call( assert len(aioclient_mock.mock_calls) == 1
tts.DOMAIN, assert len(calls) == 1
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1
def test_service_say_specified_low_speed(self, aioclient_mock): async def test_service_say_specified_low_speed(hass, aioclient_mock):
"""Test service call say.""" """Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA) calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = { url_param = {
"text": "HomeAssistant", "text": "HomeAssistant",
"lang": "en-US", "lang": "en-US",
"key": "1234567xx", "key": "1234567xx",
"speaker": "zahar", "speaker": "zahar",
"format": "mp3", "format": "mp3",
"emotion": "neutral", "emotion": "neutral",
"speed": "0.1", "speed": "0.1",
} }
aioclient_mock.get( aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)
self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
)
config = { config = {
tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx", "speed": 0.1} tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx", "speed": 0.1}
} }
with assert_setup_component(1, tts.DOMAIN): with assert_setup_component(1, tts.DOMAIN):
setup_component(self.hass, tts.DOMAIN, config) await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
self.hass.services.call( await hass.services.async_call(
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"},
) )
self.hass.block_till_done() await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1 assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1 assert len(calls) == 1
def test_service_say_specified_speed(self, aioclient_mock):
"""Test service call say."""
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = { async def test_service_say_specified_speed(hass, aioclient_mock):
"text": "HomeAssistant", """Test service call say."""
"lang": "en-US", calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
"key": "1234567xx",
"speaker": "zahar",
"format": "mp3",
"emotion": "neutral",
"speed": 2,
}
aioclient_mock.get(
self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
)
config = { url_param = {
tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx", "speed": 2} "text": "HomeAssistant",
} "lang": "en-US",
"key": "1234567xx",
"speaker": "zahar",
"format": "mp3",
"emotion": "neutral",
"speed": 2,
}
aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)
with assert_setup_component(1, tts.DOMAIN): config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx", "speed": 2}}
setup_component(self.hass, tts.DOMAIN, config)
self.hass.services.call( with assert_setup_component(1, tts.DOMAIN):
tts.DOMAIN, await async_setup_component(hass, tts.DOMAIN, config)
"yandextts_say", await hass.async_block_till_done()
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 1 await hass.services.async_call(
assert len(calls) == 1 tts.DOMAIN,
"yandextts_say",
{"entity_id": "media_player.something", tts.ATTR_MESSAGE: "HomeAssistant"},
)
await hass.async_block_till_done()
def test_service_say_specified_options(self, aioclient_mock): assert len(aioclient_mock.mock_calls) == 1
"""Test service call say with options.""" assert len(calls) == 1
calls = mock_service(self.hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
url_param = {
"text": "HomeAssistant",
"lang": "en-US",
"key": "1234567xx",
"speaker": "zahar",
"format": "mp3",
"emotion": "evil",
"speed": 2,
}
aioclient_mock.get(
self._base_url, status=HTTPStatus.OK, content=b"test", params=url_param
)
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
with assert_setup_component(1, tts.DOMAIN): async def test_service_say_specified_options(hass, aioclient_mock):
setup_component(self.hass, tts.DOMAIN, config) """Test service call say with options."""
calls = async_mock_service(hass, DOMAIN_MP, SERVICE_PLAY_MEDIA)
self.hass.services.call( url_param = {
tts.DOMAIN, "text": "HomeAssistant",
"yandextts_say", "lang": "en-US",
{ "key": "1234567xx",
"entity_id": "media_player.something", "speaker": "zahar",
tts.ATTR_MESSAGE: "HomeAssistant", "format": "mp3",
"options": {"emotion": "evil", "speed": 2}, "emotion": "evil",
}, "speed": 2,
) }
self.hass.block_till_done() aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)
config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx"}}
assert len(aioclient_mock.mock_calls) == 1 with assert_setup_component(1, tts.DOMAIN):
assert len(calls) == 1 await async_setup_component(hass, tts.DOMAIN, config)
await hass.async_block_till_done()
await hass.services.async_call(
tts.DOMAIN,
"yandextts_say",
{
"entity_id": "media_player.something",
tts.ATTR_MESSAGE: "HomeAssistant",
"options": {"emotion": "evil", "speed": 2},
},
)
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 1
assert len(calls) == 1