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

Add Google Drive integration for backup (#134576) 

* Add Google Drive integration for backup

* Add test_config_flow

* Stop using aiogoogle

* address a few comments

* Check folder exists in setup

* fix test

* address comments

* fix

* fix

* Use ChunkAsyncStreamIterator in helpers

* repair-issues: todo

* Remove check if folder exists in the reatuh flow. This is done in setup.

* single_config_entry": true

* Add test_init.py

* Store into backups.json to avoid 124 bytes per property limit

* Address comments

* autouse=True on setup_credentials

* Store metadata in description and remove backups.json

* improvements

* timeout downloads

* library

* fixes

* strings

* review

* ruff

* fix test

* Set unique_id

* Use slugify in homeassistant.util

* Fix

* Remove RefreshError

* review

* push more fields to the test constant

---------

Co-authored-by: Joostlek <joostlek@outlook.com>
pull/136803/head
tronikos 2025-01-28 21:43:30 -08:00 committed by GitHub
parent 94e4863cbe
commit a2b5a96bc9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
25 changed files with 2098 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -217,6 +217,7 @@ homeassistant.components.goalzero.*
homeassistant.components.google.*
homeassistant.components.google_assistant_sdk.*
homeassistant.components.google_cloud.*
homeassistant.components.google_drive.*
homeassistant.components.google_photos.*
homeassistant.components.google_sheets.*
homeassistant.components.govee_ble.*

2
CODEOWNERS generated
View File

@ -566,6 +566,8 @@ build.json @home-assistant/supervisor
/tests/components/google_assistant_sdk/ @tronikos
/homeassistant/components/google_cloud/ @lufton @tronikos
/tests/components/google_cloud/ @lufton @tronikos
/homeassistant/components/google_drive/ @tronikos
/tests/components/google_drive/ @tronikos
/homeassistant/components/google_generative_ai_conversation/ @tronikos
/tests/components/google_generative_ai_conversation/ @tronikos
/homeassistant/components/google_mail/ @tkdrob

1
homeassistant/brands/google.json
View File

@ -5,6 +5,7 @@
"google_assistant",
"google_assistant_sdk",
"google_cloud",
"google_drive",
"google_generative_ai_conversation",
"google_mail",
"google_maps",

65
homeassistant/components/google_drive/__init__.py Normal file
View File

@ -0,0 +1,65 @@
"""The Google Drive integration."""
from __future__ import annotations
from collections.abc import Callable
from google_drive_api.exceptions import GoogleDriveApiError
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import instance_id
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.config_entry_oauth2_flow import (
OAuth2Session,
async_get_config_entry_implementation,
)
from homeassistant.util.hass_dict import HassKey
from .api import AsyncConfigEntryAuth, DriveClient
from .const import DOMAIN
DATA_BACKUP_AGENT_LISTENERS: HassKey[list[Callable[[], None]]] = HassKey(
f"{DOMAIN}.backup_agent_listeners"
)
type GoogleDriveConfigEntry = ConfigEntry[DriveClient]
async def async_setup_entry(hass: HomeAssistant, entry: GoogleDriveConfigEntry) -> bool:
"""Set up Google Drive from a config entry."""
auth = AsyncConfigEntryAuth(
async_get_clientsession(hass),
OAuth2Session(
hass, entry, await async_get_config_entry_implementation(hass, entry)
),
)
# Test we can refresh the token and raise ConfigEntryAuthFailed or ConfigEntryNotReady if not
await auth.async_get_access_token()
client = DriveClient(await instance_id.async_get(hass), auth)
entry.runtime_data = client
# Test we can access Google Drive and raise if not
try:
await client.async_create_ha_root_folder_if_not_exists()
except GoogleDriveApiError as err:
raise ConfigEntryNotReady from err
return True
async def async_unload_entry(
hass: HomeAssistant, entry: GoogleDriveConfigEntry
) -> bool:
"""Unload a config entry."""
hass.loop.call_soon(_notify_backup_listeners, hass)
return True
def _notify_backup_listeners(hass: HomeAssistant) -> None:
for listener in hass.data.get(DATA_BACKUP_AGENT_LISTENERS, []):
listener()

201
homeassistant/components/google_drive/api.py Normal file
View File

@ -0,0 +1,201 @@
"""API for Google Drive bound to Home Assistant OAuth."""
from __future__ import annotations
from collections.abc import AsyncIterator, Callable, Coroutine
import json
import logging
from typing import Any
from aiohttp import ClientSession, ClientTimeout, StreamReader
from aiohttp.client_exceptions import ClientError, ClientResponseError
from google_drive_api.api import AbstractAuth, GoogleDriveApi
from homeassistant.components.backup import AgentBackup
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryNotReady,
HomeAssistantError,
)
from homeassistant.helpers import config_entry_oauth2_flow
_UPLOAD_AND_DOWNLOAD_TIMEOUT = 12 * 3600
_LOGGER = logging.getLogger(__name__)
class AsyncConfigEntryAuth(AbstractAuth):
"""Provide Google Drive authentication tied to an OAuth2 based config entry."""
def __init__(
self,
websession: ClientSession,
oauth_session: config_entry_oauth2_flow.OAuth2Session,
) -> None:
"""Initialize AsyncConfigEntryAuth."""
super().__init__(websession)
self._oauth_session = oauth_session
async def async_get_access_token(self) -> str:
"""Return a valid access token."""
try:
await self._oauth_session.async_ensure_token_valid()
except ClientError as ex:
if (
self._oauth_session.config_entry.state
is ConfigEntryState.SETUP_IN_PROGRESS
):
if isinstance(ex, ClientResponseError) and 400 <= ex.status < 500:
raise ConfigEntryAuthFailed(
"OAuth session is not valid, reauth required"
) from ex
raise ConfigEntryNotReady from ex
if hasattr(ex, "status") and ex.status == 400:
self._oauth_session.config_entry.async_start_reauth(
self._oauth_session.hass
)
raise HomeAssistantError(ex) from ex
return str(self._oauth_session.token[CONF_ACCESS_TOKEN])
class AsyncConfigFlowAuth(AbstractAuth):
"""Provide authentication tied to a fixed token for the config flow."""
def __init__(
self,
websession: ClientSession,
token: str,
) -> None:
"""Initialize AsyncConfigFlowAuth."""
super().__init__(websession)
self._token = token
async def async_get_access_token(self) -> str:
"""Return a valid access token."""
return self._token
class DriveClient:
"""Google Drive client."""
def __init__(
self,
ha_instance_id: str,
auth: AbstractAuth,
) -> None:
"""Initialize Google Drive client."""
self._ha_instance_id = ha_instance_id
self._api = GoogleDriveApi(auth)
async def async_get_email_address(self) -> str:
"""Get email address of the current user."""
res = await self._api.get_user(params={"fields": "user(emailAddress)"})
return str(res["user"]["emailAddress"])
async def async_create_ha_root_folder_if_not_exists(self) -> tuple[str, str]:
"""Create Home Assistant folder if it doesn't exist."""
fields = "id,name"
query = " and ".join(
[
"properties has { key='home_assistant' and value='root' }",
f"properties has {{ key='instance_id' and value='{self._ha_instance_id}' }}",
"trashed=false",
]
)
res = await self._api.list_files(
params={"q": query, "fields": f"files({fields})"}
)
for file in res["files"]:
_LOGGER.debug("Found existing folder: %s", file)
return str(file["id"]), str(file["name"])
file_metadata = {
"name": "Home Assistant",
"mimeType": "application/vnd.google-apps.folder",
"properties": {
"home_assistant": "root",
"instance_id": self._ha_instance_id,
},
}
_LOGGER.debug("Creating new folder with metadata: %s", file_metadata)
res = await self._api.create_file(params={"fields": fields}, json=file_metadata)
_LOGGER.debug("Created folder: %s", res)
return str(res["id"]), str(res["name"])
async def async_upload_backup(
self,
open_stream: Callable[[], Coroutine[Any, Any, AsyncIterator[bytes]]],
backup: AgentBackup,
) -> None:
"""Upload a backup."""
folder_id, _ = await self.async_create_ha_root_folder_if_not_exists()
backup_metadata = {
"name": f"{backup.name} {backup.date}.tar",
"description": json.dumps(backup.as_dict()),
"parents": [folder_id],
"properties": {
"home_assistant": "backup",
"instance_id": self._ha_instance_id,
"backup_id": backup.backup_id,
},
}
_LOGGER.debug(
"Uploading backup: %s with Google Drive metadata: %s",
backup.backup_id,
backup_metadata,
)
await self._api.upload_file(
backup_metadata,
open_stream,
timeout=ClientTimeout(total=_UPLOAD_AND_DOWNLOAD_TIMEOUT),
)
_LOGGER.debug(
"Uploaded backup: %s to: '%s'",
backup.backup_id,
backup_metadata["name"],
)
async def async_list_backups(self) -> list[AgentBackup]:
"""List backups."""
query = " and ".join(
[
"properties has { key='home_assistant' and value='backup' }",
f"properties has {{ key='instance_id' and value='{self._ha_instance_id}' }}",
"trashed=false",
]
)
res = await self._api.list_files(
params={"q": query, "fields": "files(description)"}
)
backups = []
for file in res["files"]:
backup = AgentBackup.from_dict(json.loads(file["description"]))
backups.append(backup)
return backups
async def async_get_backup_file_id(self, backup_id: str) -> str | None:
"""Get file_id of backup if it exists."""
query = " and ".join(
[
"properties has { key='home_assistant' and value='backup' }",
f"properties has {{ key='instance_id' and value='{self._ha_instance_id}' }}",
f"properties has {{ key='backup_id' and value='{backup_id}' }}",
]
)
res = await self._api.list_files(params={"q": query, "fields": "files(id)"})
for file in res["files"]:
return str(file["id"])
return None
async def async_delete(self, file_id: str) -> None:
"""Delete file."""
await self._api.delete_file(file_id)
async def async_download(self, file_id: str) -> StreamReader:
"""Download a file."""
resp = await self._api.get_file_content(
file_id, timeout=ClientTimeout(total=_UPLOAD_AND_DOWNLOAD_TIMEOUT)
)
return resp.content

