Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Setup Google Cloud from the UI (#121502) 

* Google Cloud can now be setup from the UI

* mypy

* Add BaseGoogleCloudProvider

* Allow clearing options in the UI

* Address feedback

* Don't translate Google Cloud title

* mypy

* Revert strict typing changes

* Address comments
pull/120854/head
tronikos 2024-09-02 04:30:18 -07:00 committed by GitHub
parent f4a16c8dc9
commit d40e3145fe
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 696 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CODEOWNERS
View File

@ -549,7 +549,8 @@ build.json @home-assistant/supervisor
/tests/components/google_assistant/ @home-assistant/cloud
/homeassistant/components/google_assistant_sdk/ @tronikos
/tests/components/google_assistant_sdk/ @tronikos
/homeassistant/components/google_cloud/ @lufton
/homeassistant/components/google_cloud/ @lufton @tronikos
/tests/components/google_cloud/ @lufton @tronikos
/homeassistant/components/google_generative_ai_conversation/ @tronikos
/tests/components/google_generative_ai_conversation/ @tronikos
/homeassistant/components/google_mail/ @tkdrob

25
homeassistant/components/google_cloud/__init__.py
View File

@ -1 +1,26 @@
"""The google_cloud component."""
from __future__ import annotations
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
PLATFORMS = [Platform.TTS]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(async_update_options))
return True
async def async_update_options(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

169
homeassistant/components/google_cloud/config_flow.py Normal file
View File

@ -0,0 +1,169 @@
"""Config flow for the Google Cloud integration."""
from __future__ import annotations
import json
import logging
from typing import TYPE_CHECKING, Any, cast
from google.cloud import texttospeech
import voluptuous as vol
from homeassistant.components.file_upload import process_uploaded_file
from homeassistant.components.tts import CONF_LANG
from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlowWithConfigEntry,
)
from homeassistant.core import callback
from homeassistant.helpers.selector import (
FileSelector,
FileSelectorConfig,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .const import CONF_KEY_FILE, CONF_SERVICE_ACCOUNT_INFO, DEFAULT_LANG, DOMAIN, TITLE
from .helpers import (
async_tts_voices,
tts_options_schema,
tts_platform_schema,
validate_service_account_info,
)
_LOGGER = logging.getLogger(__name__)
UPLOADED_KEY_FILE = "uploaded_key_file"
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(UPLOADED_KEY_FILE): FileSelector(
FileSelectorConfig(accept=".json,application/json")
)
}
)
class GoogleCloudConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Google Cloud integration."""
VERSION = 1
_name: str | None = None
entry: ConfigEntry | None = None
abort_reason: str | None = None
def _parse_uploaded_file(self, uploaded_file_id: str) -> dict[str, Any]:
"""Read and parse an uploaded JSON file."""
with process_uploaded_file(self.hass, uploaded_file_id) as file_path:
contents = file_path.read_text()
return cast(dict[str, Any], json.loads(contents))
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, Any] = {}
if user_input is not None:
try:
service_account_info = await self.hass.async_add_executor_job(
self._parse_uploaded_file, user_input[UPLOADED_KEY_FILE]
)
validate_service_account_info(service_account_info)
except ValueError:
_LOGGER.exception("Reading uploaded JSON file failed")
errors["base"] = "invalid_file"
else:
data = {CONF_SERVICE_ACCOUNT_INFO: service_account_info}
if self.entry:
if TYPE_CHECKING:
assert self.abort_reason
return self.async_update_reload_and_abort(
self.entry, data=data, reason=self.abort_reason
)
return self.async_create_entry(title=TITLE, data=data)
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors=errors,
description_placeholders={
"url": "https://console.cloud.google.com/apis/credentials/serviceaccountkey"
},
)
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
"""Import Google Cloud configuration from YAML."""
def _read_key_file() -> dict[str, Any]:
with open(
self.hass.config.path(import_data[CONF_KEY_FILE]), encoding="utf8"
) as f:
return cast(dict[str, Any], json.load(f))
service_account_info = await self.hass.async_add_executor_job(_read_key_file)
try:
validate_service_account_info(service_account_info)
except ValueError:
_LOGGER.exception("Reading credentials JSON file failed")
return self.async_abort(reason="invalid_file")
options = {
k: v for k, v in import_data.items() if k in tts_platform_schema().schema
}
options.pop(CONF_KEY_FILE)
_LOGGER.debug("Creating imported config entry with options: %s", options)
return self.async_create_entry(
title=TITLE,
data={CONF_SERVICE_ACCOUNT_INFO: service_account_info},
options=options,
)
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> GoogleCloudOptionsFlowHandler:
"""Create the options flow."""
return GoogleCloudOptionsFlowHandler(config_entry)
class GoogleCloudOptionsFlowHandler(OptionsFlowWithConfigEntry):
"""Google Cloud options flow."""
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(data=user_input)
service_account_info = self.config_entry.data[CONF_SERVICE_ACCOUNT_INFO]
client: texttospeech.TextToSpeechAsyncClient = (
texttospeech.TextToSpeechAsyncClient.from_service_account_info(
service_account_info
)
)
voices = await async_tts_voices(client)
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Optional(
CONF_LANG,
default=DEFAULT_LANG,
): SelectSelector(
SelectSelectorConfig(
mode=SelectSelectorMode.DROPDOWN, options=list(voices)
)
),
**tts_options_schema(
self.options, voices, from_config_flow=True
).schema,
}
),
self.options,
),
)

4
homeassistant/components/google_cloud/const.py
View File

@ -2,6 +2,10 @@
from __future__ import annotations
DOMAIN = "google_cloud"
TITLE = "Google Cloud"
CONF_SERVICE_ACCOUNT_INFO = "service_account_info"
CONF_KEY_FILE = "key_file"
DEFAULT_LANG = "en-US"

44
homeassistant/components/google_cloud/helpers.py
View File

@ -2,11 +2,13 @@
from __future__ import annotations
from collections.abc import Mapping
import functools
import operator
from typing import Any
from google.cloud import texttospeech
from google.oauth2.service_account import Credentials
import voluptuous as vol
from homeassistant.components.tts import CONF_LANG
@ -52,14 +54,18 @@ async def async_tts_voices(
def tts_options_schema(
config_options: dict[str, Any],
voices: dict[str, list[str]],
from_config_flow: bool = False,
) -> vol.Schema:
"""Return schema for TTS options with default values from config or constants."""
# If we are called from the config flow we want the defaults to be from constants
# to allow clearing the current value (passed as suggested_value) in the UI.
# If we aren't called from the config flow we want the defaults to be from the config.
defaults = {} if from_config_flow else config_options
return vol.Schema(
{
vol.Optional(
CONF_GENDER,
description={"suggested_value": config_options.get(CONF_GENDER)},
default=config_options.get(
default=defaults.get(
CONF_GENDER,
texttospeech.SsmlVoiceGender.NEUTRAL.name, # type: ignore[attr-defined]
),
@ -74,8 +80,7 @@ def tts_options_schema(
),
vol.Optional(
CONF_VOICE,
description={"suggested_value": config_options.get(CONF_VOICE)},
default=config_options.get(CONF_VOICE, DEFAULT_VOICE),
default=defaults.get(CONF_VOICE, DEFAULT_VOICE),
): SelectSelector(
SelectSelectorConfig(
mode=SelectSelectorMode.DROPDOWN,
@ -84,8 +89,7 @@ def tts_options_schema(
),
vol.Optional(
CONF_ENCODING,
description={"suggested_value": config_options.get(CONF_ENCODING)},
default=config_options.get(
default=defaults.get(
CONF_ENCODING,
texttospeech.AudioEncoding.MP3.name, # type: ignore[attr-defined]
),
@ -100,23 +104,19 @@ def tts_options_schema(
),
vol.Optional(
CONF_SPEED,
description={"suggested_value": config_options.get(CONF_SPEED)},
default=config_options.get(CONF_SPEED, 1.0),
default=defaults.get(CONF_SPEED, 1.0),
): NumberSelector(NumberSelectorConfig(min=0.25, max=4.0, step=0.01)),
vol.Optional(
CONF_PITCH,
description={"suggested_value": config_options.get(CONF_PITCH)},
default=config_options.get(CONF_PITCH, 0),
default=defaults.get(CONF_PITCH, 0),
): NumberSelector(NumberSelectorConfig(min=-20.0, max=20.0, step=0.1)),
vol.Optional(
CONF_GAIN,
description={"suggested_value": config_options.get(CONF_GAIN)},
default=config_options.get(CONF_GAIN, 0),
default=defaults.get(CONF_GAIN, 0),
): NumberSelector(NumberSelectorConfig(min=-96.0, max=16.0, step=0.1)),
vol.Optional(
CONF_PROFILES,
description={"suggested_value": config_options.get(CONF_PROFILES)},
default=config_options.get(CONF_PROFILES, []),
default=defaults.get(CONF_PROFILES, []),
): SelectSelector(
SelectSelectorConfig(
mode=SelectSelectorMode.DROPDOWN,
@ -137,8 +137,7 @@ def tts_options_schema(
),
vol.Optional(
CONF_TEXT_TYPE,
description={"suggested_value": config_options.get(CONF_TEXT_TYPE)},
default=config_options.get(CONF_TEXT_TYPE, "text"),
default=defaults.get(CONF_TEXT_TYPE, "text"),
): vol.All(
vol.Lower,
SelectSelector(
@ -166,3 +165,16 @@ def tts_platform_schema() -> vol.Schema:
),
}
)
def validate_service_account_info(info: Mapping[str, str]) -> None:
"""Validate service account info.
Args:
info: The service account info in Google format.
Raises:
ValueError: If the info is not in the expected format.
"""
Credentials.from_service_account_info(info) # type:ignore[no-untyped-call]

7
homeassistant/components/google_cloud/manifest.json
View File

@ -1,8 +1,11 @@
{
"domain": "google_cloud",
"name": "Google Cloud Platform",
"codeowners": ["@lufton"],
"name": "Google Cloud",
"codeowners": ["@lufton", "@tronikos"],
"config_flow": true,
"dependencies": ["file_upload"],
"documentation": "https://www.home-assistant.io/integrations/google_cloud",
"integration_type": "service",
"iot_class": "cloud_push",
"requirements": ["google-cloud-texttospeech==2.17.2"]
}

32
homeassistant/components/google_cloud/strings.json Normal file
View File

@ -0,0 +1,32 @@
{
"config": {
"step": {
"user": {
"description": "Upload your Google Cloud service account JSON file that you can create at {url}.",
"data": {
"uploaded_key_file": "Upload service account JSON file"
}
}
},
"error": {
"invalid_file": "Invalid service account JSON file"
}
},
"options": {
"step": {
"init": {
"data": {
"language": "Default language of the voice",
"gender": "Default gender of the voice",
"voice": "Default voice name (overrides language and gender)",
"encoding": "Default audio encoder",
"speed": "Default rate/speed of the voice",
"pitch": "Default pitch of the voice",
"gain": "Default volume gain (in dB) of the voice",
"profiles": "Default audio profiles",
"text_type": "Default text type"
}
}
}
}
}

134
homeassistant/components/google_cloud/tts.py
View File

@ -1,10 +1,12 @@
"""Support for the Google Cloud TTS service."""
from __future__ import annotations
import logging
import os
from pathlib import Path
from typing import Any, cast
from google.api_core.exceptions import GoogleAPIError
from google.api_core.exceptions import GoogleAPIError, Unauthenticated
from google.cloud import texttospeech
import voluptuous as vol
@ -12,10 +14,14 @@ from homeassistant.components.tts import (
CONF_LANG,
PLATFORM_SCHEMA as TTS_PLATFORM_SCHEMA,
Provider,
TextToSpeechEntity,
TtsAudioType,
Voice,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import (
@ -25,10 +31,12 @@ from .const import (
CONF_KEY_FILE,
CONF_PITCH,
CONF_PROFILES,
CONF_SERVICE_ACCOUNT_INFO,
CONF_SPEED,
CONF_TEXT_TYPE,
CONF_VOICE,
DEFAULT_LANG,
DOMAIN,
)
from .helpers import async_tts_voices, tts_options_schema, tts_platform_schema
@ -45,13 +53,20 @@ async def async_get_engine(
"""Set up Google Cloud TTS component."""
if key_file := config.get(CONF_KEY_FILE):
key_file = hass.config.path(key_file)
if not os.path.isfile(key_file):
if not Path(key_file).is_file():
_LOGGER.error("File %s doesn't exist", key_file)
return None
if key_file:
client = texttospeech.TextToSpeechAsyncClient.from_service_account_file(
key_file
)
if not hass.config_entries.async_entries(DOMAIN):
_LOGGER.debug("Creating config entry by importing: %s", config)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
)
)
else:
client = texttospeech.TextToSpeechAsyncClient()
try:
@ -60,7 +75,6 @@ async def async_get_engine(
_LOGGER.error("Error from calling list_voices: %s", err)
return None
return GoogleCloudTTSProvider(
hass,
client,
voices,
config.get(CONF_LANG, DEFAULT_LANG),
@ -68,20 +82,51 @@ async def async_get_engine(
)
class GoogleCloudTTSProvider(Provider):
"""The Google Cloud TTS API provider."""
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Google Cloud text-to-speech."""
service_account_info = config_entry.data[CONF_SERVICE_ACCOUNT_INFO]
client: texttospeech.TextToSpeechAsyncClient = (
texttospeech.TextToSpeechAsyncClient.from_service_account_info(
service_account_info
)
)
try:
voices = await async_tts_voices(client)
except GoogleAPIError as err:
_LOGGER.error("Error from calling list_voices: %s", err)
if isinstance(err, Unauthenticated):
config_entry.async_start_reauth(hass)
return
options_schema = tts_options_schema(dict(config_entry.options), voices)
language = config_entry.options.get(CONF_LANG, DEFAULT_LANG)
async_add_entities(
[
GoogleCloudTTSEntity(
config_entry,
client,
voices,
language,
options_schema,
)
]
)
class BaseGoogleCloudProvider:
"""The Google Cloud TTS base provider."""
def __init__(
self,
hass: HomeAssistant,
client: texttospeech.TextToSpeechAsyncClient,
voices: dict[str, list[str]],
language: str,
options_schema: vol.Schema,
) -> None:
"""Init Google Cloud TTS service."""
self.hass = hass
self.name = "Google Cloud TTS"
"""Init Google Cloud TTS base provider."""
self._client = client
self._voices = voices
self._language = language
@ -114,7 +159,7 @@ class GoogleCloudTTSProvider(Provider):
return None
return [Voice(voice, voice) for voice in voices]
async def async_get_tts_audio(
async def _async_get_tts_audio(
self,
message: str,
language: str,
@ -155,11 +200,7 @@ class GoogleCloudTTSProvider(Provider):
),
)
try:
response = await self._client.synthesize_speech(request, timeout=10)
except GoogleAPIError as err:
_LOGGER.error("Error occurred during Google Cloud TTS call: %s", err)
return None, None
response = await self._client.synthesize_speech(request, timeout=10)
if encoding == texttospeech.AudioEncoding.MP3:
extension = "mp3"
@ -169,3 +210,64 @@ class GoogleCloudTTSProvider(Provider):
extension = "wav"
return extension, response.audio_content
class GoogleCloudTTSEntity(BaseGoogleCloudProvider, TextToSpeechEntity):
"""The Google Cloud TTS entity."""
def __init__(
self,
entry: ConfigEntry,
client: texttospeech.TextToSpeechAsyncClient,
voices: dict[str, list[str]],
language: str,
options_schema: vol.Schema,
) -> None:
"""Init Google Cloud TTS entity."""
super().__init__(client, voices, language, options_schema)
self._attr_unique_id = f"{entry.entry_id}-tts"
self._attr_name = entry.title
self._attr_device_info = dr.DeviceInfo(
identifiers={(DOMAIN, entry.entry_id)},
manufacturer="Google",
model="Cloud",
entry_type=dr.DeviceEntryType.SERVICE,
)
self._entry = entry
async def async_get_tts_audio(
self, message: str, language: str, options: dict[str, Any]
) -> TtsAudioType:
"""Load TTS from Google Cloud."""
try:
return await self._async_get_tts_audio(message, language, options)
except GoogleAPIError as err:
_LOGGER.error("Error occurred during Google Cloud TTS call: %s", err)
if isinstance(err, Unauthenticated):
self._entry.async_start_reauth(self.hass)
return None, None
class GoogleCloudTTSProvider(BaseGoogleCloudProvider, Provider):
"""The Google Cloud TTS API provider."""
def __init__(
self,
client: texttospeech.TextToSpeechAsyncClient,
voices: dict[str, list[str]],
language: str,
options_schema: vol.Schema,
) -> None:
"""Init Google Cloud TTS service."""
super().__init__(client, voices, language, options_schema)
self.name = "Google Cloud TTS"
async def async_get_tts_audio(
self, message: str, language: str, options: dict[str, Any]
) -> TtsAudioType:
"""Load TTS from Google Cloud."""
try:
return await self._async_get_tts_audio(message, language, options)
except GoogleAPIError as err:
_LOGGER.error("Error occurred during Google Cloud TTS call: %s", err)
return None, None

