Add sensor platform to OneDrive for drive usage (#138232)

This commit is contained in:
Josef Zweck 2025-02-12 18:37:30 +01:00 committed by GitHub
parent 620141cfb1
commit ff5ddce7b0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 647 additions and 82 deletions

View File

@ -2,8 +2,6 @@
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from html import unescape
from json import dumps, loads
import logging
@ -17,8 +15,7 @@ from onedrive_personal_sdk.exceptions import (
)
from onedrive_personal_sdk.models.items import ItemUpdate
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.const import CONF_ACCESS_TOKEN, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@ -29,18 +26,14 @@ from homeassistant.helpers.config_entry_oauth2_flow import (
from homeassistant.helpers.instance_id import async_get as async_get_instance_id
from .const import DATA_BACKUP_AGENT_LISTENERS, DOMAIN
from .coordinator import (
OneDriveConfigEntry,
OneDriveRuntimeData,
OneDriveUpdateCoordinator,
)
PLATFORMS = [Platform.SENSOR]
@dataclass
class OneDriveRuntimeData:
"""Runtime data for the OneDrive integration."""
client: OneDriveClient
token_function: Callable[[], Awaitable[str]]
backup_folder_id: str
type OneDriveConfigEntry = ConfigEntry[OneDriveRuntimeData]
_LOGGER = logging.getLogger(__name__)
@ -85,10 +78,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: OneDriveConfigEntry) ->
translation_placeholders={"folder": backup_folder_name},
) from err
coordinator = OneDriveUpdateCoordinator(hass, entry, client)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = OneDriveRuntimeData(
client=client,
token_function=get_access_token,
backup_folder_id=backup_folder.id,
coordinator=coordinator,
)
try:
@ -100,6 +97,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: OneDriveConfigEntry) ->
) from err
_async_notify_backup_listeners_soon(hass)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
@ -107,7 +105,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: OneDriveConfigEntry) ->
async def async_unload_entry(hass: HomeAssistant, entry: OneDriveConfigEntry) -> bool:
"""Unload a OneDrive config entry."""
_async_notify_backup_listeners_soon(hass)
return True
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
def _async_notify_backup_listeners(hass: HomeAssistant) -> None:

View File

@ -30,8 +30,8 @@ from homeassistant.components.backup import (
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from . import OneDriveConfigEntry
from .const import DATA_BACKUP_AGENT_LISTENERS, DOMAIN
from .coordinator import OneDriveConfigEntry
_LOGGER = logging.getLogger(__name__)
UPLOAD_CHUNK_SIZE = 16 * 320 * 1024 # 5.2MB

View File

@ -0,0 +1,70 @@
"""Coordinator for OneDrive."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import timedelta
import logging
from onedrive_personal_sdk import OneDriveClient
from onedrive_personal_sdk.exceptions import AuthenticationError, OneDriveException
from onedrive_personal_sdk.models.items import Drive
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
SCAN_INTERVAL = timedelta(minutes=5)
_LOGGER = logging.getLogger(__name__)
@dataclass
class OneDriveRuntimeData:
"""Runtime data for the OneDrive integration."""
client: OneDriveClient
token_function: Callable[[], Awaitable[str]]
backup_folder_id: str
coordinator: OneDriveUpdateCoordinator
type OneDriveConfigEntry = ConfigEntry[OneDriveRuntimeData]
class OneDriveUpdateCoordinator(DataUpdateCoordinator[Drive]):
"""Class to handle fetching data from the Graph API centrally."""
config_entry: OneDriveConfigEntry
def __init__(
self, hass: HomeAssistant, entry: OneDriveConfigEntry, client: OneDriveClient
) -> None:
"""Initialize coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=entry,
name=DOMAIN,
update_interval=SCAN_INTERVAL,
)
self._client = client
async def _async_update_data(self) -> Drive:
"""Fetch data from API endpoint."""
try:
drive = await self._client.get_drive()
except AuthenticationError as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="authentication_failed"
) from err
except OneDriveException as err:
raise UpdateFailed(
translation_domain=DOMAIN, translation_key="update_failed"
) from err
return drive

View File