21
homeassistant/components/google_drive/application_credentials.py Normal file
View File

@ -0,0 +1,21 @@
"""application_credentials platform for Google Drive."""
from homeassistant.components.application_credentials import AuthorizationServer
from homeassistant.core import HomeAssistant
async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
"""Return authorization server."""
return AuthorizationServer(
"https://accounts.google.com/o/oauth2/v2/auth",
"https://oauth2.googleapis.com/token",
)
async def async_get_description_placeholders(hass: HomeAssistant) -> dict[str, str]:
"""Return description placeholders for the credentials dialog."""
return {
"oauth_consent_url": "https://console.cloud.google.com/apis/credentials/consent",
"more_info_url": "https://www.home-assistant.io/integrations/google_drive/",
"oauth_creds_url": "https://console.cloud.google.com/apis/credentials",
}

147
homeassistant/components/google_drive/backup.py Normal file
View File

@ -0,0 +1,147 @@
"""Backup platform for the Google Drive integration."""
from __future__ import annotations
from collections.abc import AsyncIterator, Callable, Coroutine
import logging
from typing import Any
from google_drive_api.exceptions import GoogleDriveApiError
from homeassistant.components.backup import AgentBackup, BackupAgent, BackupAgentError
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import ChunkAsyncStreamIterator
from homeassistant.util import slugify
from . import DATA_BACKUP_AGENT_LISTENERS, GoogleDriveConfigEntry
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
async def async_get_backup_agents(
hass: HomeAssistant,
**kwargs: Any,
) -> list[BackupAgent]:
"""Return a list of backup agents."""
entries = hass.config_entries.async_loaded_entries(DOMAIN)
return [GoogleDriveBackupAgent(entry) for entry in entries]
@callback
def async_register_backup_agents_listener(
hass: HomeAssistant,
*,
listener: Callable[[], None],
**kwargs: Any,
) -> Callable[[], None]:
"""Register a listener to be called when agents are added or removed.
:return: A function to unregister the listener.
"""
hass.data.setdefault(DATA_BACKUP_AGENT_LISTENERS, []).append(listener)
@callback
def remove_listener() -> None:
"""Remove the listener."""
hass.data[DATA_BACKUP_AGENT_LISTENERS].remove(listener)
if not hass.data[DATA_BACKUP_AGENT_LISTENERS]:
del hass.data[DATA_BACKUP_AGENT_LISTENERS]
return remove_listener
class GoogleDriveBackupAgent(BackupAgent):
"""Google Drive backup agent."""
domain = DOMAIN
def __init__(self, config_entry: GoogleDriveConfigEntry) -> None:
"""Initialize the cloud backup sync agent."""
super().__init__()
assert config_entry.unique_id
self.name = config_entry.title
self.unique_id = slugify(config_entry.unique_id)
self._client = config_entry.runtime_data
async def async_upload_backup(
self,
*,
open_stream: Callable[[], Coroutine[Any, Any, AsyncIterator[bytes]]],
backup: AgentBackup,
**kwargs: Any,
) -> None:
"""Upload a backup.
:param open_stream: A function returning an async iterator that yields bytes.
:param backup: Metadata about the backup that should be uploaded.
"""
try:
await self._client.async_upload_backup(open_stream, backup)
except (GoogleDriveApiError, HomeAssistantError, TimeoutError) as err:
_LOGGER.error("Upload backup error: %s", err)
raise BackupAgentError("Failed to upload backup") from err
async def async_list_backups(self, **kwargs: Any) -> list[AgentBackup]:
"""List backups."""
try:
return await self._client.async_list_backups()
except (GoogleDriveApiError, HomeAssistantError, TimeoutError) as err:
_LOGGER.error("List backups error: %s", err)
raise BackupAgentError("Failed to list backups") from err
async def async_get_backup(
self,
backup_id: str,
**kwargs: Any,
) -> AgentBackup | None:
"""Return a backup."""
backups = await self.async_list_backups()
for backup in backups:
if backup.backup_id == backup_id:
return backup
return None
async def async_download_backup(
self,
backup_id: str,
**kwargs: Any,
) -> AsyncIterator[bytes]:
"""Download a backup file.
:param backup_id: The ID of the backup that was returned in async_list_backups.
:return: An async iterator that yields bytes.
"""
_LOGGER.debug("Downloading backup_id: %s", backup_id)
try:
file_id = await self._client.async_get_backup_file_id(backup_id)
if file_id:
_LOGGER.debug("Downloading file_id: %s", file_id)
stream = await self._client.async_download(file_id)
return ChunkAsyncStreamIterator(stream)
except (GoogleDriveApiError, HomeAssistantError, TimeoutError) as err:
_LOGGER.error("Download backup error: %s", err)
raise BackupAgentError("Failed to download backup") from err
_LOGGER.error("Download backup_id: %s not found", backup_id)
raise BackupAgentError("Backup not found")
async def async_delete_backup(
self,
backup_id: str,
**kwargs: Any,
) -> None:
"""Delete a backup file.
:param backup_id: The ID of the backup that was returned in async_list_backups.
"""
_LOGGER.debug("Deleting backup_id: %s", backup_id)
try:
file_id = await self._client.async_get_backup_file_id(backup_id)
if file_id:
_LOGGER.debug("Deleting file_id: %s", file_id)
await self._client.async_delete(file_id)
_LOGGER.debug("Deleted backup_id: %s", backup_id)
except (GoogleDriveApiError, HomeAssistantError, TimeoutError) as err:
_LOGGER.error("Delete backup error: %s", err)
raise BackupAgentError("Failed to delete backup") from err

