core/tests/components/yandextts/test_tts.py

"""The tests for the Yandex SpeechKit speech platform."""
import asyncio
from http import HTTPStatus
import os
import shutil

import pytest

from homeassistant.components import media_source, tts
from homeassistant.components.media_player import (
    ATTR_MEDIA_CONTENT_ID,
    DOMAIN as DOMAIN_MP,
    SERVICE_PLAY_MEDIA,
)
from homeassistant.setup import async_setup_component

from tests.common import assert_setup_component, async_mock_service
from tests.components.tts.conftest import (  # noqa: F401, pylint: disable=unused-import
    mutagen_mock,
)

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, None)
    return resolved.url


@pytest.fixture(autouse=True)
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)


async def test_setup_component(hass):
    """Test setup component."""
    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()


async def test_setup_component_without_api_key(hass):
    """Test setup component without api key."""
    config = {tts.DOMAIN: {"platform": "yandextts"}}

    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)

    url_param = {
        "text": "HomeAssistant",
        "lang": "en-US",
        "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"}}

    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"},
        blocking=True,
    )
    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):
    """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",
        }
    }

    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"},
        blocking=True,
    )

    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):
    """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"}}

    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",
        },
        blocking=True,
    )
    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):
    """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"},
        blocking=True,
    )
    await hass.async_block_till_done()

    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


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"},
        blocking=True,
    )

    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):
    """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",
        }
    }

    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"},
        blocking=True,
    )
    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):
    """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": "evil",
        "speed": 1,
    }
    aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)

    config = {
        tts.DOMAIN: {
            "platform": "yandextts",
            "api_key": "1234567xx",
            "emotion": "evil",
        }
    }

    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"},
        blocking=True,
    )
    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_low_speed(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": "0.1",
    }
    aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)

    config = {
        tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx", "speed": 0.1}
    }

    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"},
        blocking=True,
    )
    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_speed(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": 2,
    }
    aioclient_mock.get(URL, status=HTTPStatus.OK, content=b"test", params=url_param)

    config = {tts.DOMAIN: {"platform": "yandextts", "api_key": "1234567xx", "speed": 2}}

    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"},
        blocking=True,
    )
    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_options(hass, aioclient_mock):
    """Test service call say with options."""
    calls = async_mock_service(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(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",
            "options": {"emotion": "evil", "speed": 2},
        },
        blocking=True,
    )
    assert len(calls) == 1
    await get_media_source_url(hass, calls[0].data[ATTR_MEDIA_CONTENT_ID])

    assert len(aioclient_mock.mock_calls) == 1