@ -0,0 +1,24 @@
{
"entity": {
"sensor": {
"total_size": {
"default": "mdi:database"
},
"used_size": {
"default": "mdi:database"
},
"remaining_size": {
"default": "mdi:database"
},
"drive_state": {
"default": "mdi:harddisk",
"state": {
"normal": "mdi:harddisk",
"nearing": "mdi:alert-circle-outline",
"critical": "mdi:alert",
"exceeded": "mdi:alert-octagon"
}
}
}
}
}

View File

@ -3,10 +3,7 @@ rules:
action-setup:
status: exempt
comment: Integration does not register custom actions.
appropriate-polling:
status: exempt
comment: |
This integration does not poll.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
@ -23,14 +20,8 @@ rules:
status: exempt
comment: |
Entities of this integration does not explicitly subscribe to events.
entity-unique-id:
status: exempt
comment: |
This integration does not have entities.
has-entity-name:
status: exempt
comment: |
This integration does not have entities.
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
@ -44,27 +35,15 @@ rules:
comment: |
No Options flow.
docs-installation-parameters: done
entity-unavailable:
status: exempt
comment: |
This integration does not have entities.
entity-unavailable: done
integration-owner: done
log-when-unavailable:
status: exempt
comment: |
This integration does not have entities.
parallel-updates:
status: exempt
comment: |
This integration does not have platforms.
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: done
test-coverage: todo
# Gold
devices:
status: exempt
comment: |
This integration connects to a single service.
devices: done
diagnostics:
status: exempt
comment: |
@ -77,53 +56,26 @@ rules:
status: exempt
comment: |
This integration is a cloud service and does not support discovery.
docs-data-update:
status: exempt
comment: |
This integration does not poll or push.
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
comment: |
This integration is a cloud service.
docs-supported-functions:
status: exempt
comment: |
This integration does not have entities.
docs-troubleshooting:
status: exempt
comment: |
No issues known to troubleshoot.
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
dynamic-devices:
status: exempt
comment: |
This integration connects to a single service.
entity-category:
status: exempt
comment: |
This integration does not have entities.
entity-device-class:
status: exempt
comment: |
This integration does not have entities.
entity-disabled-by-default:
status: exempt
comment: |
This integration does not have entities.
entity-translations:
status: exempt
comment: |
This integration does not have entities.
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
exception-translations: done
icon-translations:
status: exempt
comment: |
This integration does not have entities.
icon-translations: done
reconfiguration-flow:
status: exempt
comment: |

View File