114
homeassistant/components/google_drive/config_flow.py Normal file
View File

@ -0,0 +1,114 @@
"""Config flow for the Google Drive integration."""
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any, cast
from google_drive_api.exceptions import GoogleDriveApiError
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_TOKEN
from homeassistant.helpers import config_entry_oauth2_flow, instance_id
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .api import AsyncConfigFlowAuth, DriveClient
from .const import DOMAIN
DEFAULT_NAME = "Google Drive"
DRIVE_FOLDER_URL_PREFIX = "https://drive.google.com/drive/folders/"
OAUTH2_SCOPES = [
"https://www.googleapis.com/auth/drive.file",
]
class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Config flow to handle Google Drive OAuth2 authentication."""
DOMAIN = DOMAIN
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return logging.getLogger(__name__)
@property
def extra_authorize_data(self) -> dict[str, Any]:
"""Extra data that needs to be appended to the authorize url."""
return {
"scope": " ".join(OAUTH2_SCOPES),
# Add params to ensure we get back a refresh token
"access_type": "offline",
"prompt": "consent",
}
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon an API authentication error."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm reauth dialog."""
if user_input is None:
return self.async_show_form(step_id="reauth_confirm")
return await self.async_step_user()
async def async_oauth_create_entry(self, data: dict[str, Any]) -> ConfigFlowResult:
"""Create an entry for the flow, or update existing entry."""
client = DriveClient(
await instance_id.async_get(self.hass),
AsyncConfigFlowAuth(
async_get_clientsession(self.hass), data[CONF_TOKEN][CONF_ACCESS_TOKEN]
),
)
try:
email_address = await client.async_get_email_address()
except GoogleDriveApiError as err:
self.logger.error("Error getting email address: %s", err)
return self.async_abort(
reason="access_not_configured",
description_placeholders={"message": str(err)},
)
except Exception:
self.logger.exception("Unknown error occurred")
return self.async_abort(reason="unknown")
await self.async_set_unique_id(email_address)
if self.source == SOURCE_REAUTH:
reauth_entry = self._get_reauth_entry()
self._abort_if_unique_id_mismatch(
reason="wrong_account",
description_placeholders={"email": cast(str, reauth_entry.unique_id)},
)
return self.async_update_reload_and_abort(reauth_entry, data=data)
self._abort_if_unique_id_configured()
try:
(
folder_id,
folder_name,
) = await client.async_create_ha_root_folder_if_not_exists()
except GoogleDriveApiError as err:
self.logger.error("Error creating folder: %s", str(err))
return self.async_abort(
reason="create_folder_failure",
description_placeholders={"message": str(err)},
)
return self.async_create_entry(
title=DEFAULT_NAME,
data=data,
description_placeholders={
"folder_name": folder_name,
"url": f"{DRIVE_FOLDER_URL_PREFIX}{folder_id}",
},
)

5
homeassistant/components/google_drive/const.py Normal file
View File

@ -0,0 +1,5 @@
"""Constants for the Google Drive integration."""
from __future__ import annotations
DOMAIN = "google_drive"

14
homeassistant/components/google_drive/manifest.json Normal file
View File

@ -0,0 +1,14 @@
{
"domain": "google_drive",
"name": "Google Drive",
"after_dependencies": ["backup"],
"codeowners": ["@tronikos"],
"config_flow": true,
"dependencies": ["application_credentials"],
"documentation": "https://www.home-assistant.io/integrations/google_drive",
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["google_drive_api"],
"quality_scale": "platinum",
"requirements": ["python-google-drive-api==0.0.2"]
}

113
homeassistant/components/google_drive/quality_scale.yaml Normal file
View File

