Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00
Code Activity

Add integration for onedrive for business (#155709)

Browse Source 
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Josef Zweck
2026-02-13 07:01:52 -08:00
committed by GitHub
parent 267caf2365
commit 1bb31892c2
23 changed files with 1937 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@@ -385,6 +385,7 @@ homeassistant.components.ohme.*
homeassistant.components.onboarding.*
homeassistant.components.oncue.*
homeassistant.components.onedrive.*
homeassistant.components.onedrive_for_business.*
homeassistant.components.onewire.*
homeassistant.components.onkyo.*
homeassistant.components.open_meteo.*

2
CODEOWNERS generated
View File

@@ -1180,6 +1180,8 @@ build.json @home-assistant/supervisor
/tests/components/ondilo_ico/ @JeromeHXP
/homeassistant/components/onedrive/ @zweckj
/tests/components/onedrive/ @zweckj
/homeassistant/components/onedrive_for_business/ @zweckj
/tests/components/onedrive_for_business/ @zweckj
/homeassistant/components/onewire/ @garbled1 @epenet
/tests/components/onewire/ @garbled1 @epenet
/homeassistant/components/onkyo/ @arturpragacz @eclair4151

1
homeassistant/brands/microsoft.json
View File

@@ -13,6 +13,7 @@
"microsoft",
"msteams",
"onedrive",
"onedrive_for_business",
"xbox"
]
}

122
homeassistant/components/onedrive_for_business/__init__.py Normal file
View File

