1
0
mirror of https://github.com/home-assistant/core.git synced 2026-05-08 09:38:58 +01:00

Add sensors to Google Drive (#156167)

Co-authored-by: mik-laj <12058428+mik-laj@users.noreply.github.com>
This commit is contained in:
Kamil Breguła
2025-12-25 06:24:36 +01:00
committed by GitHub
parent 2683b893c4
commit 2d6ae8f907
20 changed files with 1168 additions and 66 deletions
@@ -6,7 +6,7 @@ from collections.abc import Callable
from google_drive_api.exceptions import GoogleDriveApiError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import instance_id
@@ -19,13 +19,13 @@ from homeassistant.util.hass_dict import HassKey
from .api import AsyncConfigEntryAuth, DriveClient
from .const import DOMAIN
from .coordinator import GoogleDriveConfigEntry, GoogleDriveDataUpdateCoordinator
DATA_BACKUP_AGENT_LISTENERS: HassKey[list[Callable[[], None]]] = HassKey(
f"{DOMAIN}.backup_agent_listeners"
)
type GoogleDriveConfigEntry = ConfigEntry[DriveClient]
_PLATFORMS = (Platform.SENSOR,)
async def async_setup_entry(hass: HomeAssistant, entry: GoogleDriveConfigEntry) -> bool:
@@ -41,11 +41,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: GoogleDriveConfigEntry)
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()
folder_id, _ = await client.async_create_ha_root_folder_if_not_exists()
except GoogleDriveApiError as err:
raise ConfigEntryNotReady from err
@@ -55,6 +54,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: GoogleDriveConfigEntry)
entry.async_on_unload(entry.async_on_state_change(async_notify_backup_listeners))
entry.runtime_data = GoogleDriveDataUpdateCoordinator(
hass, entry=entry, client=client, backup_folder_id=folder_id
)
await entry.runtime_data.async_config_entry_first_refresh()
await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
return True
@@ -62,4 +68,6 @@ async def async_unload_entry(
hass: HomeAssistant, entry: GoogleDriveConfigEntry
) -> bool:
"""Unload a config entry."""
await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)
return True
@@ -3,6 +3,7 @@
from __future__ import annotations
from collections.abc import AsyncIterator, Callable, Coroutine
from dataclasses import dataclass
import json
import logging
from typing import Any
@@ -27,6 +28,16 @@ _UPLOAD_MAX_RETRIES = 20
_LOGGER = logging.getLogger(__name__)
@dataclass
class StorageQuotaData:
"""Class to represent storage quota data."""
limit: int | None
usage: int
usage_in_drive: int
usage_in_trash: int
class AsyncConfigEntryAuth(AbstractAuth):
"""Provide Google Drive authentication tied to an OAuth2 based config entry."""
@@ -95,6 +106,19 @@ class DriveClient:
res = await self._api.get_user(params={"fields": "user(emailAddress)"})
return str(res["user"]["emailAddress"])
async def async_get_storage_quota(self) -> StorageQuotaData:
"""Get storage quota of the current user."""
res = await self._api.get_user(params={"fields": "storageQuota"})
storageQuota = res["storageQuota"]
limit = storageQuota.get("limit")
return StorageQuotaData(
limit=int(limit) if limit is not None else None,
usage=int(storageQuota.get("usage", 0)),
usage_in_drive=int(storageQuota.get("usageInDrive", 0)),
usage_in_trash=int(storageQuota.get("usageInTrash", 0)),
)
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"
@@ -178,6 +202,12 @@ class DriveClient:
backups.append(backup)
return backups
async def async_get_size_of_all_backups(self) -> int:
"""Get size of all backups."""
backups = await self.async_list_backups()
return sum(backup.size for backup in 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(
@@ -68,7 +68,7 @@ class GoogleDriveBackupAgent(BackupAgent):
assert config_entry.unique_id
self.name = config_entry.title
self.unique_id = slugify(config_entry.unique_id)
self._client = config_entry.runtime_data
self._client = config_entry.runtime_data.client
async def async_upload_backup(
self,
@@ -14,10 +14,9 @@ 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
from .const import DOMAIN, DRIVE_FOLDER_URL_PREFIX
DEFAULT_NAME = "Google Drive"
DRIVE_FOLDER_URL_PREFIX = "https://drive.google.com/drive/folders/"
OAUTH2_SCOPES = [
"https://www.googleapis.com/auth/drive.file",
]
@@ -2,4 +2,9 @@
from __future__ import annotations
from datetime import timedelta
DOMAIN = "google_drive"
SCAN_INTERVAL = timedelta(hours=6)
DRIVE_FOLDER_URL_PREFIX = "https://drive.google.com/drive/folders/"
@@ -0,0 +1,76 @@
"""DataUpdateCoordinator for Google Drive."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from google_drive_api.exceptions import GoogleDriveApiError
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .api import DriveClient, StorageQuotaData
from .const import DOMAIN, SCAN_INTERVAL
type GoogleDriveConfigEntry = ConfigEntry[GoogleDriveDataUpdateCoordinator]
_LOGGER = logging.getLogger(__name__)
@dataclass
class SensorData:
"""Class to represent sensor data."""
storage_quota: StorageQuotaData
all_backups_size: int
class GoogleDriveDataUpdateCoordinator(DataUpdateCoordinator[SensorData]):
"""Class to manage fetching Google Drive data from single endpoint."""
client: DriveClient
config_entry: GoogleDriveConfigEntry
email_address: str
backup_folder_id: str
def __init__(
self,
hass: HomeAssistant,
*,
client: DriveClient,
backup_folder_id: str,
entry: GoogleDriveConfigEntry,
) -> None:
"""Initialize Google Drive data updater."""
self.client = client
self.backup_folder_id = backup_folder_id
super().__init__(
hass,
_LOGGER,
config_entry=entry,
name=DOMAIN,
update_interval=SCAN_INTERVAL,
)
async def _async_setup(self) -> None:
"""Do initialization logic."""
self.email_address = await self.client.async_get_email_address()
async def _async_update_data(self) -> SensorData:
"""Fetch data from Google Drive."""
try:
storage_quota = await self.client.async_get_storage_quota()
all_backups_size = await self.client.async_get_size_of_all_backups()
return SensorData(
storage_quota=storage_quota,
all_backups_size=all_backups_size,
)
except GoogleDriveApiError as error:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="invalid_response_google_drive_error",
translation_placeholders={"error": str(error)},
) from error
@@ -0,0 +1,48 @@
"""Diagnostics support for Google Drive."""
from __future__ import annotations
import dataclasses
from typing import Any
from homeassistant.components.backup import (
DATA_MANAGER as BACKUP_DATA_MANAGER,
BackupManager,
)
from homeassistant.components.diagnostics import async_redact_data
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.core import HomeAssistant
from .const import DOMAIN
from .coordinator import GoogleDriveConfigEntry
TO_REDACT = (CONF_ACCESS_TOKEN, "refresh_token")
async def async_get_config_entry_diagnostics(
hass: HomeAssistant,
entry: GoogleDriveConfigEntry,
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator = entry.runtime_data
backup_manager: BackupManager = hass.data[BACKUP_DATA_MANAGER]
backups = await coordinator.client.async_list_backups()
data = {
"coordinator_data": dataclasses.asdict(coordinator.data),
"config": {
**entry.data,
**entry.options,
},
"backup_folder_id": coordinator.backup_folder_id,
"backup_agents": [
{"name": agent.name}
for agent in backup_manager.backup_agents.values()
if agent.domain == DOMAIN
],
"backup": [backup.as_dict() for backup in backups],
}
return async_redact_data(data, TO_REDACT)
@@ -0,0 +1,25 @@
"""Define the Google Drive entity."""
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, DRIVE_FOLDER_URL_PREFIX
from .coordinator import GoogleDriveDataUpdateCoordinator
class GoogleDriveEntity(CoordinatorEntity[GoogleDriveDataUpdateCoordinator]):
"""Defines a base Google Drive entity."""
_attr_has_entity_name = True
@property
def device_info(self) -> DeviceInfo:
"""Return device information about this Google Drive device."""
return DeviceInfo(
identifiers={(DOMAIN, str(self.coordinator.config_entry.unique_id))},
name=self.coordinator.email_address,
manufacturer="Google",
model="Google Drive",
configuration_url=f"{DRIVE_FOLDER_URL_PREFIX}{self.coordinator.backup_folder_id}",
entry_type=DeviceEntryType.SERVICE,
)
@@ -0,0 +1,21 @@
{
"entity": {
"sensor": {
"backups_size": {
"default": "mdi:database"
},
"storage_total": {
"default": "mdi:database"
},
"storage_used": {
"default": "mdi:database"
},
"storage_used_in_drive": {
"default": "mdi:database"
},
"storage_used_in_drive_trash": {
"default": "mdi:database"
}
}
}
}
@@ -3,9 +3,7 @@ rules:
action-setup:
status: exempt
comment: No actions.
appropriate-polling:
status: exempt
comment: No polling.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
@@ -17,12 +15,8 @@ rules:
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.
entity-event-setup: done
entity-unique-id: done
has-entity-name:
status: exempt
comment: No entities.
@@ -38,39 +32,24 @@ rules:
status: exempt
comment: No configuration options.
docs-installation-parameters: done
entity-unavailable:
status: exempt
comment: No entities.
entity-unavailable: done
integration-owner: done
log-when-unavailable:
status: exempt
comment: No entities.
parallel-updates:
status: exempt
comment: No actions and no entities.
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: done
test-coverage: done
# Gold
devices:
status: exempt
comment: No devices.
diagnostics:
status: exempt
comment: No data to diagnose.
devices: done
diagnostics: done
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-data-update: done
docs-examples: done
docs-known-limitations: done
docs-supported-devices:
status: exempt
@@ -79,20 +58,13 @@ rules:
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.
status: done
comment: |
This integration has a fixed single service.
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
exception-translations: done
icon-translations:
status: exempt
@@ -104,8 +76,9 @@ rules:
status: exempt
comment: No repairs.
stale-devices:
status: exempt
comment: No devices.
status: done
comment: |
This integration has a fixed single service.
# Platinum
async-dependency: done
@@ -0,0 +1,127 @@
"""Support for GoogleDrive sensors."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.const import EntityCategory, UnitOfInformation
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from .coordinator import (
GoogleDriveConfigEntry,
GoogleDriveDataUpdateCoordinator,
SensorData,
)
from .entity import GoogleDriveEntity
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class GoogleDriveSensorEntityDescription(SensorEntityDescription):
"""Describes GoogleDrive sensor entity."""
exists_fn: Callable[[SensorData], bool] = lambda _: True
value_fn: Callable[[SensorData], StateType]
SENSORS: tuple[GoogleDriveSensorEntityDescription, ...] = (
GoogleDriveSensorEntityDescription(
key="storage_total",
translation_key="storage_total",
native_unit_of_measurement=UnitOfInformation.BYTES,
suggested_unit_of_measurement=UnitOfInformation.GIBIBYTES,
suggested_display_precision=0,
device_class=SensorDeviceClass.DATA_SIZE,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: data.storage_quota.limit,
exists_fn=lambda data: data.storage_quota.limit is not None,
),
GoogleDriveSensorEntityDescription(
key="storage_used",
translation_key="storage_used",
native_unit_of_measurement=UnitOfInformation.BYTES,
suggested_unit_of_measurement=UnitOfInformation.GIBIBYTES,
suggested_display_precision=0,
device_class=SensorDeviceClass.DATA_SIZE,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: data.storage_quota.usage,
),
GoogleDriveSensorEntityDescription(
key="storage_used_in_drive",
translation_key="storage_used_in_drive",
native_unit_of_measurement=UnitOfInformation.BYTES,
suggested_unit_of_measurement=UnitOfInformation.GIBIBYTES,
suggested_display_precision=0,
device_class=SensorDeviceClass.DATA_SIZE,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: data.storage_quota.usage_in_drive,
entity_registry_enabled_default=False,
),
GoogleDriveSensorEntityDescription(
key="storage_used_in_drive_trash",
translation_key="storage_used_in_drive_trash",
native_unit_of_measurement=UnitOfInformation.BYTES,
suggested_unit_of_measurement=UnitOfInformation.GIBIBYTES,
suggested_display_precision=0,
device_class=SensorDeviceClass.DATA_SIZE,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: data.storage_quota.usage_in_trash,
entity_registry_enabled_default=False,
),
GoogleDriveSensorEntityDescription(
key="backups_size",
translation_key="backups_size",
native_unit_of_measurement=UnitOfInformation.BYTES,
suggested_unit_of_measurement=UnitOfInformation.MEBIBYTES,
suggested_display_precision=0,
device_class=SensorDeviceClass.DATA_SIZE,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda data: data.all_backups_size,
entity_registry_enabled_default=False,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: GoogleDriveConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up GoogleDrive sensor based on a config entry."""
coordinator = entry.runtime_data
async_add_entities(
GoogleDriveSensorEntity(coordinator, description)
for description in SENSORS
if description.exists_fn(coordinator.data)
)
class GoogleDriveSensorEntity(GoogleDriveEntity, SensorEntity):
"""Defines a Google Drive sensor entity."""
entity_description: GoogleDriveSensorEntityDescription
def __init__(
self,
coordinator: GoogleDriveDataUpdateCoordinator,
description: GoogleDriveSensorEntityDescription,
) -> None:
"""Initialize a Google Drive sensor entity."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.config_entry.unique_id}_{description.key}"
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data)
@@ -42,5 +42,24 @@
"title": "[%key:common::config_flow::title::reauth%]"
}
}
},
"entity": {
"sensor": {
"backups_size": {
"name": "Total size of backups"
},
"storage_total": {
"name": "Total available storage"
},
"storage_used": {
"name": "Used storage"
},
"storage_used_in_drive": {
"name": "Used storage in Drive"
},
"storage_used_in_drive_trash": {
"name": "Used storage in Drive Trash"
}
}
}
}
+59
View File
@@ -10,9 +10,12 @@ from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.backup import AddonInfo, AgentBackup
from homeassistant.components.google_drive.const import DOMAIN
from homeassistant.const import UnitOfInformation
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util.unit_conversion import InformationConverter
from tests.common import MockConfigEntry
@@ -42,6 +45,40 @@ def mock_api() -> Generator[MagicMock]:
"homeassistant.components.google_drive.api.GoogleDriveApi"
) as mock_api_cl:
mock_api = mock_api_cl.return_value
def mock_get_user(params=None):
params = params or {}
fields = params.get("fields")
result = {}
if not fields or "storageQuota" in fields:
result["storageQuota"] = {
"limit": InformationConverter.convert(
10, UnitOfInformation.GIBIBYTES, UnitOfInformation.BYTES
),
"usage": InformationConverter.convert(
5, UnitOfInformation.GIBIBYTES, UnitOfInformation.BYTES
),
"usageInDrive": InformationConverter.convert(
2, UnitOfInformation.GIBIBYTES, UnitOfInformation.BYTES
),
"usageInTrash": InformationConverter.convert(
1, UnitOfInformation.GIBIBYTES, UnitOfInformation.BYTES
),
}
if not fields or "user(emailAddress)" in fields:
result["user"] = {"emailAddress": TEST_USER_EMAIL}
return result
mock_api.get_user = AsyncMock(side_effect=mock_get_user)
# Setup looks up existing folder to make sure it still exists
# and list backups during coordinator update
mock_api.list_files = AsyncMock(
side_effect=[
{"files": [{"id": "HA folder ID", "name": "HA folder name"}]},
{"files": []},
]
)
yield mock_api
@@ -78,3 +115,25 @@ def mock_config_entry(expires_at: int) -> MockConfigEntry:
},
},
)
@pytest.fixture
def mock_agent_backup() -> AgentBackup:
"""Return a mocked AgentBackup."""
return 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=InformationConverter.convert(
100, UnitOfInformation.MEBIBYTES, UnitOfInformation.BYTES
),
)
@@ -12,6 +12,37 @@
}),
}),
),
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'user(emailAddress)',
}),
}),
),
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'storageQuota',
}),
}),
),
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(
@@ -46,6 +77,37 @@
}),
}),
),
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'user(emailAddress)',
}),
}),
),
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'storageQuota',
}),
}),
),
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(
@@ -98,6 +160,37 @@
}),
}),
),
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'user(emailAddress)',
}),
}),
),
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'storageQuota',
}),
}),
),
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(
@@ -124,6 +217,37 @@
}),
}),
),
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'user(emailAddress)',
}),
}),
),
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'storageQuota',
}),
}),
),
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(
@@ -179,6 +303,37 @@
}),
}),
),
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'user(emailAddress)',
}),
}),
),
tuple(
'get_user',
tuple(
),
dict({
'params': dict({
'fields': 'storageQuota',
}),
}),
),
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(
@@ -0,0 +1,53 @@
# serializer version: 1
# name: test_entry_diagnostics
dict({
'backup': list([
dict({
'addons': list([
dict({
'name': 'Test',
'slug': 'test',
'version': '1.0.0',
}),
]),
'backup_id': 'test-backup',
'database_included': True,
'date': '2025-01-01T01:23:45.678Z',
'extra_metadata': dict({
'with_automatic_settings': False,
}),
'folders': list([
]),
'homeassistant_included': True,
'homeassistant_version': '2024.12.0',
'name': 'Test',
'protected': False,
'size': 104857600.0,
}),
]),
'backup_agents': list([
dict({
'name': 'Google Drive entry title',
}),
]),
'backup_folder_id': 'HA folder ID',
'config': dict({
'auth_implementation': 'google_drive',
'token': dict({
'access_token': '**REDACTED**',
'expires_at': 1636047419.0,
'refresh_token': '**REDACTED**',
'scope': 'https://www.googleapis.com/auth/drive.file',
}),
}),
'coordinator_data': dict({
'all_backups_size': 104857600.0,
'storage_quota': dict({
'limit': 10737418240,
'usage': 5368709120,
'usage_in_drive': 2147483648,
'usage_in_trash': 1073741824,
}),
}),
})
# ---
@@ -0,0 +1,312 @@
# serializer version: 1
# name: test_sensor.10
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': 'https://drive.google.com/drive/folders/HA folder ID',
'connections': set({
}),
'disabled_by': None,
'entry_type': <DeviceEntryType.SERVICE: 'service'>,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'google_drive',
'testuser@domain.com',
),
}),
'labels': set({
}),
'manufacturer': 'Google',
'model': 'Google Drive',
'model_id': None,
'name': 'testuser@domain.com',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'sw_version': None,
'via_device_id': None,
})
# ---
# name: test_sensor[sensor.testuser_domain_com_total_available_storage-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.testuser_domain_com_total_available_storage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_SIZE: 'data_size'>,
'original_icon': None,
'original_name': 'Total available storage',
'platform': 'google_drive',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'storage_total',
'unique_id': 'testuser@domain.com_storage_total',
'unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
})
# ---
# name: test_sensor[sensor.testuser_domain_com_total_available_storage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_size',
'friendly_name': 'testuser@domain.com Total available storage',
'unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
}),
'context': <ANY>,
'entity_id': 'sensor.testuser_domain_com_total_available_storage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '10.0',
})
# ---
# name: test_sensor[sensor.testuser_domain_com_total_size_of_backups-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.testuser_domain_com_total_size_of_backups',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfInformation.MEBIBYTES: 'MiB'>,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_SIZE: 'data_size'>,
'original_icon': None,
'original_name': 'Total size of backups',
'platform': 'google_drive',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'backups_size',
'unique_id': 'testuser@domain.com_backups_size',
'unit_of_measurement': <UnitOfInformation.MEBIBYTES: 'MiB'>,
})
# ---
# name: test_sensor[sensor.testuser_domain_com_total_size_of_backups-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_size',
'friendly_name': 'testuser@domain.com Total size of backups',
'unit_of_measurement': <UnitOfInformation.MEBIBYTES: 'MiB'>,
}),
'context': <ANY>,
'entity_id': 'sensor.testuser_domain_com_total_size_of_backups',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.0',
})
# ---
# name: test_sensor[sensor.testuser_domain_com_used_storage-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.testuser_domain_com_used_storage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_SIZE: 'data_size'>,
'original_icon': None,
'original_name': 'Used storage',
'platform': 'google_drive',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'storage_used',
'unique_id': 'testuser@domain.com_storage_used',
'unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
})
# ---
# name: test_sensor[sensor.testuser_domain_com_used_storage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_size',
'friendly_name': 'testuser@domain.com Used storage',
'unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
}),
'context': <ANY>,
'entity_id': 'sensor.testuser_domain_com_used_storage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '5.0',
})
# ---
# name: test_sensor[sensor.testuser_domain_com_used_storage_in_drive-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.testuser_domain_com_used_storage_in_drive',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_SIZE: 'data_size'>,
'original_icon': None,
'original_name': 'Used storage in Drive',
'platform': 'google_drive',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'storage_used_in_drive',
'unique_id': 'testuser@domain.com_storage_used_in_drive',
'unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
})
# ---
# name: test_sensor[sensor.testuser_domain_com_used_storage_in_drive-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_size',
'friendly_name': 'testuser@domain.com Used storage in Drive',
'unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
}),
'context': <ANY>,
'entity_id': 'sensor.testuser_domain_com_used_storage_in_drive',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2.0',
})
# ---
# name: test_sensor[sensor.testuser_domain_com_used_storage_in_drive_trash-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.testuser_domain_com_used_storage_in_drive_trash',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_SIZE: 'data_size'>,
'original_icon': None,
'original_name': 'Used storage in Drive Trash',
'platform': 'google_drive',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'storage_used_in_drive_trash',
'unique_id': 'testuser@domain.com_storage_used_in_drive_trash',
'unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
})
# ---
# name: test_sensor[sensor.testuser_domain_com_used_storage_in_drive_trash-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_size',
'friendly_name': 'testuser@domain.com Used storage in Drive Trash',
'unit_of_measurement': <UnitOfInformation.GIBIBYTES: 'GiB'>,
}),
'context': <ANY>,
'entity_id': 'sensor.testuser_domain_com_used_storage_in_drive_trash',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '1.0',
})
# ---
+9 -2
View File
@@ -68,11 +68,11 @@ async def setup_integration(
"""Set up Google Drive integration."""
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, BACKUP_DOMAIN, {BACKUP_DOMAIN: {}})
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
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(
@@ -364,6 +364,13 @@ async def test_agents_upload_fail(
mock_api.resumable_upload_file = AsyncMock(
side_effect=GoogleDriveApiError("some error")
)
mock_api.list_files = AsyncMock(
side_effect=[
{"files": [{"id": "HA folder ID", "name": "HA folder name"}]},
{"files": []},
{"files": [{"id": "HA folder ID", "name": "HA folder name"}]},
]
)
client = await hass_client()
@@ -0,0 +1,50 @@
"""Test GIOS diagnostics."""
import json
from unittest.mock import AsyncMock, MagicMock
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.backup import DOMAIN as BACKUP_DOMAIN, AgentBackup
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
pytestmark = [
pytest.mark.freeze_time("2021-11-04 17:36:59+01:00"),
]
async def test_entry_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
mock_api: MagicMock,
snapshot: SnapshotAssertion,
mock_agent_backup: AgentBackup,
) -> None:
"""Test config entry diagnostics."""
mock_api.list_files = AsyncMock(
return_value={
"files": [
{
"id": "HA folder ID",
"name": "HA folder name",
"description": json.dumps(mock_agent_backup.as_dict()),
}
]
}
)
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, BACKUP_DOMAIN, {BACKUP_DOMAIN: {}})
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert (
await get_diagnostics_for_config_entry(hass, hass_client, config_entry)
== snapshot
)
+1 -9
View File
@@ -41,10 +41,6 @@ async def test_setup_success(
) -> 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)
@@ -76,7 +72,7 @@ async def test_create_folder_if_missing(
assert len(entries) == 1
assert entries[0].state is ConfigEntryState.LOADED
mock_api.list_files.assert_called_once()
assert mock_api.list_files.call_count == 2
mock_api.create_file.assert_called_once()
@@ -104,10 +100,6 @@ async def test_expired_token_refresh_success(
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={
@@ -0,0 +1,143 @@
"""Tests for the Google Drive sensor platform."""
import json
from unittest.mock import AsyncMock, MagicMock
from freezegun.api import FrozenDateTimeFactory
from google_drive_api.exceptions import GoogleDriveApiError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.backup import AgentBackup
from homeassistant.components.google_drive.const import SCAN_INTERVAL
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
pytestmark = [
pytest.mark.freeze_time("2021-11-04 17:36:59+01:00"),
]
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up Google Drive integration."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensor(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
config_entry: MockConfigEntry,
mock_api: MagicMock,
) -> None:
"""Test the creation and values of the Google Drive sensors."""
await setup_integration(hass, config_entry)
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
assert (
entity_entry := entity_registry.async_get(
"sensor.testuser_domain_com_total_available_storage"
)
)
assert entity_entry.device_id
assert (device_entry := device_registry.async_get(entity_entry.device_id))
assert device_entry == snapshot
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensor_unknown_when_unlimited_plan(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_api: MagicMock,
) -> None:
"""Test the total storage are unknown when the user is on an unlimited plan."""
mock_api.get_user = AsyncMock(
return_value={
"storageQuota": {
"limit": None,
"usage": "100",
"usageInDrive": "50",
"usageInTrash": "10",
}
}
)
assert not hass.states.get("sensor.testuser_domain_com_total_available_storage")
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensor_availability(
hass: HomeAssistant,
mock_api: MagicMock,
config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the availability handling of the Google Drive sensors."""
await setup_integration(hass, config_entry)
mock_api.get_user.side_effect = GoogleDriveApiError("API error")
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (
state := hass.states.get("sensor.testuser_domain_com_total_available_storage")
)
assert state.state == STATE_UNAVAILABLE
mock_api.list_files.side_effect = [{"files": []}]
mock_api.get_user.side_effect = None
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (
state := hass.states.get("sensor.testuser_domain_com_total_available_storage")
)
assert state.state != STATE_UNAVAILABLE
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_calculate_backups_size(
hass: HomeAssistant,
mock_api: MagicMock,
config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
mock_agent_backup: AgentBackup,
) -> None:
"""Test the availability handling of the Google Drive sensors."""
await setup_integration(hass, config_entry)
assert (
state := hass.states.get("sensor.testuser_domain_com_total_size_of_backups")
)
assert state.state == "0.0"
mock_api.list_files = AsyncMock(
return_value={
"files": [
{
"id": "HA folder ID",
"name": "HA folder name",
"description": json.dumps(mock_agent_backup.as_dict()),
}
]
}
)
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (
state := hass.states.get("sensor.testuser_domain_com_total_size_of_backups")
)
assert state.state == "100.0"