@ -0,0 +1,113 @@
rules:
# Bronze
action-setup:
status: exempt
comment: No actions.
appropriate-polling:
status: exempt
comment: No polling.
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: No actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: No entities.
entity-unique-id:
status: exempt
comment: No entities.
has-entity-name:
status: exempt
comment: No entities.
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions: done
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: No configuration options.
docs-installation-parameters: done
entity-unavailable:
status: exempt
comment: No entities.
integration-owner: done
log-when-unavailable:
status: exempt
comment: No entities.
parallel-updates:
status: exempt
comment: No actions and no entities.
reauthentication-flow: done
test-coverage: done
# Gold
devices:
status: exempt
comment: No devices.
diagnostics:
status: exempt
comment: No data to diagnose.
discovery-update-info:
status: exempt
comment: No discovery.
discovery:
status: exempt
comment: No discovery.
docs-data-update:
status: exempt
comment: No updates.
docs-examples:
status: exempt
comment: |
This integration only serves backup.
docs-known-limitations: done
docs-supported-devices:
status: exempt
comment: No devices.
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
dynamic-devices:
status: exempt
comment: No devices.
entity-category:
status: exempt
comment: No entities.
entity-device-class:
status: exempt
comment: No entities.
entity-disabled-by-default:
status: exempt
comment: No entities.
entity-translations:
status: exempt
comment: No entities.
exception-translations: done
icon-translations:
status: exempt
comment: No entities.
reconfiguration-flow:
status: exempt
comment: No configuration options.
repair-issues:
status: exempt
comment: No repairs.
stale-devices:
status: exempt
comment: No devices.
# Platinum
async-dependency: done
inject-websession: done
strict-typing: done

40
homeassistant/components/google_drive/strings.json Normal file
View File

@ -0,0 +1,40 @@
{
"config": {
"step": {
"pick_implementation": {
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
},
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "The Google Drive integration needs to re-authenticate your account"
},
"auth": {
"title": "Link Google Account"
}
},
"abort": {
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"missing_credentials": "[%key:common::config_flow::abort::oauth2_missing_credentials%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
"user_rejected_authorize": "[%key:common::config_flow::abort::oauth2_user_rejected_authorize%]",
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
"oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"access_not_configured": "Unable to access the Google Drive API:\n\n{message}",
"create_folder_failure": "Error while creating Google Drive folder:\n\n{message}",
"unknown": "[%key:common::config_flow::error::unknown%]",
"wrong_account": "Wrong account: Please authenticate with {email}."
},
"create_entry": {
"default": "Using [{folder_name}]({url}) folder. Feel free to rename it in Google Drive as you wish."
}
},
"application_credentials": {
"description": "Follow the [instructions]({more_info_url}) for [OAuth consent screen]({oauth_consent_url}) to give Home Assistant access to your Google Drive. You also need to create Application Credentials linked to your account:\n1. Go to [Credentials]({oauth_creds_url}) and select **Create Credentials**.\n1. From the drop-down list select **OAuth client ID**.\n1. Select **Web application** for the Application Type."
}
}

1
homeassistant/generated/application_credentials.py generated
View File