@@ -0,0 +1,122 @@
"""The OneDrive for Business integration."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
import logging
from typing import cast
from onedrive_personal_sdk import OneDriveClient
from onedrive_personal_sdk.exceptions import (
AuthenticationError,
NotFoundError,
OneDriveException,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.config_entry_oauth2_flow import (
OAuth2Session,
async_get_config_entry_implementation,
)
from .application_credentials import tenant_id_context
from .const import (
CONF_FOLDER_ID,
CONF_FOLDER_PATH,
CONF_TENANT_ID,
DATA_BACKUP_AGENT_LISTENERS,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
@dataclass
class OneDriveRuntimeData:
"""Runtime data for the OneDrive integration."""
client: OneDriveClient
token_function: Callable[[], Awaitable[str]]
type OneDriveConfigEntry = ConfigEntry[OneDriveRuntimeData]
async def async_setup_entry(hass: HomeAssistant, entry: OneDriveConfigEntry) -> bool:
"""Set up OneDrive from a config entry."""
client, get_access_token = await _get_onedrive_client(hass, entry)
folder_path = entry.data[CONF_FOLDER_PATH]
try:
backup_folder = await _handle_item_operation(
lambda: client.get_drive_item(path_or_id=entry.data[CONF_FOLDER_ID]),
folder_path,
)
except NotFoundError:
_LOGGER.info("Creating backup folder %s", folder_path)
backup_folder = await _handle_item_operation(
lambda: client.create_folder(parent_id="root", name=folder_path),
folder_path,
)
hass.config_entries.async_update_entry(
entry, data={**entry.data, CONF_FOLDER_ID: backup_folder.id}
)
entry.runtime_data = OneDriveRuntimeData(
client=client,
token_function=get_access_token,
)
def async_notify_backup_listeners() -> None:
for listener in hass.data.get(DATA_BACKUP_AGENT_LISTENERS, []):
listener()
entry.async_on_unload(entry.async_on_state_change(async_notify_backup_listeners))
return True
async def async_unload_entry(hass: HomeAssistant, _: OneDriveConfigEntry) -> bool:
"""Unload a OneDrive config entry."""
return True
async def _get_onedrive_client(
hass: HomeAssistant, entry: OneDriveConfigEntry
) -> tuple[OneDriveClient, Callable[[], Awaitable[str]]]:
"""Get OneDrive client."""
with tenant_id_context(entry.data[CONF_TENANT_ID]):
implementation = await async_get_config_entry_implementation(hass, entry)
session = OAuth2Session(hass, entry, implementation)
async def get_access_token() -> str:
await session.async_ensure_token_valid()
return cast(str, session.token[CONF_ACCESS_TOKEN])
return (
OneDriveClient(get_access_token, async_get_clientsession(hass)),
get_access_token,
)
async def _handle_item_operation[T](func: Callable[[], Awaitable[T]], folder: str) -> T:
try:
return await func()
except NotFoundError:
raise
except AuthenticationError as err:
raise ConfigEntryError(
translation_domain=DOMAIN, translation_key="authentication_failed"
) from err
except (OneDriveException, TimeoutError) as err:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="failed_to_get_folder",
translation_placeholders={"folder": folder},
) from err

33
homeassistant/components/onedrive_for_business/application_credentials.py Normal file
View File

@@ -0,0 +1,33 @@
"""Application credentials platform for the OneDrive for Business integration."""
from __future__ import annotations
from collections.abc import Generator
from contextlib import contextmanager
from contextvars import ContextVar
from homeassistant.components.application_credentials import AuthorizationServer
from homeassistant.core import HomeAssistant
from .const import CONF_TENANT_ID, DOMAIN, OAUTH2_AUTHORIZE, OAUTH2_TOKEN
_tenant_id_context: ContextVar[str] = ContextVar(f"{DOMAIN}_{CONF_TENANT_ID}")
@contextmanager
def tenant_id_context(tenant_id: str) -> Generator[None]:
"""Context manager for setting the active tenant ID."""
token = _tenant_id_context.set(tenant_id)
try:
yield
finally:
_tenant_id_context.reset(token)
async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
"""Return authorization server."""
tenant_id = _tenant_id_context.get()
return AuthorizationServer(
authorize_url=OAUTH2_AUTHORIZE.format(tenant_id=tenant_id),
token_url=OAUTH2_TOKEN.format(tenant_id=tenant_id),
)

278
homeassistant/components/onedrive_for_business/backup.py Normal file
View File

@@ -0,0 +1,278 @@
"""Support for OneDrive backup."""
from __future__ import annotations
from collections.abc import AsyncIterator, Callable, Coroutine
from functools import wraps
import logging
from time import time
from typing import Any, Concatenate
from aiohttp import ClientTimeout
from onedrive_personal_sdk.clients.large_file_upload import LargeFileUploadClient
from onedrive_personal_sdk.exceptions import (
AuthenticationError,
HashMismatchError,
OneDriveException,
)
from onedrive_personal_sdk.models.upload import FileInfo
from homeassistant.components.backup import (
AgentBackup,
BackupAgent,
BackupAgentError,
BackupNotFound,
suggested_filename,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.json import json_dumps
from homeassistant.util.json import json_loads_object
from . import OneDriveConfigEntry
from .const import CONF_FOLDER_ID, DATA_BACKUP_AGENT_LISTENERS, DOMAIN
_LOGGER = logging.getLogger(__name__)
MAX_CHUNK_SIZE = 60 * 1024 * 1024 # largest chunk possible, must be <= 60 MiB
TARGET_CHUNKS = 20
TIMEOUT = ClientTimeout(connect=10, total=43200) # 12 hours
CACHE_TTL = 300
async def async_get_backup_agents(
hass: HomeAssistant,
) -> list[BackupAgent]:
"""Return a list of backup agents."""
entries: list[OneDriveConfigEntry] = hass.config_entries.async_loaded_entries(
DOMAIN
)
return [OneDriveBackupAgent(hass, 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."""
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
def handle_backup_errors[_R, **P](
func: Callable[Concatenate[OneDriveBackupAgent, P], Coroutine[Any, Any, _R]],
) -> Callable[Concatenate[OneDriveBackupAgent, P], Coroutine[Any, Any, _R]]:
"""Handle backup errors."""
@wraps(func)
async def wrapper(
self: OneDriveBackupAgent, *args: P.args, **kwargs: P.kwargs
) -> _R:
try:
return await func(self, *args, **kwargs)
except AuthenticationError as err:
raise BackupAgentError("Authentication error") from err
except OneDriveException as err:
_LOGGER.error(
"Error during backup in %s:, message %s",
func.__name__,
err,
)
_LOGGER.debug("Full error: %s", err, exc_info=True)
raise BackupAgentError("Backup operation failed") from err
except TimeoutError as err:
_LOGGER.error(
"Error during backup in %s: Timeout",
func.__name__,
)
raise BackupAgentError("Backup operation timed out") from err
return wrapper
def suggested_filenames(backup: AgentBackup) -> tuple[str, str]:
"""Return the suggested filenames for the backup and metadata."""
base_name = suggested_filename(backup).rsplit(".", 1)[0]
return f"{base_name}.tar", f"{base_name}.metadata.json"
class OneDriveBackupAgent(BackupAgent):
"""OneDrive backup agent."""
domain = DOMAIN
def __init__(self, hass: HomeAssistant, entry: OneDriveConfigEntry) -> None:
"""Initialize the OneDrive backup agent."""
super().__init__()
self._hass = hass
self._entry = entry
self._client = entry.runtime_data.client
self._token_function = entry.runtime_data.token_function
self._folder_id = entry.data[CONF_FOLDER_ID]
self.name = entry.title
assert entry.unique_id
self.unique_id = entry.unique_id
self._cache_backup_metadata: dict[str, AgentBackup] = {}
self._cache_expiration = time()
@handle_backup_errors
async def async_download_backup(
self, backup_id: str, **kwargs: Any
) -> AsyncIterator[bytes]:
"""Download a backup file."""
backup = await self._find_backup_by_id(backup_id)
backup_filename, _ = suggested_filenames(backup)
stream = await self._client.download_drive_item(
f"{self._folder_id}:/{backup_filename}:", timeout=TIMEOUT
)
return stream.iter_chunked(1024)
@handle_backup_errors
async def async_upload_backup(
self,
*,
open_stream: Callable[[], Coroutine[Any, Any, AsyncIterator[bytes]]],
backup: AgentBackup,
**kwargs: Any,
) -> None:
"""Upload a backup."""
backup_filename, metadata_filename = suggested_filenames(backup)
file = FileInfo(
backup_filename,
backup.size,
self._folder_id,
await open_stream(),
)
# determine chunk based on target chunks
upload_chunk_size = backup.size / TARGET_CHUNKS
# find the nearest multiple of 320KB
upload_chunk_size = round(upload_chunk_size / (320 * 1024)) * (320 * 1024)
# limit to max chunk size
upload_chunk_size = min(upload_chunk_size, MAX_CHUNK_SIZE)
# ensure minimum chunk size of 320KB
upload_chunk_size = max(upload_chunk_size, 320 * 1024)
try:
await LargeFileUploadClient.upload(
self._token_function,
file,
upload_chunk_size=upload_chunk_size,
session=async_get_clientsession(self._hass),
smart_chunk_size=True,
)
except HashMismatchError as err:
raise BackupAgentError(
"Hash validation failed, backup file might be corrupt"
) from err
_LOGGER.debug("Uploaded backup to %s", backup_filename)
# Store metadata in separate metadata file (just backup.as_dict(), no extra fields)
metadata_content = json_dumps(backup.as_dict())
try:
await self._client.upload_file(
self._folder_id,
metadata_filename,
metadata_content,
)
except OneDriveException:
# Clean up the backup file if metadata upload fails
_LOGGER.debug(
"Uploading metadata failed, deleting backup file %s", backup_filename
)
await self._client.delete_drive_item(
f"{self._folder_id}:/{backup_filename}:"
)
raise
_LOGGER.debug("Uploaded metadata file %s", metadata_filename)
self._cache_expiration = time()
@handle_backup_errors
async def async_delete_backup(
self,
backup_id: str,
**kwargs: Any,
) -> None:
"""Delete a backup file."""
backup = await self._find_backup_by_id(backup_id)
backup_filename, metadata_filename = suggested_filenames(backup)
await self._client.delete_drive_item(f"{self._folder_id}:/{backup_filename}:")
await self._client.delete_drive_item(f"{self._folder_id}:/{metadata_filename}:")
_LOGGER.debug("Deleted backup %s", backup_filename)
self._cache_expiration = time()
@handle_backup_errors
async def async_list_backups(self, **kwargs: Any) -> list[AgentBackup]:
"""List backups."""
return list((await self._list_cached_metadata_files()).values())
@handle_backup_errors
async def async_get_backup(self, backup_id: str, **kwargs: Any) -> AgentBackup:
"""Return a backup."""
return await self._find_backup_by_id(backup_id)
async def _list_cached_metadata_files(self) -> dict[str, AgentBackup]:
"""List metadata files with a cache."""
if time() <= self._cache_expiration:
return self._cache_backup_metadata
async def _download_metadata(item_id: str) -> AgentBackup | None:
"""Download metadata file."""
try:
metadata_stream = await self._client.download_drive_item(item_id)
except OneDriveException as err:
_LOGGER.warning("Error downloading metadata for %s: %s", item_id, err)
return None
return AgentBackup.from_dict(
json_loads_object(await metadata_stream.read())
)
items = await self._client.list_drive_items(self._folder_id)
# Build a set of backup filenames to check for orphaned metadata
backup_filenames = {
item.name for item in items if item.name and item.name.endswith(".tar")
}
metadata_files: dict[str, AgentBackup] = {}
for item in items:
if item.name and item.name.endswith(".metadata.json"):
# Check if corresponding backup file exists
backup_filename = item.name.replace(".metadata.json", ".tar")
if backup_filename not in backup_filenames:
_LOGGER.warning(
"Backup file %s not found for metadata %s",
backup_filename,
item.name,
)
continue
if metadata := await _download_metadata(item.id):
metadata_files[metadata.backup_id] = metadata
self._cache_backup_metadata = metadata_files
self._cache_expiration = time() + CACHE_TTL
return self._cache_backup_metadata
async def _find_backup_by_id(self, backup_id: str) -> AgentBackup:
"""Find a backup by its backup ID on remote."""
metadata_files = await self._list_cached_metadata_files()
if backup := metadata_files.get(backup_id):
return backup
raise BackupNotFound(f"Backup {backup_id} not found")