1
homeassistant/generated/config_flows.py
View File

@ -222,6 +222,7 @@ FLOWS = {
"goodwe",
"google",
"google_assistant_sdk",
"google_cloud",
"google_generative_ai_conversation",
"google_mail",
"google_photos",

6
homeassistant/generated/integrations.json
View File

@ -2251,10 +2251,10 @@
"name": "Google Assistant SDK"
},
"google_cloud": {
"integration_type": "hub",
"config_flow": false,
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_push",
"name": "Google Cloud Platform"
"name": "Google Cloud"
},
"google_domains": {
"integration_type": "hub",

3
requirements_test_all.txt
View File

@ -836,6 +836,9 @@ google-api-python-client==2.71.0
# homeassistant.components.google_pubsub
google-cloud-pubsub==2.23.0
# homeassistant.components.google_cloud
google-cloud-texttospeech==2.17.2
# homeassistant.components.google_generative_ai_conversation
google-generativeai==0.7.2

1
tests/components/google_cloud/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Google Cloud integration."""

122
tests/components/google_cloud/conftest.py Normal file
View File

@ -0,0 +1,122 @@
"""Tests helpers."""
from collections.abc import Generator
import json
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
from google.cloud.texttospeech_v1.types import cloud_tts
import pytest
from homeassistant.components.google_cloud.const import (
CONF_SERVICE_ACCOUNT_INFO,
DOMAIN,
)
from tests.common import MockConfigEntry
VALID_SERVICE_ACCOUNT_INFO = {
"type": "service_account",
"project_id": "my project id",
"private_key_id": "my private key if",
"private_key": "-----BEGIN PRIVATE KEY-----\nMIICdwIBADANBgkqhkiG9w0BAQEFAASCAmEwggJdAgEAAoGBAKYscIlwm7soDsHAz6L6YvUkCvkrX19rS6yeYOmovvhoK5WeYGWUsd8V72zmsyHB7XO94YgJVjvxfzn5K8bLePjFzwoSJjZvhBJ/ZQ05d8VmbvgyWUoPdG9oEa4fZ/lCYrXoaFdTot2xcJvrb/ZuiRl4s4eZpNeFYvVK/Am7UeFPAgMBAAECgYAUetOfzLYUudofvPCaKHu7tKZ5kQPfEa0w6BAPnBF1Mfl1JiDBRDMryFtKs6AOIAVwx00dY/Ex0BCbB3+Cr58H7t4NaPTJxCpmR09pK7o17B7xAdQv8+SynFNud9/5vQ5AEXMOLNwKiU7wpXT6Z7ZIibUBOR7ewsWgsHCDpN1iqQJBAOMODPTPSiQMwRAUHIc6GPleFSJnIz2PAoG3JOG9KFAL6RtIc19lob2ZXdbQdzKtjSkWo+O5W20WDNAl1k32h6MCQQC7W4ZCIY67mPbL6CxXfHjpSGF4Dr9VWJ7ZrKHr6XUoOIcEvsn/pHvWonjMdy93rQMSfOE8BKd/I1+GHRmNVgplAkAnSo4paxmsZVyfeKt7Jy2dMY+8tVZe17maUuQaAE7Sk00SgJYegwrbMYgQnWCTL39HBfj0dmYA2Zj8CCAuu6O7AkEAryFiYjaUAO9+4iNoL27+ZrFtypeeadyov7gKs0ZKaQpNyzW8A+Zwi7TbTeSqzic/E+z/bOa82q7p/6b7141xsQJBANCAcIwMcVb6KVCHlQbOtKspo5Eh4ZQi8bGl+IcwbQ6JSxeTx915IfAldgbuU047wOB04dYCFB2yLDiUGVXTifU=\n-----END PRIVATE KEY-----\n",
"client_email": "my client email",
"client_id": "my client id",
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
"token_uri": "https://oauth2.googleapis.com/token",
"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/service-account",
"universe_domain": "googleapis.com",
}
@pytest.fixture
def create_google_credentials_json(tmp_path: Path) -> str:
"""Create googlecredentials.json."""
file_path = tmp_path / "googlecredentials.json"
with open(file_path, "w", encoding="utf8") as f:
json.dump(VALID_SERVICE_ACCOUNT_INFO, f)
return str(file_path)
@pytest.fixture
def create_invalid_google_credentials_json(create_google_credentials_json: str) -> str:
"""Create invalid googlecredentials.json."""
invalid_service_account_info = VALID_SERVICE_ACCOUNT_INFO.copy()
invalid_service_account_info.pop("client_email")
with open(create_google_credentials_json, "w", encoding="utf8") as f:
json.dump(invalid_service_account_info, f)
return create_google_credentials_json
@pytest.fixture
def mock_process_uploaded_file(
create_google_credentials_json: str,
) -> Generator[MagicMock]:
"""Mock upload certificate files."""
with patch(
"homeassistant.components.google_cloud.config_flow.process_uploaded_file",
return_value=Path(create_google_credentials_json),
) as mock_upload:
yield mock_upload
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="my Google Cloud title",
domain=DOMAIN,
data={CONF_SERVICE_ACCOUNT_INFO: VALID_SERVICE_ACCOUNT_INFO},
)
@pytest.fixture
def mock_api_tts() -> AsyncMock:
"""Return a mocked TTS client."""
mock_client = AsyncMock()
mock_client.list_voices.return_value = cloud_tts.ListVoicesResponse(
voices=[
cloud_tts.Voice(language_codes=["en-US"], name="en-US-Standard-A"),
cloud_tts.Voice(language_codes=["en-US"], name="en-US-Standard-B"),
cloud_tts.Voice(language_codes=["el-GR"], name="el-GR-Standard-A"),
]
)
return mock_client
@pytest.fixture
def mock_api_tts_from_service_account_info(
mock_api_tts: AsyncMock,
) -> Generator[AsyncMock]:
"""Return a mocked TTS client created with from_service_account_info."""
with (
patch(
"google.cloud.texttospeech.TextToSpeechAsyncClient.from_service_account_info",
return_value=mock_api_tts,
),
):
yield mock_api_tts
@pytest.fixture
def mock_api_tts_from_service_account_file(
mock_api_tts: AsyncMock,
) -> Generator[AsyncMock]:
"""Return a mocked TTS client created with from_service_account_file."""
with (
patch(
"google.cloud.texttospeech.TextToSpeechAsyncClient.from_service_account_file",
return_value=mock_api_tts,
),
):
yield mock_api_tts
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.google_cloud.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry

183
tests/components/google_cloud/test_config_flow.py Normal file
View File

@ -0,0 +1,183 @@
"""Test the Google Cloud config flow."""
from unittest.mock import AsyncMock, MagicMock
from uuid import uuid4
from homeassistant import config_entries
from homeassistant.components import tts
from homeassistant.components.google_cloud.config_flow import UPLOADED_KEY_FILE
from homeassistant.components.google_cloud.const import (
CONF_KEY_FILE,
CONF_SERVICE_ACCOUNT_INFO,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_PLATFORM
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.setup import async_setup_component
from .conftest import VALID_SERVICE_ACCOUNT_INFO
from tests.common import MockConfigEntry
async def test_user_flow_success(
hass: HomeAssistant,
mock_process_uploaded_file: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test user flow creates entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert not result["errors"]
uploaded_file = str(uuid4())
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{UPLOADED_KEY_FILE: uploaded_file},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Google Cloud"
assert result["data"] == {CONF_SERVICE_ACCOUNT_INFO: VALID_SERVICE_ACCOUNT_INFO}
mock_process_uploaded_file.assert_called_with(hass, uploaded_file)
assert len(mock_setup_entry.mock_calls) == 1
async def test_user_flow_missing_file(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
) -> None:
"""Test user flow when uploaded file is missing."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{UPLOADED_KEY_FILE: str(uuid4())},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_file"}
assert len(mock_setup_entry.mock_calls) == 0
async def test_user_flow_invalid_file(
hass: HomeAssistant,
create_invalid_google_credentials_json: str,
mock_process_uploaded_file: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test user flow when uploaded file is invalid."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
uploaded_file = str(uuid4())
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{UPLOADED_KEY_FILE: uploaded_file},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_file"}
mock_process_uploaded_file.assert_called_with(hass, uploaded_file)
assert len(mock_setup_entry.mock_calls) == 0
async def test_import_flow(
hass: HomeAssistant,
create_google_credentials_json: str,
mock_api_tts_from_service_account_file: AsyncMock,
mock_api_tts_from_service_account_info: AsyncMock,
) -> None:
"""Test the import flow."""
assert not hass.config_entries.async_entries(DOMAIN)
assert await async_setup_component(
hass,
tts.DOMAIN,
{
tts.DOMAIN: {CONF_PLATFORM: DOMAIN}
| {CONF_KEY_FILE: create_google_credentials_json}
},
)
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
assert config_entry.state is config_entries.ConfigEntryState.LOADED
async def test_import_flow_invalid_file(
hass: HomeAssistant,
create_invalid_google_credentials_json: str,
mock_api_tts_from_service_account_file: AsyncMock,
) -> None:
"""Test the import flow when the key file is invalid."""
assert not hass.config_entries.async_entries(DOMAIN)
assert await async_setup_component(
hass,
tts.DOMAIN,
{
tts.DOMAIN: {CONF_PLATFORM: DOMAIN}
| {CONF_KEY_FILE: create_invalid_google_credentials_json}
},
)
await hass.async_block_till_done()
assert not hass.config_entries.async_entries(DOMAIN)
assert mock_api_tts_from_service_account_file.list_voices.call_count == 1
async def test_options_flow(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_api_tts_from_service_account_info: AsyncMock,
) -> None:
"""Test options flow."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert mock_api_tts_from_service_account_info.list_voices.call_count == 1
assert mock_config_entry.options == {}
result = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
data_schema = result["data_schema"].schema
assert set(data_schema) == {
"language",
"gender",
"voice",
"encoding",
"speed",
"pitch",
"gain",
"profiles",
"text_type",
}
assert mock_api_tts_from_service_account_info.list_voices.call_count == 2
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"language": "el-GR"},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert mock_config_entry.options == {
"language": "el-GR",
"gender": "NEUTRAL",
"voice": "",
"encoding": "MP3",
"speed": 1.0,
"pitch": 0.0,
"gain": 0.0,
"profiles": [],
"text_type": "text",
}
assert mock_api_tts_from_service_account_info.list_voices.call_count == 3