@ -9,6 +9,7 @@ APPLICATION_CREDENTIALS = [
"geocaching",
"google",
"google_assistant_sdk",
"google_drive",
"google_mail",
"google_photos",
"google_sheets",

1
homeassistant/generated/config_flows.py generated
View File

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

6
homeassistant/generated/integrations.json
View File

@ -2295,6 +2295,12 @@
"iot_class": "cloud_push",
"name": "Google Cloud"
},
"google_drive": {
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_polling",
"name": "Google Drive"
},
"google_generative_ai_conversation": {
"integration_type": "service",
"config_flow": true,

10
mypy.ini generated
View File

@ -1926,6 +1926,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.google_drive.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.google_photos.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_all.txt generated
View File

@ -2384,6 +2384,9 @@ python-gc100==1.0.3a0
# homeassistant.components.gitlab_ci
python-gitlab==1.6.0
# homeassistant.components.google_drive
python-google-drive-api==0.0.2
# homeassistant.components.analytics_insights
python-homeassistant-analytics==0.8.1

3
requirements_test_all.txt generated
View File

@ -1929,6 +1929,9 @@ python-fullykiosk==0.0.14
# homeassistant.components.sms
# python-gammu==3.2.4
# homeassistant.components.google_drive
python-google-drive-api==0.0.2
# homeassistant.components.analytics_insights
python-homeassistant-analytics==0.8.1

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

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

80
tests/components/google_drive/conftest.py Normal file
View File

@ -0,0 +1,80 @@
"""PyTest fixtures and test helpers."""
from collections.abc import Generator
import time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.google_drive.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
HA_UUID = "0a123c"
TEST_AGENT_ID = "google_drive.testuser_domain_com"
TEST_USER_EMAIL = "testuser@domain.com"
CONFIG_ENTRY_TITLE = "Google Drive entry title"
@pytest.fixture(autouse=True)
async def setup_credentials(hass: HomeAssistant) -> None:
"""Fixture to setup credentials."""
assert await async_setup_component(hass, "application_credentials", {})
await async_import_client_credential(
hass,
DOMAIN,
ClientCredential(CLIENT_ID, CLIENT_SECRET),
)
@pytest.fixture
def mock_api() -> Generator[MagicMock]:
"""Return a mocked GoogleDriveApi."""
with patch(
"homeassistant.components.google_drive.api.GoogleDriveApi"
) as mock_api_cl:
mock_api = mock_api_cl.return_value
yield mock_api
@pytest.fixture(autouse=True)
def mock_instance_id() -> Generator[AsyncMock]:
"""Mock instance_id."""
with patch(
"homeassistant.components.google_drive.config_flow.instance_id.async_get",
return_value=HA_UUID,
):
yield
@pytest.fixture(name="expires_at")
def mock_expires_at() -> int:
"""Fixture to set the oauth token expiration time."""
return time.time() + 3600
@pytest.fixture(name="config_entry")
def mock_config_entry(expires_at: int) -> MockConfigEntry:
"""Fixture for MockConfigEntry."""
return MockConfigEntry(
domain=DOMAIN,
unique_id=TEST_USER_EMAIL,
title=CONFIG_ENTRY_TITLE,
data={
"auth_implementation": DOMAIN,
"token": {
"access_token": "mock-access-token",
"refresh_token": "mock-refresh-token",
"expires_at": expires_at,
"scope": "https://www.googleapis.com/auth/drive.file",
},
},
)

237
tests/components/google_drive/snapshots/test_backup.ambr Normal file
View File

@ -0,0 +1,237 @@
# serializer version: 1
# name: test_agents_delete
list([
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(id,name)',
'q': "properties has { key='home_assistant' and value='root' } and properties has { key='instance_id' and value='0a123c' } and trashed=false",
}),
}),
),
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(id)',
'q': "properties has { key='home_assistant' and value='backup' } and properties has { key='instance_id' and value='0a123c' } and properties has { key='backup_id' and value='test-backup' }",
}),
}),
),
tuple(
'delete_file',
tuple(
'backup-file-id',
),
dict({
}),
),
])
# ---
# name: test_agents_download
list([
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(id,name)',
'q': "properties has { key='home_assistant' and value='root' } and properties has { key='instance_id' and value='0a123c' } and trashed=false",
}),
}),
),
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(description)',
'q': "properties has { key='home_assistant' and value='backup' } and properties has { key='instance_id' and value='0a123c' } and trashed=false",
}),
}),
),
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(id)',
'q': "properties has { key='home_assistant' and value='backup' } and properties has { key='instance_id' and value='0a123c' } and properties has { key='backup_id' and value='test-backup' }",
}),
}),
),
tuple(
'get_file_content',
tuple(
'backup-file-id',
),
dict({
'timeout': dict({
'ceil_threshold': 5,
'connect': None,
'sock_connect': None,
'sock_read': None,
'total': 43200,
}),
}),
),
])
# ---
# name: test_agents_list_backups
list([
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(id,name)',
'q': "properties has { key='home_assistant' and value='root' } and properties has { key='instance_id' and value='0a123c' } and trashed=false",
}),
}),
),
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(description)',
'q': "properties has { key='home_assistant' and value='backup' } and properties has { key='instance_id' and value='0a123c' } and trashed=false",
}),
}),
),
])
# ---
# name: test_agents_upload
list([
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(id,name)',
'q': "properties has { key='home_assistant' and value='root' } and properties has { key='instance_id' and value='0a123c' } and trashed=false",
}),
}),
),
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(id,name)',
'q': "properties has { key='home_assistant' and value='root' } and properties has { key='instance_id' and value='0a123c' } and trashed=false",
}),
}),
),
tuple(
'upload_file',
tuple(
dict({
'description': '{"addons": [{"name": "Test", "slug": "test", "version": "1.0.0"}], "backup_id": "test-backup", "date": "2025-01-01T01:23:45.678Z", "database_included": true, "extra_metadata": {"with_automatic_settings": false}, "folders": [], "homeassistant_included": true, "homeassistant_version": "2024.12.0", "name": "Test", "protected": false, "size": 987}',
'name': 'Test 2025-01-01T01:23:45.678Z.tar',
'parents': list([
'HA folder ID',
]),
'properties': dict({
'backup_id': 'test-backup',
'home_assistant': 'backup',
'instance_id': '0a123c',
}),
}),
"CoreBackupReaderWriter.async_receive_backup.<locals>.open_backup() -> 'AsyncIterator[bytes]'",
),
dict({
'timeout': dict({
'ceil_threshold': 5,
'connect': None,
'sock_connect': None,
'sock_read': None,
'total': 43200,
}),
}),
),
])
# ---
# name: test_agents_upload_create_folder_if_missing
list([
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(id,name)',
'q': "properties has { key='home_assistant' and value='root' } and properties has { key='instance_id' and value='0a123c' } and trashed=false",
}),
}),
),
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(id,name)',
'q': "properties has { key='home_assistant' and value='root' } and properties has { key='instance_id' and value='0a123c' } and trashed=false",
}),
}),
),
tuple(
'create_file',
tuple(
),
dict({
'json': dict({
'mimeType': 'application/vnd.google-apps.folder',
'name': 'Home Assistant',
'properties': dict({
'home_assistant': 'root',
'instance_id': '0a123c',
}),
}),
'params': dict({
'fields': 'id,name',
}),
}),
),
tuple(
'upload_file',
tuple(
dict({
'description': '{"addons": [{"name": "Test", "slug": "test", "version": "1.0.0"}], "backup_id": "test-backup", "date": "2025-01-01T01:23:45.678Z", "database_included": true, "extra_metadata": {"with_automatic_settings": false}, "folders": [], "homeassistant_included": true, "homeassistant_version": "2024.12.0", "name": "Test", "protected": false, "size": 987}',
'name': 'Test 2025-01-01T01:23:45.678Z.tar',
'parents': list([
'new folder id',
]),
'properties': dict({
'backup_id': 'test-backup',
'home_assistant': 'backup',
'instance_id': '0a123c',
}),
}),
"CoreBackupReaderWriter.async_receive_backup.<locals>.open_backup() -> 'AsyncIterator[bytes]'",
),
dict({
'timeout': dict({
'ceil_threshold': 5,
'connect': None,
'sock_connect': None,
'sock_read': None,
'total': 43200,
}),
}),
),
])
# ---

44
tests/components/google_drive/snapshots/test_config_flow.ambr Normal file
View File

@ -0,0 +1,44 @@
# serializer version: 1
# name: test_full_flow
list([
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'user(emailAddress)',
}),
}),
),
tuple(
'list_files',
tuple(
),
dict({
'params': dict({
'fields': 'files(id,name)',
'q': "properties has { key='home_assistant' and value='root' } and properties has { key='instance_id' and value='0a123c' } and trashed=false",
}),
}),
),
tuple(
'create_file',
tuple(
),
dict({
'json': dict({
'mimeType': 'application/vnd.google-apps.folder',
'name': 'Home Assistant',
'properties': dict({
'home_assistant': 'root',
'instance_id': '0a123c',
}),
}),
'params': dict({
'fields': 'id,name',
}),
}),
),
])
# ---

461
tests/components/google_drive/test_backup.py Normal file
View File