@ -0,0 +1,122 @@
"""Sensors for OneDrive."""
from collections.abc import Callable
from dataclasses import dataclass
from onedrive_personal_sdk.const import DriveState
from onedrive_personal_sdk.models.items import DriveQuota
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.const import EntityCategory, UnitOfInformation
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import OneDriveConfigEntry, OneDriveUpdateCoordinator
PARALLEL_UPDATES = 0
@dataclass(kw_only=True, frozen=True)
class OneDriveSensorEntityDescription(SensorEntityDescription):
"""Describes OneDrive sensor entity."""
value_fn: Callable[[DriveQuota], StateType]
DRIVE_STATE_ENTITIES: tuple[OneDriveSensorEntityDescription, ...] = (
OneDriveSensorEntityDescription(
key="total_size",
value_fn=lambda quota: quota.total,
native_unit_of_measurement=UnitOfInformation.BYTES,
suggested_unit_of_measurement=UnitOfInformation.GIGABYTES,
suggested_display_precision=0,
device_class=SensorDeviceClass.DATA_SIZE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
OneDriveSensorEntityDescription(
key="used_size",
value_fn=lambda quota: quota.used,
native_unit_of_measurement=UnitOfInformation.BYTES,
suggested_unit_of_measurement=UnitOfInformation.GIGABYTES,
suggested_display_precision=2,
device_class=SensorDeviceClass.DATA_SIZE,
entity_category=EntityCategory.DIAGNOSTIC,
),
OneDriveSensorEntityDescription(
key="remaining_size",
value_fn=lambda quota: quota.remaining,
native_unit_of_measurement=UnitOfInformation.BYTES,
suggested_unit_of_measurement=UnitOfInformation.GIGABYTES,
suggested_display_precision=2,
device_class=SensorDeviceClass.DATA_SIZE,
entity_category=EntityCategory.DIAGNOSTIC,
),
OneDriveSensorEntityDescription(
key="drive_state",
value_fn=lambda quota: quota.state.value,
options=[state.value for state in DriveState],
device_class=SensorDeviceClass.ENUM,
entity_category=EntityCategory.DIAGNOSTIC,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: OneDriveConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up OneDrive sensors based on a config entry."""
coordinator = entry.runtime_data.coordinator
async_add_entities(
OneDriveDriveStateSensor(coordinator, description)
for description in DRIVE_STATE_ENTITIES
)
class OneDriveDriveStateSensor(
CoordinatorEntity[OneDriveUpdateCoordinator], SensorEntity
):
"""Define a OneDrive sensor."""
entity_description: OneDriveSensorEntityDescription
_attr_has_entity_name = True
def __init__(
self,
coordinator: OneDriveUpdateCoordinator,
description: OneDriveSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = description
self._attr_translation_key = description.key
self._attr_unique_id = f"{coordinator.data.id}_{description.key}"
self._attr_device_info = DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
name=coordinator.data.name,
identifiers={(DOMAIN, coordinator.data.id)},
manufacturer="Microsoft",
model=f"OneDrive {coordinator.data.drive_type.value.capitalize()}",
configuration_url=f"https://onedrive.live.com/?id=root&cid={coordinator.data.id}",
)
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
assert self.coordinator.data.quota
return self.entity_description.value_fn(self.coordinator.data.quota)
@property
def available(self) -> bool:
"""Availability of the sensor."""
return super().available and self.coordinator.data.quota is not None

View File

@ -38,6 +38,31 @@
},
"failed_to_migrate_files": {
"message": "Failed to migrate metadata to separate files"
},
"update_failed": {
"message": "Failed to update drive state"
}
},
"entity": {
"sensor": {
"total_size": {
"name": "Total available storage"
},
"used_size": {
"name": "Used storage"
},
"remaining_size": {
"name": "Remaining storage"
},
"drive_state": {
"name": "Drive state",
"state": {
"normal": "Normal",
"nearing": "Nearing limit",
"critical": "Critical",
"exceeded": "Exceeded"
}
}
}
}
}

View File

@ -22,6 +22,7 @@ from .const import (
MOCK_APPROOT,
MOCK_BACKUP_FILE,
MOCK_BACKUP_FOLDER,
MOCK_DRIVE,
MOCK_METADATA_FILE,
)
@ -104,7 +105,7 @@ def mock_onedrive_client(mock_onedrive_client_init: MagicMock) -> Generator[Magi
return dumps(BACKUP_METADATA).encode()
client.download_drive_item.return_value = MockStreamReader()
client.get_drive.return_value = MOCK_DRIVE
return client

View File

@ -3,8 +3,11 @@
from html import escape
from json import dumps
from onedrive_personal_sdk.const import DriveState, DriveType
from onedrive_personal_sdk.models.items import (
AppRoot,
Drive,
DriveQuota,
File,
Folder,
Hashes,
@ -98,3 +101,18 @@ MOCK_METADATA_FILE = File(
),
created_by=IDENTITY_SET,
)
MOCK_DRIVE = Drive(
id="mock_drive_id",
name="My Drive",
drive_type=DriveType.PERSONAL,
owner=IDENTITY_SET,
quota=DriveQuota(
deleted=5,
remaining=750000000,
state=DriveState.NEARING,
total=5000000000,
used=4250000000,
),
)

View File

@ -0,0 +1,34 @@
# serializer version: 1
# name: test_device
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': 'https://onedrive.live.com/?id=root&cid=mock_drive_id',
'connections': set({
}),
'disabled_by': None,
'entry_type': <DeviceEntryType.SERVICE: 'service'>,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'onedrive',
'mock_drive_id',
),
}),
'is_new': False,
'labels': set({
}),
'manufacturer': 'Microsoft',
'model': 'OneDrive Personal',
'model_id': None,
'name': 'My Drive',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'suggested_area': None,
'sw_version': None,
'via_device_id': None,
})
# ---

View File

@ -0,0 +1,227 @@
# serializer version: 1
# name: test_sensors[sensor.my_drive_drive_state-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'normal',
'nearing',
'critical',
'exceeded',
]),
}),
'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.my_drive_drive_state',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Drive state',
'platform': 'onedrive',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'drive_state',
'unique_id': 'mock_drive_id_drive_state',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[sensor.my_drive_drive_state-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'My Drive Drive state',
'options': list([
'normal',
'nearing',
'critical',
'exceeded',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.my_drive_drive_state',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'nearing',
})
# ---
# name: test_sensors[sensor.my_drive_remaining_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.my_drive_remaining_storage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_SIZE: 'data_size'>,
'original_icon': None,
'original_name': 'Remaining storage',
'platform': 'onedrive',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'remaining_size',
'unique_id': 'mock_drive_id_remaining_size',
'unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
})
# ---
# name: test_sensors[sensor.my_drive_remaining_storage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_size',
'friendly_name': 'My Drive Remaining storage',
'unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
}),
'context': <ANY>,
'entity_id': 'sensor.my_drive_remaining_storage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.75',
})
# ---
# name: test_sensors[sensor.my_drive_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.my_drive_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.GIGABYTES: 'GB'>,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_SIZE: 'data_size'>,
'original_icon': None,
'original_name': 'Total available storage',
'platform': 'onedrive',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'total_size',
'unique_id': 'mock_drive_id_total_size',
'unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
})
# ---
# name: test_sensors[sensor.my_drive_total_available_storage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_size',
'friendly_name': 'My Drive Total available storage',
'unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
}),
'context': <ANY>,
'entity_id': 'sensor.my_drive_total_available_storage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '5.0',
})
# ---
# name: test_sensors[sensor.my_drive_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.my_drive_used_storage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
}),
}),
'original_device_class': <SensorDeviceClass.DATA_SIZE: 'data_size'>,
'original_icon': None,
'original_name': 'Used storage',
'platform': 'onedrive',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'used_size',
'unique_id': 'mock_drive_id_used_size',
'unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
})
# ---
# name: test_sensors[sensor.my_drive_used_storage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'data_size',
'friendly_name': 'My Drive Used storage',
'unit_of_measurement': <UnitOfInformation.GIGABYTES: 'GB'>,
}),
'context': <ANY>,
'entity_id': 'sensor.my_drive_used_storage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '4.25',
})
# ---

View File

@ -6,12 +6,15 @@ from unittest.mock import MagicMock
from onedrive_personal_sdk.exceptions import AuthenticationError, OneDriveException
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.onedrive.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import setup_integration
from .const import BACKUP_METADATA, MOCK_BACKUP_FILE
from .const import BACKUP_METADATA, MOCK_BACKUP_FILE, MOCK_DRIVE
from tests.common import MockConfigEntry
@ -101,3 +104,30 @@ async def test_migrate_metadata_files_errors(
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_auth_error_during_update(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_onedrive_client: MagicMock,
) -> None:
"""Test auth error during update."""
mock_onedrive_client.get_drive.side_effect = AuthenticationError(403, "Auth failed")
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
async def test_device(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test the device."""
await setup_integration(hass, mock_config_entry)
device = device_registry.async_get_device({(DOMAIN, MOCK_DRIVE.id)})
assert device
assert device == snapshot

View File

@ -0,0 +1,64 @@
"""Tests for OneDrive sensors."""
from datetime import timedelta
from typing import Any
from unittest.mock import MagicMock
from freezegun.api import FrozenDateTimeFactory
from onedrive_personal_sdk.const import DriveType
from onedrive_personal_sdk.exceptions import HttpRequestException
from onedrive_personal_sdk.models.items import Drive
import pytest
from syrupy import SnapshotAssertion
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensors(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test the OneDrive sensors."""
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
("attr", "side_effect"),
[
("side_effect", HttpRequestException(503, "Service Unavailable")),
("return_value", Drive(id="id", name="name", drive_type=DriveType.PERSONAL)),
],
)
async def test_update_failure(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_onedrive_client: MagicMock,
freezer: FrozenDateTimeFactory,
attr: str,
side_effect: Any,
) -> None:
"""Ensure sensors are going unavailable on update failure."""
await setup_integration(hass, mock_config_entry)
state = hass.states.get("sensor.my_drive_remaining_storage")
assert state.state == "0.75"
setattr(mock_onedrive_client.get_drive, attr, side_effect)
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("sensor.my_drive_remaining_storage")
assert state.state == STATE_UNAVAILABLE