149
homeassistant/components/onedrive_for_business/config_flow.py Normal file
View File

@@ -0,0 +1,149 @@
"""Config flow for OneDrive for Business."""
from __future__ import annotations
import logging
from typing import Any, cast
from onedrive_personal_sdk.clients.client import OneDriveClient
from onedrive_personal_sdk.exceptions import OneDriveException
from onedrive_personal_sdk.models.items import AppRoot
import voluptuous as vol
from homeassistant.config_entries import ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_TOKEN
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.config_entry_oauth2_flow import AbstractOAuth2FlowHandler
from .application_credentials import tenant_id_context
from .const import (
CONF_FOLDER_ID,
CONF_FOLDER_PATH,
CONF_TENANT_ID,
DOMAIN,
OAUTH_SCOPES,
)
FOLDER_NAME_SCHEMA = vol.Schema({vol.Required(CONF_FOLDER_PATH): str})
class OneDriveForBusinessConfigFlow(AbstractOAuth2FlowHandler, domain=DOMAIN):
"""Config flow to handle OneDrive OAuth2 authentication."""
DOMAIN = DOMAIN
client: OneDriveClient
approot: AppRoot
@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(OAUTH_SCOPES)}
def __init__(self) -> None:
"""Initialize the OneDrive config flow."""
super().__init__()
self._data: dict[str, Any] = {}
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
return await self.async_step_pick_tenant()
async def async_step_pick_tenant(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Select the tenant id."""
if user_input is not None:
self._data[CONF_TENANT_ID] = user_input[CONF_TENANT_ID]
# Continue with OAuth flow using tenant context
with tenant_id_context(user_input[CONF_TENANT_ID]):
return await self.async_step_pick_implementation()
return self.async_show_form(
step_id="pick_tenant",
data_schema=vol.Schema(
{
vol.Required(CONF_TENANT_ID): str,
}
),
description_placeholders={
"entra_url": "https://entra.microsoft.com/",
"redirect_url": "https://my.home-assistant.io/redirect/oauth",
},
)
async def async_step_pick_implementation(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the pick implementation step with tenant context."""
with tenant_id_context(self._data[CONF_TENANT_ID]):
return await super().async_step_pick_implementation(user_input)
async def async_oauth_create_entry(self, data: dict[str, Any]) -> ConfigFlowResult:
"""Handle the initial step."""
async def get_access_token() -> str:
return cast(str, data[CONF_TOKEN][CONF_ACCESS_TOKEN])
self.client = OneDriveClient(
get_access_token, async_get_clientsession(self.hass)
)
try:
self.approot = await self.client.get_approot()
drive = await self.client.get_drive()
except OneDriveException:
self.logger.exception("Failed to connect to OneDrive")
return self.async_abort(reason="connection_error")
except Exception:
self.logger.exception("Unknown error")
return self.async_abort(reason="unknown")
await self.async_set_unique_id(drive.id)
self._abort_if_unique_id_configured()
self._data.update(data)
return await self.async_step_select_folder()
async def async_step_select_folder(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Step to ask for the folder name."""
errors: dict[str, str] = {}
if user_input is not None:
try:
path = str(user_input[CONF_FOLDER_PATH]).lstrip("/")
folder = await self.client.create_folder("root", path)
except OneDriveException:
self.logger.debug("Failed to create folder", exc_info=True)
errors["base"] = "folder_creation_error"
if not errors:
title = (
f"{self.approot.created_by.user.display_name}'s OneDrive"
if self.approot.created_by.user
and self.approot.created_by.user.display_name
else "OneDrive"
)
return self.async_create_entry(
title=title,
data={
**self._data,
CONF_FOLDER_ID: folder.id,
CONF_FOLDER_PATH: user_input[CONF_FOLDER_PATH],
},
)
return self.async_show_form(
step_id="select_folder",
data_schema=FOLDER_NAME_SCHEMA,
errors=errors,
)

27
homeassistant/components/onedrive_for_business/const.py Normal file
View File

@@ -0,0 +1,27 @@
"""Constants for the OneDrive for Business integration."""
from collections.abc import Callable
from typing import Final
from homeassistant.util.hass_dict import HassKey
DOMAIN: Final = "onedrive_for_business"
CONF_FOLDER_PATH: Final = "folder_path"
CONF_FOLDER_ID: Final = "folder_id"
CONF_TENANT_ID: Final = "tenant_id"
OAUTH2_AUTHORIZE: Final = (
"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize"
)
OAUTH2_TOKEN: Final = "https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
OAUTH_SCOPES: Final = [
"Files.ReadWrite.All",
"offline_access",
"openid",
]
DATA_BACKUP_AGENT_LISTENERS: HassKey[list[Callable[[], None]]] = HassKey(
f"{DOMAIN}.backup_agent_listeners"
)

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

@@ -0,0 +1,14 @@
{
"domain": "onedrive_for_business",
"name": "OneDrive for Business",
"after_dependencies": ["cloud"],
"codeowners": ["@zweckj"],
"config_flow": true,
"dependencies": ["application_credentials"],
"documentation": "https://www.home-assistant.io/integrations/onedrive_for_business",
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["onedrive_personal_sdk"],
"quality_scale": "bronze",
"requirements": ["onedrive-personal-sdk==0.1.2"]
}

131
homeassistant/components/onedrive_for_business/quality_scale.yaml Normal file
View File

@@ -0,0 +1,131 @@
rules:
# Bronze
action-setup:
status: exempt
comment: |
This integration does not define any actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
This integration does not define any actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: |
Entities of this integration does not explicitly subscribe to events.
entity-unique-id:
status: exempt
comment: |
This integration does not create entities.
has-entity-name:
status: exempt
comment: |
This integration does not create 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: |
This integration does not have configuration parameters.
docs-installation-parameters: done
entity-unavailable:
status: exempt
comment: |
This integration does not create entities.
integration-owner: done
log-when-unavailable:
status: exempt
comment: |
This integration does not create entities.
parallel-updates:
status: exempt
comment: |
This integration does not create entities.
reauthentication-flow: todo
test-coverage: done
# Gold
devices:
status: exempt
comment: |
This integration does not create entities.
diagnostics: todo
discovery-update-info:
status: exempt
comment: |
This integration is a cloud service and does not support discovery.
discovery:
status: exempt
comment: |
This integration is a cloud service and does not support discovery.
docs-data-update:
status: exempt
comment: |
This integration does not create entities.
docs-examples:
status: exempt
comment: |
This integration does not create entities or services.
docs-known-limitations:
status: exempt
comment: |
No known limitations.
docs-supported-devices:
status: exempt
comment: |
This integration is a cloud service.
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 create entities.
entity-device-class:
status: exempt
comment: |
This integration does not create entities.
entity-disabled-by-default:
status: exempt
comment: |
This integration does not create entities.
entity-translations:
status: exempt
comment: |
This integration does not create entities.
exception-translations: done
icon-translations:
status: exempt
comment: |
This integration does not create entities.
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: |
No repairs yet.
stale-devices:
status: exempt
comment: |
This integration connects to a single service.
# Platinum
async-dependency: done
inject-websession: done
strict-typing: done

63
homeassistant/components/onedrive_for_business/strings.json Normal file
View File

@@ -0,0 +1,63 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"connection_error": "[%key:component::onedrive::config::abort::connection_error%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]",
"oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"user_rejected_authorize": "[%key:common::config_flow::abort::oauth2_user_rejected_authorize%]"
},
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
},
"error": {
"folder_creation_error": "[%key:component::onedrive::config::error::folder_creation_error%]"
},
"step": {
"pick_implementation": {
"data": {
"implementation": "[%key:common::config_flow::data::implementation%]"
},
"data_description": {
"implementation": "[%key:common::config_flow::description::implementation%]"
},
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
},
"pick_tenant": {
"data": {
"tenant_id": "Tenant ID"
},
"data_description": {
"tenant_id": "The tenant ID of your OneDrive for Business account's tenant."
},
"description": "To set up Onedrive for Business you need to create an app registration in the [Microsoft Entra admin center]({entra_url}) and set the redirect URI to `{redirect_url}`. In this step enter the tenant ID of the tenant where you created the app registration. In the next step you will be asked to provide the client ID and client secret of the app registration.",
"title": "Select tenant"
},
"select_folder": {
"data": {
"folder_path": "Folder path"
},
"data_description": {
"folder_path": "Path to the backup folder"
},
"description": "Path of folder which will be used to create a folder to store backups. If the folder does not exist, it will be created inside your OneDrive root directory.",
"title": "Define backup location"
}
}
},
"exceptions": {
"authentication_failed": {
"message": "[%key:component::onedrive::exceptions::authentication_failed::message%]"
},
"failed_to_get_folder": {
"message": "[%key:component::onedrive::exceptions::failed_to_get_folder::message%]"
}
}
}

1
homeassistant/generated/application_credentials.py generated
View File

@@ -33,6 +33,7 @@ APPLICATION_CREDENTIALS = [
"netatmo",
"ondilo_ico",
"onedrive",
"onedrive_for_business",
"point",
"senz",
"smartthings",

1
homeassistant/generated/config_flows.py generated
View File

@@ -491,6 +491,7 @@ FLOWS = {
"omnilogic",
"ondilo_ico",
"onedrive",
"onedrive_for_business",
"onewire",
"onkyo",
"onvif",

6
homeassistant/generated/integrations.json
View File

@@ -4133,6 +4133,12 @@
"iot_class": "cloud_polling",
"name": "OneDrive"
},
"onedrive_for_business": {
"integration_type": "service",
"config_flow": true,
"iot_class": "cloud_polling",
"name": "OneDrive for Business"
},
"xbox": {
"integration_type": "hub",
"config_flow": true,

10
mypy.ini generated
View File

@@ -3606,6 +3606,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.onedrive_for_business.*]
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.onewire.*]
check_untyped_defs = true
disallow_incomplete_defs = true

1
requirements_all.txt generated
View File

@@ -1666,6 +1666,7 @@ omnilogic==0.4.5
ondilo==0.5.0
# homeassistant.components.onedrive
# homeassistant.components.onedrive_for_business
onedrive-personal-sdk==0.1.2
# homeassistant.components.onvif

1
requirements_test_all.txt generated
View File

@@ -1452,6 +1452,7 @@ omnilogic==0.4.5
ondilo==0.5.0
# homeassistant.components.onedrive
# homeassistant.components.onedrive_for_business
onedrive-personal-sdk==0.1.2
# homeassistant.components.onvif

14
tests/components/onedrive_for_business/__init__.py Normal file
View File

@@ -0,0 +1,14 @@
"""Tests for the OneDrive for Business integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Set up the OneDrive integration for testing."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()

245
tests/components/onedrive_for_business/conftest.py Normal file
View File

@@ -0,0 +1,245 @@
"""Fixtures for OneDrive tests."""
from collections.abc import AsyncIterator, Generator
from json import dumps
import time
from unittest.mock import AsyncMock, MagicMock, patch
from onedrive_personal_sdk.const import DriveState, DriveType
from onedrive_personal_sdk.models.items import (
AppRoot,
Drive,
DriveQuota,
File,
Folder,
Hashes,
IdentitySet,
ItemParentReference,
User,
)
import pytest
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.onedrive_for_business.const import (
CONF_FOLDER_ID,
CONF_FOLDER_PATH,
CONF_TENANT_ID,
DOMAIN,
OAUTH_SCOPES,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .const import BACKUP_METADATA, CLIENT_ID, CLIENT_SECRET, IDENTITY_SET
from tests.common import MockConfigEntry
@pytest.fixture(name="scopes")
def mock_scopes() -> list[str]:
"""Fixture to set the scopes present in the OAuth token."""
return OAUTH_SCOPES
@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(name="expires_at")
def mock_expires_at() -> int:
"""Fixture to set the oauth token expiration time."""
return time.time() + 3600
@pytest.fixture
def mock_config_entry(expires_at: int, scopes: list[str]) -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="John Doe's OneDrive",
domain=DOMAIN,
data={
"auth_implementation": DOMAIN,
"token": {
"access_token": "mock-access-token",
"refresh_token": "mock-refresh-token",
"expires_at": expires_at,
"scope": " ".join(scopes),
},
CONF_FOLDER_PATH: "backups/home_assistant",
CONF_FOLDER_ID: "my_folder_id",
CONF_TENANT_ID: "mock-tenant-id",
},
unique_id="mock_drive_id",
)
@pytest.fixture
def mock_onedrive_client_init() -> Generator[MagicMock]:
"""Return a mocked OneDrive client."""
with (
patch(
"homeassistant.components.onedrive_for_business.config_flow.OneDriveClient",
autospec=True,
) as onedrive_client,
patch(
"homeassistant.components.onedrive_for_business.OneDriveClient",
new=onedrive_client,
),
):
yield onedrive_client
@pytest.fixture
def mock_approot() -> AppRoot:
"""Return a mocked approot."""
return AppRoot(
id="id",
child_count=0,
size=0,
name="name",
parent_reference=ItemParentReference(
drive_id="mock_drive_id", id="id", path="path"
),
created_by=IdentitySet(
user=User(
display_name="John Doe",
id="id",
email="john@doe.com",
)
),
)
@pytest.fixture
def mock_drive() -> Drive:
"""Return a mocked drive."""
return Drive(
id="mock_drive_id",
name="My Drive",
drive_type=DriveType.PERSONAL,
owner=IDENTITY_SET,
quota=DriveQuota(
deleted=5,
remaining=805306368,
state=DriveState.NEARING,
total=5368709120,
used=4250000000,
),
)
@pytest.fixture
def mock_folder() -> Folder:
"""Return a mocked backup folder."""
return Folder(
id="my_folder_id",
name="name",
size=0,
child_count=0,
description="9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0",
parent_reference=ItemParentReference(
drive_id="mock_drive_id", id="id", path="path"
),
created_by=IdentitySet(
user=User(
display_name="John Doe",
id="id",
email="john@doe.com",
),
),
)
@pytest.fixture
def mock_backup_file() -> File:
"""Return a mocked backup file."""
return File(
id="id",
name="23e64aec.tar",
size=34519040,
parent_reference=ItemParentReference(
drive_id="mock_drive_id", id="id", path="path"
),
hashes=Hashes(
quick_xor_hash="hash",
),
mime_type="application/x-tar",
created_by=IDENTITY_SET,
)
@pytest.fixture
def mock_metadata_file() -> File:
"""Return a mocked metadata file."""
return File(
id="id",
name="23e64aec.metadata.json",
size=34519040,
parent_reference=ItemParentReference(
drive_id="mock_drive_id", id="id", path="path"
),
hashes=Hashes(
quick_xor_hash="hash",
),
mime_type="application/json",
created_by=IDENTITY_SET,
)
@pytest.fixture(autouse=True)
def mock_onedrive_client(
mock_onedrive_client_init: MagicMock,
mock_approot: AppRoot,
mock_drive: Drive,
mock_folder: Folder,
mock_backup_file: File,
mock_metadata_file: File,
) -> Generator[MagicMock]:
"""Return a mocked GraphServiceClient."""
client = mock_onedrive_client_init.return_value
client.get_approot.return_value = mock_approot
client.create_folder.return_value = mock_folder
client.list_drive_items.return_value = [mock_backup_file, mock_metadata_file]
client.get_drive_item.return_value = mock_folder
client.upload_file.return_value = mock_metadata_file
class MockStreamReader:
async def iter_chunked(self, chunk_size: int) -> AsyncIterator[bytes]:
yield b"backup data"
async def read(self) -> bytes:
return dumps(BACKUP_METADATA).encode()
client.download_drive_item.return_value = MockStreamReader()
client.get_drive.return_value = mock_drive
return client
@pytest.fixture
def mock_large_file_upload_client(mock_backup_file: File) -> Generator[AsyncMock]:
"""Return a mocked LargeFileUploadClient upload."""
with patch(
"homeassistant.components.onedrive_for_business.backup.LargeFileUploadClient.upload"
) as mock_upload:
mock_upload.return_value = mock_backup_file
yield mock_upload
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.onedrive_for_business.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry

32
tests/components/onedrive_for_business/const.py Normal file
View File

@@ -0,0 +1,32 @@
"""Consts for OneDrive tests."""
from onedrive_personal_sdk.models.items import IdentitySet, User
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
TENANT_ID = "test_tenant_id"
BACKUP_METADATA = {
"addons": [],
"backup_id": "23e64aec",
"date": "2024-11-22T11:48:48.727189+01:00",
"database_included": True,
"extra_metadata": {},
"folders": [],
"homeassistant_included": True,
"homeassistant_version": "2024.12.0.dev0",
"name": "Core 2024.12.0.dev0",
"protected": False,
"size": 34519040,
}
INSTANCE_ID = "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0"
IDENTITY_SET = IdentitySet(
user=User(
display_name="John Doe",
id="id",
email="john@doe.com",
)
)

436
tests/components/onedrive_for_business/test_backup.py Normal file
View File

@@ -0,0 +1,436 @@
"""Test the backups for OneDrive for Business."""
from __future__ import annotations
from collections.abc import AsyncGenerator
from io import StringIO
from unittest.mock import Mock, patch
from onedrive_personal_sdk.exceptions import HashMismatchError, OneDriveException
from onedrive_personal_sdk.models.items import File
import pytest
from homeassistant.components.backup import DOMAIN as BACKUP_DOMAIN, AgentBackup
from homeassistant.components.onedrive_for_business.backup import (
async_register_backup_agents_listener,
)
from homeassistant.components.onedrive_for_business.const import (
DATA_BACKUP_AGENT_LISTENERS,
DOMAIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from . import setup_integration
from .const import BACKUP_METADATA
from tests.common import AsyncMock, MockConfigEntry
from tests.typing import ClientSessionGenerator, MagicMock, WebSocketGenerator
@pytest.fixture(autouse=True)
async def setup_backup_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> AsyncGenerator[None]:
"""Set up onedrive integration."""
with (
patch("homeassistant.components.backup.is_hassio", return_value=False),
patch("homeassistant.components.backup.store.STORE_DELAY_SAVE", 0),
):
assert await async_setup_component(hass, BACKUP_DOMAIN, {BACKUP_DOMAIN: {}})
await setup_integration(hass, mock_config_entry)
await hass.async_block_till_done()
yield
async def test_agents_info(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_config_entry: MockConfigEntry,
) -> 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": f"{DOMAIN}.{mock_config_entry.unique_id}",
"name": mock_config_entry.title,
},
],
}
async def test_agents_list_backups(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test agent list backups."""
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"] == [
{
"addons": [],
"agents": {
"onedrive_for_business.mock_drive_id": {
"protected": False,
"size": 34519040,
}
},
"backup_id": "23e64aec",
"database_included": True,
"date": "2024-11-22T11:48:48.727189+01:00",
"extra_metadata": {},
"failed_addons": [],
"failed_agent_ids": [],
"failed_folders": [],
"folders": [],
"homeassistant_included": True,
"homeassistant_version": "2024.12.0.dev0",
"name": "Core 2024.12.0.dev0",
"with_automatic_settings": None,
}
]
async def test_agents_list_backups_with_download_failure(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_onedrive_client: MagicMock,
) -> None:
"""Test agent list backups still works if one of the items fails to download."""
mock_onedrive_client.download_drive_item.side_effect = OneDriveException("test")
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"] == []
async def test_agents_get_backup(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test agent get backup."""
backup_id = BACKUP_METADATA["backup_id"]
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"] == {
"addons": [],
"agents": {
f"{DOMAIN}.{mock_config_entry.unique_id}": {
"protected": False,
"size": 34519040,
}
},
"backup_id": "23e64aec",
"database_included": True,
"date": "2024-11-22T11:48:48.727189+01:00",
"extra_metadata": {},
"failed_addons": [],
"failed_agent_ids": [],
"failed_folders": [],
"folders": [],
"homeassistant_included": True,
"homeassistant_version": "2024.12.0.dev0",
"name": "Core 2024.12.0.dev0",
"with_automatic_settings": None,
}
async def test_agents_get_backup_missing_file(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_config_entry: MockConfigEntry,
mock_onedrive_client: MagicMock,
mock_metadata_file: File,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test what happens when only metadata exists."""
mock_onedrive_client.list_drive_items.return_value = [mock_metadata_file]
backup_id = BACKUP_METADATA["backup_id"]
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"] is None
assert (
"Backup file 23e64aec.tar not found for metadata 23e64aec.metadata.json"
in caplog.text
)
async def test_agents_delete(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_onedrive_client: MagicMock,
) -> None:
"""Test agent delete backup."""
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "backup/delete",
"backup_id": BACKUP_METADATA["backup_id"],
}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {"agent_errors": {}}
assert mock_onedrive_client.delete_drive_item.call_count == 2
async def test_agents_upload(
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
mock_onedrive_client: MagicMock,
mock_large_file_upload_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test agent upload backup."""
client = await hass_client()
test_backup = AgentBackup.from_dict(BACKUP_METADATA)
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
) as fetch_backup,
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=test_backup,
),
patch("pathlib.Path.open") as mocked_open,
):
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
fetch_backup.return_value = test_backup
resp = await client.post(
f"/api/backup/upload?agent_id={DOMAIN}.{mock_config_entry.unique_id}",
data={"file": StringIO("test")},
)
assert resp.status == 201
assert f"Uploading backup {test_backup.backup_id}" in caplog.text
mock_large_file_upload_client.assert_called_once()
async def test_agents_upload_corrupt_upload(
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
mock_onedrive_client: MagicMock,
mock_large_file_upload_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test hash validation fails."""
mock_large_file_upload_client.side_effect = HashMismatchError("test")
client = await hass_client()
test_backup = AgentBackup.from_dict(BACKUP_METADATA)
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
) as fetch_backup,
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=test_backup,
),
patch("pathlib.Path.open") as mocked_open,
):
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
fetch_backup.return_value = test_backup
resp = await client.post(
f"/api/backup/upload?agent_id={DOMAIN}.{mock_config_entry.unique_id}",
data={"file": StringIO("test")},
)
assert resp.status == 201
assert f"Uploading backup {test_backup.backup_id}" in caplog.text
mock_large_file_upload_client.assert_called_once()
assert mock_onedrive_client.update_drive_item.call_count == 0
assert "Hash validation failed, backup file might be corrupt" in caplog.text
async def test_agents_upload_metadata_upload_failed(
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
mock_onedrive_client: MagicMock,
mock_large_file_upload_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test metadata upload fails."""
client = await hass_client()
test_backup = AgentBackup.from_dict(BACKUP_METADATA)
mock_onedrive_client.upload_file.side_effect = OneDriveException("test")
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
) as fetch_backup,
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=test_backup,
),
patch("pathlib.Path.open") as mocked_open,
):
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
fetch_backup.return_value = test_backup
resp = await client.post(
f"/api/backup/upload?agent_id={DOMAIN}.{mock_config_entry.unique_id}",
data={"file": StringIO("test")},
)
assert resp.status == 201
assert f"Uploading backup {test_backup.backup_id}" in caplog.text
mock_large_file_upload_client.assert_called_once()
mock_onedrive_client.delete_drive_item.assert_called_once()
assert mock_onedrive_client.update_drive_item.call_count == 0
async def test_agents_download(
hass_client: ClientSessionGenerator,
mock_onedrive_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test agent download backup."""
client = await hass_client()
backup_id = BACKUP_METADATA["backup_id"]
resp = await client.get(
f"/api/backup/download/{backup_id}?agent_id={DOMAIN}.{mock_config_entry.unique_id}"
)
assert resp.status == 200
assert await resp.content.read() == b"backup data"
async def test_error_on_agents_download(
hass_client: ClientSessionGenerator,
mock_onedrive_client: MagicMock,
mock_config_entry: MockConfigEntry,
mock_backup_file: File,
mock_metadata_file: File,
) -> None:
"""Test we get not found on an not existing backup on download."""
client = await hass_client()
backup_id = BACKUP_METADATA["backup_id"]
mock_onedrive_client.list_drive_items.side_effect = [
[mock_backup_file, mock_metadata_file],
[],
]
with patch("homeassistant.components.onedrive_for_business.backup.CACHE_TTL", -1):
resp = await client.get(
f"/api/backup/download/{backup_id}?agent_id={DOMAIN}.{mock_config_entry.unique_id}"
)
assert resp.status == 404
@pytest.mark.parametrize(
("side_effect", "error"),
[
(
OneDriveException(),
"Backup operation failed",
),
(TimeoutError(), "Backup operation timed out"),
],
)
async def test_delete_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_onedrive_client: MagicMock,
mock_config_entry: MockConfigEntry,
side_effect: Exception,
error: str,
) -> None:
"""Test error during delete."""
mock_onedrive_client.delete_drive_item.side_effect = AsyncMock(
side_effect=side_effect
)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "backup/delete",
"backup_id": BACKUP_METADATA["backup_id"],
}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"agent_errors": {f"{DOMAIN}.{mock_config_entry.unique_id}": error}
}
async def test_agents_delete_not_found_does_not_throw(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_onedrive_client: MagicMock,
) -> None:
"""Test agent delete backup."""
mock_onedrive_client.list_drive_items.return_value = []
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "backup/delete",
"backup_id": BACKUP_METADATA["backup_id"],
}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {"agent_errors": {}}
async def test_agents_backup_not_found(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_onedrive_client: MagicMock,
) -> None:
"""Test backup not found."""
mock_onedrive_client.list_drive_items.return_value = []
backup_id = BACKUP_METADATA["backup_id"]
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"]["backup"] is None
async def test_listeners_get_cleaned_up(hass: HomeAssistant) -> None:
"""Test listener gets cleaned up."""
listener = MagicMock()
remove_listener = async_register_backup_agents_listener(hass, listener=listener)
# make sure it's the last listener
hass.data[DATA_BACKUP_AGENT_LISTENERS] = [listener]
remove_listener()
assert hass.data.get(DATA_BACKUP_AGENT_LISTENERS) is None

264
tests/components/onedrive_for_business/test_config_flow.py Normal file
View File

@@ -0,0 +1,264 @@
"""Test the OneDrive config flow."""
from http import HTTPStatus
from unittest.mock import AsyncMock, MagicMock
from onedrive_personal_sdk.exceptions import OneDriveException
import pytest
from homeassistant import config_entries
from homeassistant.components.onedrive_for_business.const import (
CONF_FOLDER_ID,
CONF_FOLDER_PATH,
CONF_TENANT_ID,
DOMAIN,
OAUTH2_AUTHORIZE,
OAUTH2_TOKEN,
)
from homeassistant.config_entries import ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from . import setup_integration
from .const import CLIENT_ID, TENANT_ID
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
async def _do_get_token(
hass: HomeAssistant,
result: ConfigFlowResult,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
) -> None:
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
scope = "Files.ReadWrite.All+offline_access+openid"
authorize_url = OAUTH2_AUTHORIZE.format(tenant_id=TENANT_ID)
token_url = OAUTH2_TOKEN.format(tenant_id=TENANT_ID)
assert result["url"] == (
f"{authorize_url}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope={scope}"
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == HTTPStatus.OK
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
token_url,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
@pytest.mark.usefixtures("current_request_with_host")
async def test_full_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_setup_entry: AsyncMock,
mock_onedrive_client_init: MagicMock,
) -> None:
"""Check full flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pick_tenant"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_TENANT_ID: TENANT_ID}
)
await _do_get_token(hass, result, hass_client_no_auth, aioclient_mock)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
# Ensure the token callback is set up correctly
token_callback = mock_onedrive_client_init.call_args[0][0]
assert await token_callback() == "mock-access-token"
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_FOLDER_PATH: "myFolder"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup_entry.mock_calls) == 1
assert result["title"] == "John Doe's OneDrive"
assert result["result"].unique_id == "mock_drive_id"
assert result["data"][CONF_TOKEN][CONF_ACCESS_TOKEN] == "mock-access-token"
assert result["data"][CONF_TOKEN]["refresh_token"] == "mock-refresh-token"
assert result["data"][CONF_FOLDER_PATH] == "myFolder"
assert result["data"][CONF_FOLDER_ID] == "my_folder_id"
@pytest.mark.usefixtures("current_request_with_host")
async def test_full_flow_with_owner_not_found(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_setup_entry: AsyncMock,
mock_onedrive_client: MagicMock,
mock_approot: MagicMock,
) -> None:
"""Ensure we get a default title if the drive's owner can't be read."""
mock_approot.created_by.user = None
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pick_tenant"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_TENANT_ID: TENANT_ID}
)
await _do_get_token(hass, result, hass_client_no_auth, aioclient_mock)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_FOLDER_PATH: "myFolder"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup_entry.mock_calls) == 1
assert result["title"] == "OneDrive"
assert result["result"].unique_id == "mock_drive_id"
assert result["data"][CONF_TOKEN][CONF_ACCESS_TOKEN] == "mock-access-token"
assert result["data"][CONF_TOKEN]["refresh_token"] == "mock-refresh-token"
assert result["data"][CONF_FOLDER_PATH] == "myFolder"
assert result["data"][CONF_FOLDER_ID] == "my_folder_id"
mock_onedrive_client.reset_mock()
@pytest.mark.usefixtures("current_request_with_host")
async def test_error_during_folder_creation(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_setup_entry: AsyncMock,
mock_onedrive_client: MagicMock,
) -> None:
"""Ensure we can create the backup folder."""
mock_onedrive_client.create_folder.side_effect = OneDriveException()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pick_tenant"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_TENANT_ID: TENANT_ID}
)
await _do_get_token(hass, result, hass_client_no_auth, aioclient_mock)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_FOLDER_PATH: "myFolder"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "folder_creation_error"}
mock_onedrive_client.create_folder.side_effect = None
# clear error and try again
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_FOLDER_PATH: "myFolder"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "John Doe's OneDrive"
assert result["result"].unique_id == "mock_drive_id"
assert result["data"][CONF_TOKEN][CONF_ACCESS_TOKEN] == "mock-access-token"
assert result["data"][CONF_TOKEN]["refresh_token"] == "mock-refresh-token"
assert result["data"][CONF_FOLDER_PATH] == "myFolder"
assert result["data"][CONF_FOLDER_ID] == "my_folder_id"
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.parametrize(
("exception", "error"),
[
(Exception, "unknown"),
(OneDriveException, "connection_error"),
],
)
async def test_flow_errors(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_onedrive_client: MagicMock,
exception: Exception,
error: str,
) -> None:
"""Test errors during flow."""
mock_onedrive_client.get_approot.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pick_tenant"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_TENANT_ID: TENANT_ID}
)
await _do_get_token(hass, result, hass_client_no_auth, aioclient_mock)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == error
@pytest.mark.usefixtures("current_request_with_host")
async def test_already_configured(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_setup_entry: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test already configured account."""
await setup_integration(hass, mock_config_entry)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pick_tenant"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_TENANT_ID: TENANT_ID}
)
await _do_get_token(hass, result, hass_client_no_auth, aioclient_mock)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

105
tests/components/onedrive_for_business/test_init.py Normal file
View File

@@ -0,0 +1,105 @@
"""Test the OneDrive setup."""
from copy import copy
from unittest.mock import MagicMock
from onedrive_personal_sdk.exceptions import (
AuthenticationError,
NotFoundError,
OneDriveException,
)
from onedrive_personal_sdk.models.items import AppRoot, Folder
import pytest
from homeassistant.components.onedrive_for_business.const import (
CONF_FOLDER_ID,
CONF_FOLDER_PATH,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
async def test_load_unload_config_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_onedrive_client_init: MagicMock,
mock_onedrive_client: MagicMock,
) -> None:
"""Test loading and unloading the integration."""
await setup_integration(hass, mock_config_entry)
# Ensure the token callback is set up correctly
token_callback = mock_onedrive_client_init.call_args[0][0]
assert await token_callback() == "mock-access-token"
# make sure metadata migration is not called
assert mock_onedrive_client.upload_file.call_count == 0
assert mock_onedrive_client.update_drive_item.call_count == 0
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.parametrize(
("side_effect", "state"),
[
(AuthenticationError(403, "Auth failed"), ConfigEntryState.SETUP_ERROR),
(OneDriveException(), ConfigEntryState.SETUP_RETRY),
],
)
async def test_approot_errors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_onedrive_client: MagicMock,
side_effect: Exception,
state: ConfigEntryState,
) -> None:
"""Test errors during approot retrieval."""
mock_onedrive_client.get_drive_item.side_effect = side_effect
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is state
async def test_get_integration_folder_creation(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_onedrive_client: MagicMock,
mock_approot: AppRoot,
mock_folder: Folder,
) -> None:
"""Test faulty integration folder creation."""
folder_name = copy(mock_config_entry.data[CONF_FOLDER_PATH])
mock_onedrive_client.get_drive_item.side_effect = NotFoundError(404, "Not found")
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_onedrive_client.create_folder.assert_called_once_with(
parent_id="root",
name=folder_name,
)
# ensure the folder id and name are updated
assert mock_config_entry.data[CONF_FOLDER_ID] == mock_folder.id
assert mock_config_entry.data[CONF_FOLDER_PATH] == folder_name
async def test_get_integration_folder_creation_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_onedrive_client: MagicMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test faulty integration folder creation error."""
mock_onedrive_client.get_drive_item.side_effect = NotFoundError(404, "Not found")
mock_onedrive_client.create_folder.side_effect = OneDriveException()
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
assert "Failed to get backups/home_assistant folder" in caplog.text