@ -0,0 +1,461 @@
"""Test the Google Drive backup platform."""
from io import StringIO
import json
from typing import Any
from unittest.mock import AsyncMock, MagicMock, Mock, patch
from aiohttp import ClientResponse
from google_drive_api.exceptions import GoogleDriveApiError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.backup import (
DOMAIN as BACKUP_DOMAIN,
AddonInfo,
AgentBackup,
)
from homeassistant.components.google_drive import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import CONFIG_ENTRY_TITLE, TEST_AGENT_ID
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import mock_stream
from tests.typing import ClientSessionGenerator, WebSocketGenerator
FOLDER_ID = "google-folder-id"
TEST_AGENT_BACKUP = AgentBackup(
addons=[AddonInfo(name="Test", slug="test", version="1.0.0")],
backup_id="test-backup",
database_included=True,
date="2025-01-01T01:23:45.678Z",
extra_metadata={
"with_automatic_settings": False,
},
folders=[],
homeassistant_included=True,
homeassistant_version="2024.12.0",
name="Test",
protected=False,
size=987,
)
TEST_AGENT_BACKUP_RESULT = {
"addons": [{"name": "Test", "slug": "test", "version": "1.0.0"}],
"backup_id": "test-backup",
"database_included": True,
"date": "2025-01-01T01:23:45.678Z",
"folders": [],
"homeassistant_included": True,
"homeassistant_version": "2024.12.0",
"name": "Test",
"protected": False,
"size": 987,
"agent_ids": [TEST_AGENT_ID],
"failed_agent_ids": [],
"with_automatic_settings": None,
}
@pytest.fixture(autouse=True)
async def setup_integration(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_api: MagicMock,
) -> None:
"""Set up Google Drive integration."""
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, BACKUP_DOMAIN, {BACKUP_DOMAIN: {}})
mock_api.list_files = AsyncMock(
return_value={"files": [{"id": "HA folder ID", "name": "HA folder name"}]}
)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
async def test_agents_info(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test backup agent info."""
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/agents/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"agents": [
{"agent_id": "backup.local", "name": "local"},
{"agent_id": TEST_AGENT_ID, "name": CONFIG_ENTRY_TITLE},
],
}
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
await client.send_json_auto_id({"type": "backup/agents/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"agents": [{"agent_id": "backup.local", "name": "local"}]
}
async def test_agents_list_backups(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_api: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test agent list backups."""
mock_api.list_files = AsyncMock(
return_value={
"files": [{"description": json.dumps(TEST_AGENT_BACKUP.as_dict())}]
}
)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"]["agent_errors"] == {}
assert response["result"]["backups"] == [TEST_AGENT_BACKUP_RESULT]
assert [tuple(mock_call) for mock_call in mock_api.mock_calls] == snapshot
async def test_agents_list_backups_fail(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_api: MagicMock,
) -> None:
"""Test agent list backups fails."""
mock_api.list_files = AsyncMock(side_effect=GoogleDriveApiError("some error"))
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"]["backups"] == []
assert response["result"]["agent_errors"] == {
TEST_AGENT_ID: "Failed to list backups"
}
@pytest.mark.parametrize(
("backup_id", "expected_result"),
[
(TEST_AGENT_BACKUP.backup_id, TEST_AGENT_BACKUP_RESULT),
("12345", None),
],
ids=["found", "not_found"],
)
async def test_agents_get_backup(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_api: MagicMock,
backup_id: str,
expected_result: dict[str, Any] | None,
) -> None:
"""Test agent get backup."""
mock_api.list_files = AsyncMock(
return_value={
"files": [{"description": json.dumps(TEST_AGENT_BACKUP.as_dict())}]
}
)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/details", "backup_id": backup_id})
response = await client.receive_json()
assert response["success"]
assert response["result"]["agent_errors"] == {}
assert response["result"]["backup"] == expected_result
async def test_agents_download(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_api: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test agent download backup."""
mock_api.list_files = AsyncMock(
side_effect=[
{"files": [{"description": json.dumps(TEST_AGENT_BACKUP.as_dict())}]},
{"files": [{"id": "backup-file-id"}]},
]
)
mock_response = AsyncMock(spec=ClientResponse)
mock_response.content = mock_stream(b"backup data")
mock_api.get_file_content = AsyncMock(return_value=mock_response)
client = await hass_client()
resp = await client.get(
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
)
assert resp.status == 200
assert await resp.content.read() == b"backup data"
assert [tuple(mock_call) for mock_call in mock_api.mock_calls] == snapshot
async def test_agents_download_fail(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_api: MagicMock,
) -> None:
"""Test agent download backup fails."""
mock_api.list_files = AsyncMock(
side_effect=[
{"files": [{"description": json.dumps(TEST_AGENT_BACKUP.as_dict())}]},
{"files": [{"id": "backup-file-id"}]},
]
)
mock_response = AsyncMock(spec=ClientResponse)
mock_response.content = mock_stream(b"backup data")
mock_api.get_file_content = AsyncMock(side_effect=GoogleDriveApiError("some error"))
client = await hass_client()
resp = await client.get(
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
)
assert resp.status == 500
content = await resp.content.read()
assert "Failed to download backup" in content.decode()
async def test_agents_download_file_not_found(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_api: MagicMock,
) -> None:
"""Test agent download backup raises error if not found."""
mock_api.list_files = AsyncMock(
side_effect=[
{"files": [{"description": json.dumps(TEST_AGENT_BACKUP.as_dict())}]},
{"files": []},
]
)
client = await hass_client()
resp = await client.get(
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
)
assert resp.status == 500
content = await resp.content.read()
assert "Backup not found" in content.decode()
async def test_agents_download_metadata_not_found(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_api: MagicMock,
) -> None:
"""Test agent download backup raises error if not found."""
mock_api.list_files = AsyncMock(
return_value={
"files": [{"description": json.dumps(TEST_AGENT_BACKUP.as_dict())}]
}
)
client = await hass_client()
backup_id = "1234"
assert backup_id != TEST_AGENT_BACKUP.backup_id
resp = await client.get(
f"/api/backup/download/{backup_id}?agent_id={TEST_AGENT_ID}"
)
assert resp.status == 404
assert await resp.content.read() == b""
async def test_agents_upload(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
mock_api: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test agent upload backup."""
mock_api.upload_file = AsyncMock(return_value=None)
client = await hass_client()
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
) as fetch_backup,
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=TEST_AGENT_BACKUP,
),
patch("pathlib.Path.open") as mocked_open,
):
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
fetch_backup.return_value = TEST_AGENT_BACKUP
resp = await client.post(
f"/api/backup/upload?agent_id={TEST_AGENT_ID}",
data={"file": StringIO("test")},
)
assert resp.status == 201
assert f"Uploading backup: {TEST_AGENT_BACKUP.backup_id}" in caplog.text
assert f"Uploaded backup: {TEST_AGENT_BACKUP.backup_id}" in caplog.text
mock_api.upload_file.assert_called_once()
assert [tuple(mock_call) for mock_call in mock_api.mock_calls] == snapshot
async def test_agents_upload_create_folder_if_missing(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
mock_api: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test agent upload backup creates folder if missing."""
mock_api.list_files = AsyncMock(return_value={"files": []})
mock_api.create_file = AsyncMock(
return_value={"id": "new folder id", "name": "Home Assistant"}
)
mock_api.upload_file = AsyncMock(return_value=None)
client = await hass_client()
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
) as fetch_backup,
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=TEST_AGENT_BACKUP,
),
patch("pathlib.Path.open") as mocked_open,
):
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
fetch_backup.return_value = TEST_AGENT_BACKUP
resp = await client.post(
f"/api/backup/upload?agent_id={TEST_AGENT_ID}",
data={"file": StringIO("test")},
)
assert resp.status == 201
assert f"Uploading backup: {TEST_AGENT_BACKUP.backup_id}" in caplog.text
assert f"Uploaded backup: {TEST_AGENT_BACKUP.backup_id}" in caplog.text
mock_api.create_file.assert_called_once()
mock_api.upload_file.assert_called_once()
assert [tuple(mock_call) for mock_call in mock_api.mock_calls] == snapshot
async def test_agents_upload_fail(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
mock_api: MagicMock,
) -> None:
"""Test agent upload backup fails."""
mock_api.upload_file = AsyncMock(side_effect=GoogleDriveApiError("some error"))
client = await hass_client()
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
) as fetch_backup,
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=TEST_AGENT_BACKUP,
),
patch("pathlib.Path.open") as mocked_open,
):
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
fetch_backup.return_value = TEST_AGENT_BACKUP
resp = await client.post(
f"/api/backup/upload?agent_id={TEST_AGENT_ID}",
data={"file": StringIO("test")},
)
await hass.async_block_till_done()
assert resp.status == 201
assert "Upload backup error: some error" in caplog.text
async def test_agents_delete(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_api: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test agent delete backup."""
mock_api.list_files = AsyncMock(return_value={"files": [{"id": "backup-file-id"}]})
mock_api.delete_file = AsyncMock(return_value=None)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "backup/delete",
"backup_id": TEST_AGENT_BACKUP.backup_id,
}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {"agent_errors": {}}
mock_api.delete_file.assert_called_once()
assert [tuple(mock_call) for mock_call in mock_api.mock_calls] == snapshot
async def test_agents_delete_fail(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_api: MagicMock,
) -> None:
"""Test agent delete backup fails."""
mock_api.list_files = AsyncMock(return_value={"files": [{"id": "backup-file-id"}]})
mock_api.delete_file = AsyncMock(side_effect=GoogleDriveApiError("some error"))
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "backup/delete",
"backup_id": TEST_AGENT_BACKUP.backup_id,
}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"agent_errors": {TEST_AGENT_ID: "Failed to delete backup"}
}
async def test_agents_delete_not_found(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_api: MagicMock,
) -> None:
"""Test agent delete backup not found."""
mock_api.list_files = AsyncMock(return_value={"files": []})
client = await hass_ws_client(hass)
backup_id = "1234"
await client.send_json_auto_id(
{
"type": "backup/delete",
"backup_id": backup_id,
}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {"agent_errors": {}}
mock_api.delete_file.assert_not_called()

363
tests/components/google_drive/test_config_flow.py Normal file
View File

@ -0,0 +1,363 @@
"""Test the Google Drive config flow."""
from unittest.mock import AsyncMock, MagicMock, patch
from google_drive_api.exceptions import GoogleDriveApiError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.google_drive.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from .conftest import CLIENT_ID, TEST_USER_EMAIL
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
GOOGLE_AUTH_URI = "https://accounts.google.com/o/oauth2/v2/auth"
GOOGLE_TOKEN_URI = "https://oauth2.googleapis.com/token"
FOLDER_ID = "google-folder-id"
FOLDER_NAME = "folder name"
TITLE = "Google Drive"
@pytest.mark.usefixtures("current_request_with_host")
async def test_full_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_api: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Check full flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["url"] == (
f"{GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope=https://www.googleapis.com/auth/drive.file"
"&access_type=offline&prompt=consent"
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
# Prepare API responses
mock_api.get_user = AsyncMock(
return_value={"user": {"emailAddress": TEST_USER_EMAIL}}
)
mock_api.list_files = AsyncMock(return_value={"files": []})
mock_api.create_file = AsyncMock(
return_value={"id": FOLDER_ID, "name": FOLDER_NAME}
)
aioclient_mock.post(
GOOGLE_TOKEN_URI,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.components.google_drive.async_setup_entry", return_value=True
) as mock_setup:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
assert len(aioclient_mock.mock_calls) == 1
assert [tuple(mock_call) for mock_call in mock_api.mock_calls] == snapshot
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == TITLE
assert result.get("description_placeholders") == {
"folder_name": FOLDER_NAME,
"url": f"https://drive.google.com/drive/folders/{FOLDER_ID}",
}
assert "result" in result
assert result.get("result").unique_id == TEST_USER_EMAIL
assert "token" in result.get("result").data
assert result.get("result").data["token"].get("access_token") == "mock-access-token"
assert (
result.get("result").data["token"].get("refresh_token") == "mock-refresh-token"
)
@pytest.mark.usefixtures("current_request_with_host")
async def test_create_folder_error(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_api: MagicMock,
) -> None:
"""Test case where creating the folder fails."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["url"] == (
f"{GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope=https://www.googleapis.com/auth/drive.file"
"&access_type=offline&prompt=consent"
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
# Prepare API responses
mock_api.get_user = AsyncMock(
return_value={"user": {"emailAddress": TEST_USER_EMAIL}}
)
mock_api.list_files = AsyncMock(return_value={"files": []})
mock_api.create_file = AsyncMock(side_effect=GoogleDriveApiError("some error"))
aioclient_mock.post(
GOOGLE_TOKEN_URI,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "create_folder_failure"
assert result.get("description_placeholders") == {"message": "some error"}
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.parametrize(
("exception", "expected_abort_reason", "expected_placeholders"),
[
(
GoogleDriveApiError("some error"),
"access_not_configured",
{"message": "some error"},
),
(Exception, "unknown", None),
],
ids=["api_not_enabled", "general_exception"],
)
async def test_get_email_error(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_api: MagicMock,
exception: Exception,
expected_abort_reason,
expected_placeholders,
) -> None:
"""Test case where getting the email address fails."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["url"] == (
f"{GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope=https://www.googleapis.com/auth/drive.file"
"&access_type=offline&prompt=consent"
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
# Prepare API responses
mock_api.get_user = AsyncMock(side_effect=exception)
aioclient_mock.post(
GOOGLE_TOKEN_URI,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == expected_abort_reason
assert result.get("description_placeholders") == expected_placeholders
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.parametrize(
(
"new_email",
"expected_abort_reason",
"expected_placeholders",
"expected_access_token",
"expected_setup_calls",
),
[
(TEST_USER_EMAIL, "reauth_successful", None, "updated-access-token", 1),
(
"other.user@domain.com",
"wrong_account",
{"email": TEST_USER_EMAIL},
"mock-access-token",
0,
),
],
ids=["reauth_successful", "wrong_account"],
)
async def test_reauth(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
config_entry: MockConfigEntry,
aioclient_mock: AiohttpClientMocker,
mock_api: MagicMock,
new_email: str,
expected_abort_reason: str,
expected_placeholders: dict[str, str] | None,
expected_access_token: str,
expected_setup_calls: int,
) -> None:
"""Test the reauthentication flow."""
config_entry.add_to_hass(hass)
result = await config_entry.start_reauth_flow(hass)
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["url"] == (
f"{GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope=https://www.googleapis.com/auth/drive.file"
"&access_type=offline&prompt=consent"
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
# Prepare API responses
mock_api.get_user = AsyncMock(return_value={"user": {"emailAddress": new_email}})
aioclient_mock.post(
GOOGLE_TOKEN_URI,
json={
"refresh_token": "mock-refresh-token",
"access_token": "updated-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.components.google_drive.async_setup_entry", return_value=True
) as mock_setup:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == expected_setup_calls
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == expected_abort_reason
assert result.get("description_placeholders") == expected_placeholders
assert config_entry.unique_id == TEST_USER_EMAIL
assert "token" in config_entry.data
# Verify access token is refreshed
assert config_entry.data["token"].get("access_token") == expected_access_token
assert config_entry.data["token"].get("refresh_token") == "mock-refresh-token"
@pytest.mark.usefixtures("current_request_with_host")
async def test_already_configured(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
config_entry: MockConfigEntry,
aioclient_mock: AiohttpClientMocker,
mock_api: MagicMock,
) -> None:
"""Test already configured account."""
config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["url"] == (
f"{GOOGLE_AUTH_URI}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope=https://www.googleapis.com/auth/drive.file"
"&access_type=offline&prompt=consent"
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
# Prepare API responses
mock_api.get_user = AsyncMock(
return_value={"user": {"emailAddress": TEST_USER_EMAIL}}
)
aioclient_mock.post(
GOOGLE_TOKEN_URI,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "already_configured"

164
tests/components/google_drive/test_init.py Normal file
View File

@ -0,0 +1,164 @@
"""Tests for Google Drive."""
from collections.abc import Awaitable, Callable, Coroutine
import http
import time
from typing import Any
from unittest.mock import AsyncMock, MagicMock
from google_drive_api.exceptions import GoogleDriveApiError
import pytest
from homeassistant.components.google_drive.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
type ComponentSetup = Callable[[], Awaitable[None]]
@pytest.fixture(name="setup_integration")
async def mock_setup_integration(
hass: HomeAssistant,
config_entry: MockConfigEntry,
) -> Callable[[], Coroutine[Any, Any, None]]:
"""Fixture for setting up the component."""
config_entry.add_to_hass(hass)
async def func() -> None:
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return func
async def test_setup_success(
hass: HomeAssistant,
setup_integration: ComponentSetup,
mock_api: MagicMock,
) -> None:
"""Test successful setup and unload."""
# Setup looks up existing folder to make sure it still exists
mock_api.list_files = AsyncMock(
return_value={"files": [{"id": "HA folder ID", "name": "HA folder name"}]}
)
await setup_integration()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(entries[0].entry_id)
await hass.async_block_till_done()
assert entries[0].state is ConfigEntryState.NOT_LOADED
async def test_create_folder_if_missing(
hass: HomeAssistant,
setup_integration: ComponentSetup,
mock_api: MagicMock,
) -> None:
"""Test folder is created if missing."""
# Setup looks up existing folder to make sure it still exists
# and creates it if missing
mock_api.list_files = AsyncMock(return_value={"files": []})
mock_api.create_file = AsyncMock(
return_value={"id": "new folder id", "name": "Home Assistant"}
)
await setup_integration()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].state is ConfigEntryState.LOADED
mock_api.list_files.assert_called_once()
mock_api.create_file.assert_called_once()
async def test_setup_error(
hass: HomeAssistant,
setup_integration: ComponentSetup,
mock_api: MagicMock,
) -> None:
"""Test setup error."""
# Simulate failure looking up existing folder
mock_api.list_files = AsyncMock(side_effect=GoogleDriveApiError("some error"))
await setup_integration()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].state is ConfigEntryState.SETUP_RETRY
@pytest.mark.parametrize("expires_at", [time.time() - 3600], ids=["expired"])
async def test_expired_token_refresh_success(
hass: HomeAssistant,
setup_integration: ComponentSetup,
aioclient_mock: AiohttpClientMocker,
mock_api: MagicMock,
) -> None:
"""Test expired token is refreshed."""
# Setup looks up existing folder to make sure it still exists
mock_api.list_files = AsyncMock(
return_value={"files": [{"id": "HA folder ID", "name": "HA folder name"}]}
)
aioclient_mock.post(
"https://oauth2.googleapis.com/token",
json={
"access_token": "updated-access-token",
"refresh_token": "updated-refresh-token",
"expires_at": time.time() + 3600,
"expires_in": 3600,
},
)
await setup_integration()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].state is ConfigEntryState.LOADED
assert entries[0].data["token"]["access_token"] == "updated-access-token"
assert entries[0].data["token"]["expires_in"] == 3600
@pytest.mark.parametrize(
("expires_at", "status", "expected_state"),
[
(
time.time() - 3600,
http.HTTPStatus.UNAUTHORIZED,
ConfigEntryState.SETUP_ERROR,
),
(
time.time() - 3600,
http.HTTPStatus.INTERNAL_SERVER_ERROR,
ConfigEntryState.SETUP_RETRY,
),
],
ids=["failure_requires_reauth", "transient_failure"],
)
async def test_expired_token_refresh_failure(
hass: HomeAssistant,
setup_integration: ComponentSetup,
aioclient_mock: AiohttpClientMocker,
status: http.HTTPStatus,
expected_state: ConfigEntryState,
) -> None:
"""Test failure while refreshing token with a transient error."""
aioclient_mock.post(
"https://oauth2.googleapis.com/token",
status=status,
)
await setup_integration()
# Verify a transient failure has occurred
entries = hass.config_entries.async_entries(DOMAIN)
assert entries[0].state is expected_state