mirror of
https://github.com/home-assistant/core.git
synced 2026-03-20 01:34:53 +01:00
Compare commits
42 Commits
climate_ad
...
PIRUnoccup
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9cadf32e36 | ||
|
|
d44387e36b | ||
|
|
538b817bf1 | ||
|
|
7efa2d3cac | ||
|
|
3f872fd196 | ||
|
|
1824ef12bb | ||
|
|
c706e8a5b8 | ||
|
|
5bd9742eb3 | ||
|
|
26f3eb5f6d | ||
|
|
7a34d4f881 | ||
|
|
e0a37a5eeb | ||
|
|
ec3d1fd72c | ||
|
|
4edea21cb7 | ||
|
|
7f065c1942 | ||
|
|
46ce07a9a1 | ||
|
|
5807db2c60 | ||
|
|
85732543b2 | ||
|
|
054c61d73f | ||
|
|
be2c20c624 | ||
|
|
706127c9ea | ||
|
|
b163829970 | ||
|
|
7a93eb779c | ||
|
|
7d673cd9c4 | ||
|
|
44bc11580d | ||
|
|
c23795fe14 | ||
|
|
bf6f9a011b | ||
|
|
1cdbe596fe | ||
|
|
a9d52bfbe7 | ||
|
|
6eed1f9961 | ||
|
|
149607ab17 | ||
|
|
279b5be357 | ||
|
|
82b93e788b | ||
|
|
555813f84f | ||
|
|
ecf1b4e591 | ||
|
|
e17a9f12a1 | ||
|
|
e8f05f5291 | ||
|
|
a5a76e9268 | ||
|
|
edc3fb47b2 | ||
|
|
f1e514a70a | ||
|
|
5632baca5b | ||
|
|
78f9bad706 | ||
|
|
3fdaaecd0f |
@@ -173,6 +173,7 @@ homeassistant.components.dnsip.*
|
||||
homeassistant.components.doorbird.*
|
||||
homeassistant.components.dormakaba_dkey.*
|
||||
homeassistant.components.downloader.*
|
||||
homeassistant.components.dropbox.*
|
||||
homeassistant.components.droplet.*
|
||||
homeassistant.components.dsmr.*
|
||||
homeassistant.components.duckdns.*
|
||||
|
||||
2
CODEOWNERS
generated
2
CODEOWNERS
generated
@@ -397,6 +397,8 @@ build.json @home-assistant/supervisor
|
||||
/tests/components/dremel_3d_printer/ @tkdrob
|
||||
/homeassistant/components/drop_connect/ @ChandlerSystems @pfrazer
|
||||
/tests/components/drop_connect/ @ChandlerSystems @pfrazer
|
||||
/homeassistant/components/dropbox/ @bdr99
|
||||
/tests/components/dropbox/ @bdr99
|
||||
/homeassistant/components/droplet/ @sarahseidman
|
||||
/tests/components/droplet/ @sarahseidman
|
||||
/homeassistant/components/dsmr/ @Robbie1221
|
||||
|
||||
64
homeassistant/components/dropbox/__init__.py
Normal file
64
homeassistant/components/dropbox/__init__.py
Normal file
@@ -0,0 +1,64 @@
|
||||
"""The Dropbox integration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from python_dropbox_api import (
|
||||
DropboxAPIClient,
|
||||
DropboxAuthException,
|
||||
DropboxUnknownException,
|
||||
)
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
|
||||
from homeassistant.helpers import aiohttp_client
|
||||
from homeassistant.helpers.config_entry_oauth2_flow import (
|
||||
ImplementationUnavailableError,
|
||||
OAuth2Session,
|
||||
async_get_config_entry_implementation,
|
||||
)
|
||||
|
||||
from .auth import DropboxConfigEntryAuth
|
||||
from .const import DATA_BACKUP_AGENT_LISTENERS, DOMAIN
|
||||
|
||||
type DropboxConfigEntry = ConfigEntry[DropboxAPIClient]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: DropboxConfigEntry) -> bool:
|
||||
"""Set up Dropbox from a config entry."""
|
||||
try:
|
||||
oauth2_implementation = await async_get_config_entry_implementation(hass, entry)
|
||||
except ImplementationUnavailableError as err:
|
||||
raise ConfigEntryNotReady(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="oauth2_implementation_unavailable",
|
||||
) from err
|
||||
oauth2_session = OAuth2Session(hass, entry, oauth2_implementation)
|
||||
|
||||
auth = DropboxConfigEntryAuth(
|
||||
aiohttp_client.async_get_clientsession(hass), oauth2_session
|
||||
)
|
||||
|
||||
client = DropboxAPIClient(auth)
|
||||
|
||||
try:
|
||||
await client.get_account_info()
|
||||
except DropboxAuthException as err:
|
||||
raise ConfigEntryAuthFailed from err
|
||||
except (DropboxUnknownException, TimeoutError) as err:
|
||||
raise ConfigEntryNotReady from err
|
||||
|
||||
entry.runtime_data = client
|
||||
|
||||
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, entry: DropboxConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return True
|
||||
38
homeassistant/components/dropbox/application_credentials.py
Normal file
38
homeassistant/components/dropbox/application_credentials.py
Normal file
@@ -0,0 +1,38 @@
|
||||
"""Application credentials platform for the Dropbox integration."""
|
||||
|
||||
from homeassistant.components.application_credentials import ClientCredential
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.config_entry_oauth2_flow import (
|
||||
AbstractOAuth2Implementation,
|
||||
LocalOAuth2ImplementationWithPkce,
|
||||
)
|
||||
|
||||
from .const import OAUTH2_AUTHORIZE, OAUTH2_SCOPES, OAUTH2_TOKEN
|
||||
|
||||
|
||||
async def async_get_auth_implementation(
|
||||
hass: HomeAssistant, auth_domain: str, credential: ClientCredential
|
||||
) -> AbstractOAuth2Implementation:
|
||||
"""Return custom auth implementation."""
|
||||
return DropboxOAuth2Implementation(
|
||||
hass,
|
||||
auth_domain,
|
||||
credential.client_id,
|
||||
OAUTH2_AUTHORIZE,
|
||||
OAUTH2_TOKEN,
|
||||
credential.client_secret,
|
||||
)
|
||||
|
||||
|
||||
class DropboxOAuth2Implementation(LocalOAuth2ImplementationWithPkce):
|
||||
"""Custom Dropbox OAuth2 implementation to add the necessary authorize url parameters."""
|
||||
|
||||
@property
|
||||
def extra_authorize_data(self) -> dict:
|
||||
"""Extra data that needs to be appended to the authorize url."""
|
||||
data: dict = {
|
||||
"token_access_type": "offline",
|
||||
"scope": " ".join(OAUTH2_SCOPES),
|
||||
}
|
||||
data.update(super().extra_authorize_data)
|
||||
return data
|
||||
44
homeassistant/components/dropbox/auth.py
Normal file
44
homeassistant/components/dropbox/auth.py
Normal file
@@ -0,0 +1,44 @@
|
||||
"""Authentication for Dropbox."""
|
||||
|
||||
from typing import cast
|
||||
|
||||
from aiohttp import ClientSession
|
||||
from python_dropbox_api import Auth
|
||||
|
||||
from homeassistant.helpers.config_entry_oauth2_flow import OAuth2Session
|
||||
|
||||
|
||||
class DropboxConfigEntryAuth(Auth):
|
||||
"""Provide Dropbox authentication tied to an OAuth2 based config entry."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
websession: ClientSession,
|
||||
oauth_session: OAuth2Session,
|
||||
) -> None:
|
||||
"""Initialize DropboxConfigEntryAuth."""
|
||||
super().__init__(websession)
|
||||
self._oauth_session = oauth_session
|
||||
|
||||
async def async_get_access_token(self) -> str:
|
||||
"""Return a valid access token."""
|
||||
await self._oauth_session.async_ensure_token_valid()
|
||||
|
||||
return cast(str, self._oauth_session.token["access_token"])
|
||||
|
||||
|
||||
class DropboxConfigFlowAuth(Auth):
|
||||
"""Provide authentication tied to a fixed token for the config flow."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
websession: ClientSession,
|
||||
token: str,
|
||||
) -> None:
|
||||
"""Initialize DropboxConfigFlowAuth."""
|
||||
super().__init__(websession)
|
||||
self._token = token
|
||||
|
||||
async def async_get_access_token(self) -> str:
|
||||
"""Return the fixed access token."""
|
||||
return self._token
|
||||
230
homeassistant/components/dropbox/backup.py
Normal file
230
homeassistant/components/dropbox/backup.py
Normal file
@@ -0,0 +1,230 @@
|
||||
"""Backup platform for the Dropbox integration."""
|
||||
|
||||
from collections.abc import AsyncIterator, Callable, Coroutine
|
||||
from functools import wraps
|
||||
import json
|
||||
import logging
|
||||
from typing import Any, Concatenate
|
||||
|
||||
from python_dropbox_api import (
|
||||
DropboxAPIClient,
|
||||
DropboxAuthException,
|
||||
DropboxFileOrFolderNotFoundException,
|
||||
DropboxUnknownException,
|
||||
)
|
||||
|
||||
from homeassistant.components.backup import (
|
||||
AgentBackup,
|
||||
BackupAgent,
|
||||
BackupAgentError,
|
||||
BackupNotFound,
|
||||
suggested_filename,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
|
||||
from . import DropboxConfigEntry
|
||||
from .const import DATA_BACKUP_AGENT_LISTENERS, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
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"
|
||||
|
||||
|
||||
async def _async_string_iterator(content: str) -> AsyncIterator[bytes]:
|
||||
"""Yield a string as a single bytes chunk."""
|
||||
yield content.encode()
|
||||
|
||||
|
||||
def handle_backup_errors[_R, **P](
|
||||
func: Callable[Concatenate[DropboxBackupAgent, P], Coroutine[Any, Any, _R]],
|
||||
) -> Callable[Concatenate[DropboxBackupAgent, P], Coroutine[Any, Any, _R]]:
|
||||
"""Handle backup errors."""
|
||||
|
||||
@wraps(func)
|
||||
async def wrapper(
|
||||
self: DropboxBackupAgent, *args: P.args, **kwargs: P.kwargs
|
||||
) -> _R:
|
||||
try:
|
||||
return await func(self, *args, **kwargs)
|
||||
except DropboxFileOrFolderNotFoundException as err:
|
||||
raise BackupNotFound(
|
||||
f"Failed to {func.__name__.removeprefix('async_').replace('_', ' ')}"
|
||||
) from err
|
||||
except DropboxAuthException as err:
|
||||
self._entry.async_start_reauth(self._hass)
|
||||
raise BackupAgentError("Authentication error") from err
|
||||
except DropboxUnknownException as err:
|
||||
_LOGGER.error(
|
||||
"Error during %s: %s",
|
||||
func.__name__,
|
||||
err,
|
||||
)
|
||||
_LOGGER.debug("Full error: %s", err, exc_info=True)
|
||||
raise BackupAgentError(
|
||||
f"Failed to {func.__name__.removeprefix('async_').replace('_', ' ')}"
|
||||
) from err
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
async def async_get_backup_agents(
|
||||
hass: HomeAssistant,
|
||||
**kwargs: Any,
|
||||
) -> list[BackupAgent]:
|
||||
"""Return a list of backup agents."""
|
||||
entries = hass.config_entries.async_loaded_entries(DOMAIN)
|
||||
return [DropboxBackupAgent(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.
|
||||
|
||||
:return: A function to unregister the listener.
|
||||
"""
|
||||
hass.data.setdefault(DATA_BACKUP_AGENT_LISTENERS, []).append(listener)
|
||||
|
||||
@callback
|
||||
def remove_listener() -> None:
|
||||
"""Remove the listener."""
|
||||
hass.data[DATA_BACKUP_AGENT_LISTENERS].remove(listener)
|
||||
if not hass.data[DATA_BACKUP_AGENT_LISTENERS]:
|
||||
del hass.data[DATA_BACKUP_AGENT_LISTENERS]
|
||||
|
||||
return remove_listener
|
||||
|
||||
|
||||
class DropboxBackupAgent(BackupAgent):
|
||||
"""Backup agent for the Dropbox integration."""
|
||||
|
||||
domain = DOMAIN
|
||||
|
||||
def __init__(self, hass: HomeAssistant, entry: DropboxConfigEntry) -> None:
|
||||
"""Initialize the backup agent."""
|
||||
super().__init__()
|
||||
self._hass = hass
|
||||
self._entry = entry
|
||||
self.name = entry.title
|
||||
assert entry.unique_id
|
||||
self.unique_id = entry.unique_id
|
||||
self._api: DropboxAPIClient = entry.runtime_data
|
||||
|
||||
async def _async_get_backups(self) -> list[tuple[AgentBackup, str]]:
|
||||
"""Get backups and their corresponding file names."""
|
||||
files = await self._api.list_folder("")
|
||||
|
||||
tar_files = {f.name for f in files if f.name.endswith(".tar")}
|
||||
metadata_files = [f for f in files if f.name.endswith(".metadata.json")]
|
||||
|
||||
backups: list[tuple[AgentBackup, str]] = []
|
||||
for metadata_file in metadata_files:
|
||||
tar_name = metadata_file.name.removesuffix(".metadata.json") + ".tar"
|
||||
if tar_name not in tar_files:
|
||||
_LOGGER.warning(
|
||||
"Found metadata file '%s' without matching backup file",
|
||||
metadata_file.name,
|
||||
)
|
||||
continue
|
||||
|
||||
metadata_stream = self._api.download_file(f"/{metadata_file.name}")
|
||||
raw = b"".join([chunk async for chunk in metadata_stream])
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
backup = AgentBackup.from_dict(data)
|
||||
except (json.JSONDecodeError, ValueError, TypeError, KeyError) as err:
|
||||
_LOGGER.warning(
|
||||
"Skipping invalid metadata file '%s': %s",
|
||||
metadata_file.name,
|
||||
err,
|
||||
)
|
||||
continue
|
||||
backups.append((backup, tar_name))
|
||||
|
||||
return backups
|
||||
|
||||
@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)
|
||||
backup_path = f"/{backup_filename}"
|
||||
metadata_path = f"/{metadata_filename}"
|
||||
|
||||
file_stream = await open_stream()
|
||||
await self._api.upload_file(backup_path, file_stream)
|
||||
|
||||
metadata_stream = _async_string_iterator(json.dumps(backup.as_dict()))
|
||||
|
||||
try:
|
||||
await self._api.upload_file(metadata_path, metadata_stream)
|
||||
except (
|
||||
DropboxAuthException,
|
||||
DropboxUnknownException,
|
||||
):
|
||||
await self._api.delete_file(backup_path)
|
||||
raise
|
||||
|
||||
@handle_backup_errors
|
||||
async def async_list_backups(self, **kwargs: Any) -> list[AgentBackup]:
|
||||
"""List backups."""
|
||||
return [backup for backup, _ in await self._async_get_backups()]
|
||||
|
||||
@handle_backup_errors
|
||||
async def async_download_backup(
|
||||
self,
|
||||
backup_id: str,
|
||||
**kwargs: Any,
|
||||
) -> AsyncIterator[bytes]:
|
||||
"""Download a backup file."""
|
||||
backups = await self._async_get_backups()
|
||||
for backup, filename in backups:
|
||||
if backup.backup_id == backup_id:
|
||||
return self._api.download_file(f"/{filename}")
|
||||
|
||||
raise BackupNotFound(f"Backup {backup_id} not found")
|
||||
|
||||
@handle_backup_errors
|
||||
async def async_get_backup(
|
||||
self,
|
||||
backup_id: str,
|
||||
**kwargs: Any,
|
||||
) -> AgentBackup:
|
||||
"""Return a backup."""
|
||||
backups = await self._async_get_backups()
|
||||
|
||||
for backup, _ in backups:
|
||||
if backup.backup_id == backup_id:
|
||||
return backup
|
||||
|
||||
raise BackupNotFound(f"Backup {backup_id} not found")
|
||||
|
||||
@handle_backup_errors
|
||||
async def async_delete_backup(
|
||||
self,
|
||||
backup_id: str,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
"""Delete a backup file."""
|
||||
backups = await self._async_get_backups()
|
||||
for backup, tar_filename in backups:
|
||||
if backup.backup_id == backup_id:
|
||||
metadata_filename = tar_filename.removesuffix(".tar") + ".metadata.json"
|
||||
await self._api.delete_file(f"/{tar_filename}")
|
||||
await self._api.delete_file(f"/{metadata_filename}")
|
||||
return
|
||||
|
||||
raise BackupNotFound(f"Backup {backup_id} not found")
|
||||
60
homeassistant/components/dropbox/config_flow.py
Normal file
60
homeassistant/components/dropbox/config_flow.py
Normal file
@@ -0,0 +1,60 @@
|
||||
"""Config flow for Dropbox."""
|
||||
|
||||
from collections.abc import Mapping
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from python_dropbox_api import DropboxAPIClient
|
||||
|
||||
from homeassistant.config_entries import SOURCE_REAUTH, 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 .auth import DropboxConfigFlowAuth
|
||||
from .const import DOMAIN
|
||||
|
||||
|
||||
class DropboxConfigFlow(AbstractOAuth2FlowHandler, domain=DOMAIN):
|
||||
"""Config flow to handle Dropbox OAuth2 authentication."""
|
||||
|
||||
DOMAIN = DOMAIN
|
||||
|
||||
@property
|
||||
def logger(self) -> logging.Logger:
|
||||
"""Return logger."""
|
||||
return logging.getLogger(__name__)
|
||||
|
||||
async def async_oauth_create_entry(self, data: dict[str, Any]) -> ConfigFlowResult:
|
||||
"""Create an entry for the flow, or update existing entry."""
|
||||
access_token = data[CONF_TOKEN][CONF_ACCESS_TOKEN]
|
||||
|
||||
auth = DropboxConfigFlowAuth(async_get_clientsession(self.hass), access_token)
|
||||
|
||||
client = DropboxAPIClient(auth)
|
||||
account_info = await client.get_account_info()
|
||||
|
||||
await self.async_set_unique_id(account_info.account_id)
|
||||
if self.source == SOURCE_REAUTH:
|
||||
self._abort_if_unique_id_mismatch(reason="wrong_account")
|
||||
return self.async_update_reload_and_abort(
|
||||
self._get_reauth_entry(), data=data
|
||||
)
|
||||
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
return self.async_create_entry(title=account_info.email, data=data)
|
||||
|
||||
async def async_step_reauth(
|
||||
self, entry_data: Mapping[str, Any]
|
||||
) -> ConfigFlowResult:
|
||||
"""Perform reauth upon an API authentication error."""
|
||||
return await self.async_step_reauth_confirm()
|
||||
|
||||
async def async_step_reauth_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Dialog that informs the user that reauth is required."""
|
||||
if user_input is None:
|
||||
return self.async_show_form(step_id="reauth_confirm")
|
||||
return await self.async_step_user()
|
||||
19
homeassistant/components/dropbox/const.py
Normal file
19
homeassistant/components/dropbox/const.py
Normal file
@@ -0,0 +1,19 @@
|
||||
"""Constants for the Dropbox integration."""
|
||||
|
||||
from collections.abc import Callable
|
||||
|
||||
from homeassistant.util.hass_dict import HassKey
|
||||
|
||||
DOMAIN = "dropbox"
|
||||
|
||||
OAUTH2_AUTHORIZE = "https://www.dropbox.com/oauth2/authorize"
|
||||
OAUTH2_TOKEN = "https://api.dropboxapi.com/oauth2/token"
|
||||
OAUTH2_SCOPES = [
|
||||
"account_info.read",
|
||||
"files.content.read",
|
||||
"files.content.write",
|
||||
]
|
||||
|
||||
DATA_BACKUP_AGENT_LISTENERS: HassKey[list[Callable[[], None]]] = HassKey(
|
||||
f"{DOMAIN}.backup_agent_listeners"
|
||||
)
|
||||
13
homeassistant/components/dropbox/manifest.json
Normal file
13
homeassistant/components/dropbox/manifest.json
Normal file
@@ -0,0 +1,13 @@
|
||||
{
|
||||
"domain": "dropbox",
|
||||
"name": "Dropbox",
|
||||
"after_dependencies": ["backup"],
|
||||
"codeowners": ["@bdr99"],
|
||||
"config_flow": true,
|
||||
"dependencies": ["application_credentials"],
|
||||
"documentation": "https://www.home-assistant.io/integrations/dropbox",
|
||||
"integration_type": "service",
|
||||
"iot_class": "cloud_polling",
|
||||
"quality_scale": "bronze",
|
||||
"requirements": ["python-dropbox-api==0.1.3"]
|
||||
}
|
||||
112
homeassistant/components/dropbox/quality_scale.yaml
Normal file
112
homeassistant/components/dropbox/quality_scale.yaml
Normal file
@@ -0,0 +1,112 @@
|
||||
rules:
|
||||
# Bronze
|
||||
action-setup:
|
||||
status: exempt
|
||||
comment: Integration does not register any actions.
|
||||
appropriate-polling:
|
||||
status: exempt
|
||||
comment: Integration does not poll.
|
||||
brands: done
|
||||
common-modules:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities or coordinators.
|
||||
config-flow-test-coverage: done
|
||||
config-flow: done
|
||||
dependency-transparency: done
|
||||
docs-actions:
|
||||
status: exempt
|
||||
comment: Integration does not register any actions.
|
||||
docs-high-level-description: done
|
||||
docs-installation-instructions: done
|
||||
docs-removal-instructions: done
|
||||
entity-event-setup:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities.
|
||||
entity-unique-id:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities.
|
||||
has-entity-name:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities.
|
||||
runtime-data: done
|
||||
test-before-configure: done
|
||||
test-before-setup: done
|
||||
unique-config-entry: done
|
||||
|
||||
# Silver
|
||||
action-exceptions:
|
||||
status: exempt
|
||||
comment: Integration does not register any actions.
|
||||
config-entry-unloading: done
|
||||
docs-configuration-parameters:
|
||||
status: exempt
|
||||
comment: Integration does not have any configuration parameters.
|
||||
docs-installation-parameters: done
|
||||
entity-unavailable:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities.
|
||||
integration-owner: done
|
||||
log-when-unavailable: todo
|
||||
parallel-updates:
|
||||
status: exempt
|
||||
comment: Integration does not make any entity updates.
|
||||
reauthentication-flow: done
|
||||
test-coverage: done
|
||||
|
||||
# Gold
|
||||
devices:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities.
|
||||
diagnostics:
|
||||
status: exempt
|
||||
comment: Integration does not have any data to diagnose.
|
||||
discovery-update-info:
|
||||
status: exempt
|
||||
comment: Integration is a service.
|
||||
discovery:
|
||||
status: exempt
|
||||
comment: Integration is a service.
|
||||
docs-data-update:
|
||||
status: exempt
|
||||
comment: Integration does not update any data.
|
||||
docs-examples:
|
||||
status: exempt
|
||||
comment: Integration only provides backup functionality.
|
||||
docs-known-limitations: todo
|
||||
docs-supported-devices:
|
||||
status: exempt
|
||||
comment: Integration does not support any devices.
|
||||
docs-supported-functions: done
|
||||
docs-troubleshooting: todo
|
||||
docs-use-cases: done
|
||||
dynamic-devices:
|
||||
status: exempt
|
||||
comment: Integration does not use any devices.
|
||||
entity-category:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities.
|
||||
entity-device-class:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities.
|
||||
entity-disabled-by-default:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities.
|
||||
entity-translations:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities.
|
||||
exception-translations: todo
|
||||
icon-translations:
|
||||
status: exempt
|
||||
comment: Integration does not have any entities.
|
||||
reconfiguration-flow: todo
|
||||
repair-issues:
|
||||
status: exempt
|
||||
comment: Integration does not have any repairs.
|
||||
stale-devices:
|
||||
status: exempt
|
||||
comment: Integration does not have any devices.
|
||||
|
||||
# Platinum
|
||||
async-dependency: done
|
||||
inject-websession: done
|
||||
strict-typing: done
|
||||
35
homeassistant/components/dropbox/strings.json
Normal file
35
homeassistant/components/dropbox/strings.json
Normal file
@@ -0,0 +1,35 @@
|
||||
{
|
||||
"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%]",
|
||||
"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%]",
|
||||
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
|
||||
"user_rejected_authorize": "[%key:common::config_flow::abort::oauth2_user_rejected_authorize%]",
|
||||
"wrong_account": "Wrong account: Please authenticate with the correct account."
|
||||
},
|
||||
"create_entry": {
|
||||
"default": "[%key:common::config_flow::create_entry::authenticated%]"
|
||||
},
|
||||
"step": {
|
||||
"pick_implementation": {
|
||||
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
|
||||
},
|
||||
"reauth_confirm": {
|
||||
"description": "The Dropbox integration needs to re-authenticate your account.",
|
||||
"title": "[%key:common::config_flow::title::reauth%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"exceptions": {
|
||||
"oauth2_implementation_unavailable": {
|
||||
"message": "[%key:common::exceptions::oauth2_implementation_unavailable::message%]"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -399,6 +399,47 @@ DISCOVERY_SCHEMAS = [
|
||||
),
|
||||
entity_class=MatterNumber,
|
||||
required_attributes=(clusters.OccupancySensing.Attributes.HoldTime,),
|
||||
# HoldTime is shared by PIR-specific numbers as a required attribute.
|
||||
# Keep discovery open so this generic schema does not block them.
|
||||
allow_multi=True,
|
||||
),
|
||||
MatterDiscoverySchema(
|
||||
platform=Platform.NUMBER,
|
||||
entity_description=MatterNumberEntityDescription(
|
||||
key="OccupancySensingPIRUnoccupiedToOccupiedDelay",
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
translation_key="detection_delay",
|
||||
native_max_value=65534,
|
||||
native_min_value=0,
|
||||
native_unit_of_measurement=UnitOfTime.SECONDS,
|
||||
mode=NumberMode.BOX,
|
||||
),
|
||||
entity_class=MatterNumber,
|
||||
required_attributes=(
|
||||
clusters.OccupancySensing.Attributes.PIRUnoccupiedToOccupiedDelay,
|
||||
# This attribute is mandatory when the PIRUnoccupiedToOccupiedDelay is present
|
||||
clusters.OccupancySensing.Attributes.HoldTime,
|
||||
),
|
||||
featuremap_contains=clusters.OccupancySensing.Bitmaps.Feature.kPassiveInfrared,
|
||||
allow_multi=True,
|
||||
),
|
||||
MatterDiscoverySchema(
|
||||
platform=Platform.NUMBER,
|
||||
entity_description=MatterNumberEntityDescription(
|
||||
key="OccupancySensingPIRUnoccupiedToOccupiedThreshold",
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
translation_key="detection_threshold",
|
||||
native_max_value=254,
|
||||
native_min_value=1,
|
||||
mode=NumberMode.BOX,
|
||||
),
|
||||
entity_class=MatterNumber,
|
||||
required_attributes=(
|
||||
clusters.OccupancySensing.Attributes.PIRUnoccupiedToOccupiedThreshold,
|
||||
clusters.OccupancySensing.Attributes.HoldTime,
|
||||
),
|
||||
featuremap_contains=clusters.OccupancySensing.Bitmaps.Feature.kPassiveInfrared,
|
||||
allow_multi=True,
|
||||
),
|
||||
MatterDiscoverySchema(
|
||||
platform=Platform.NUMBER,
|
||||
|
||||
@@ -214,6 +214,12 @@
|
||||
"cook_time": {
|
||||
"name": "Cooking time"
|
||||
},
|
||||
"detection_delay": {
|
||||
"name": "Detection delay"
|
||||
},
|
||||
"detection_threshold": {
|
||||
"name": "Detection threshold"
|
||||
},
|
||||
"hold_time": {
|
||||
"name": "Hold time"
|
||||
},
|
||||
|
||||
@@ -6,6 +6,7 @@ To update, run python3 -m script.hassfest
|
||||
APPLICATION_CREDENTIALS = [
|
||||
"aladdin_connect",
|
||||
"august",
|
||||
"dropbox",
|
||||
"ekeybionyx",
|
||||
"electric_kiwi",
|
||||
"fitbit",
|
||||
|
||||
1
homeassistant/generated/config_flows.py
generated
1
homeassistant/generated/config_flows.py
generated
@@ -158,6 +158,7 @@ FLOWS = {
|
||||
"downloader",
|
||||
"dremel_3d_printer",
|
||||
"drop_connect",
|
||||
"dropbox",
|
||||
"droplet",
|
||||
"dsmr",
|
||||
"dsmr_reader",
|
||||
|
||||
@@ -1473,6 +1473,12 @@
|
||||
"config_flow": true,
|
||||
"iot_class": "local_push"
|
||||
},
|
||||
"dropbox": {
|
||||
"name": "Dropbox",
|
||||
"integration_type": "service",
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_polling"
|
||||
},
|
||||
"droplet": {
|
||||
"name": "Droplet",
|
||||
"integration_type": "device",
|
||||
|
||||
@@ -171,6 +171,17 @@ ENTITY_FILTER_SELECTOR_CONFIG_SCHEMA = vol.Schema(
|
||||
)
|
||||
|
||||
|
||||
class _LegacyEntityFilterSelectorConfig(TypedDict, total=False):
|
||||
"""Class for legacy entity filter support in EntitySelectorConfig.
|
||||
|
||||
Provided for backwards compatibility and remains feature frozen.
|
||||
"""
|
||||
|
||||
integration: str
|
||||
domain: str | list[str]
|
||||
device_class: str | list[str]
|
||||
|
||||
|
||||
# Legacy entity selector config schema used directly under entity selectors
|
||||
# is provided for backwards compatibility and remains feature frozen.
|
||||
# New filtering features should be added under the `filter` key instead.
|
||||
@@ -891,9 +902,15 @@ class DurationSelector(Selector[DurationSelectorConfig]):
|
||||
return cast(dict[str, float], data)
|
||||
|
||||
|
||||
class EntitySelectorConfig(BaseSelectorConfig, EntityFilterSelectorConfig, total=False):
|
||||
class EntitySelectorConfig(
|
||||
BaseSelectorConfig, _LegacyEntityFilterSelectorConfig, total=False
|
||||
):
|
||||
"""Class to represent an entity selector config."""
|
||||
|
||||
# Note: The class inherits _LegacyEntityFilterSelectorConfig to keep
|
||||
# support for legacy entity filter at top level for backwards compatibility,
|
||||
# new entity filter options should be added under the `filter` key instead.
|
||||
|
||||
exclude_entities: list[str]
|
||||
include_entities: list[str]
|
||||
multiple: bool
|
||||
@@ -1511,6 +1528,7 @@ class StateSelectorConfig(BaseSelectorConfig, total=False):
|
||||
|
||||
entity_id: str
|
||||
hide_states: list[str]
|
||||
attribute: str
|
||||
multiple: bool
|
||||
|
||||
|
||||
@@ -1533,11 +1551,7 @@ class StateSelector(Selector[StateSelectorConfig]):
|
||||
{
|
||||
vol.Optional("entity_id"): cv.entity_id,
|
||||
vol.Optional("hide_states"): [str],
|
||||
# The attribute to filter on, is currently deliberately not
|
||||
# configurable/exposed. We are considering separating state
|
||||
# selectors into two types: one for state and one for attribute.
|
||||
# Limiting the public use, prevents breaking changes in the future.
|
||||
# vol.Optional("attribute"): str,
|
||||
vol.Optional("attribute"): str,
|
||||
vol.Optional("multiple", default=False): cv.boolean,
|
||||
}
|
||||
)
|
||||
|
||||
10
mypy.ini
generated
10
mypy.ini
generated
@@ -1486,6 +1486,16 @@ disallow_untyped_defs = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.dropbox.*]
|
||||
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.droplet.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
||||
3
requirements_all.txt
generated
3
requirements_all.txt
generated
@@ -2562,6 +2562,9 @@ python-clementine-remote==1.0.1
|
||||
# homeassistant.components.digital_ocean
|
||||
python-digitalocean==1.13.2
|
||||
|
||||
# homeassistant.components.dropbox
|
||||
python-dropbox-api==0.1.3
|
||||
|
||||
# homeassistant.components.ecobee
|
||||
python-ecobee-api==0.3.2
|
||||
|
||||
|
||||
3
requirements_test_all.txt
generated
3
requirements_test_all.txt
generated
@@ -2179,6 +2179,9 @@ python-awair==0.2.5
|
||||
# homeassistant.components.bsblan
|
||||
python-bsblan==5.1.2
|
||||
|
||||
# homeassistant.components.dropbox
|
||||
python-dropbox-api==0.1.3
|
||||
|
||||
# homeassistant.components.ecobee
|
||||
python-ecobee-api==0.3.2
|
||||
|
||||
|
||||
1
tests/components/dropbox/__init__.py
Normal file
1
tests/components/dropbox/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Tests for the Dropbox integration."""
|
||||
114
tests/components/dropbox/conftest.py
Normal file
114
tests/components/dropbox/conftest.py
Normal file
@@ -0,0 +1,114 @@
|
||||
"""Shared fixtures for Dropbox integration tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Generator
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.application_credentials import (
|
||||
ClientCredential,
|
||||
async_import_client_credential,
|
||||
)
|
||||
from homeassistant.components.dropbox.const import DOMAIN, OAUTH2_SCOPES
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
CLIENT_ID = "1234"
|
||||
CLIENT_SECRET = "5678"
|
||||
ACCOUNT_ID = "dbid:1234567890abcdef"
|
||||
ACCOUNT_EMAIL = "user@example.com"
|
||||
CONFIG_ENTRY_TITLE = "Dropbox test account"
|
||||
TEST_AGENT_ID = f"{DOMAIN}.{ACCOUNT_ID}"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def setup_credentials(hass: HomeAssistant) -> None:
|
||||
"""Set up application credentials for Dropbox."""
|
||||
|
||||
assert await async_setup_component(hass, "application_credentials", {})
|
||||
await async_import_client_credential(
|
||||
hass,
|
||||
DOMAIN,
|
||||
ClientCredential(CLIENT_ID, CLIENT_SECRET),
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def account_info() -> SimpleNamespace:
|
||||
"""Return mocked Dropbox account information."""
|
||||
|
||||
return SimpleNamespace(account_id=ACCOUNT_ID, email=ACCOUNT_EMAIL)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry() -> MockConfigEntry:
|
||||
"""Return a default Dropbox config entry."""
|
||||
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id=ACCOUNT_ID,
|
||||
title=CONFIG_ENTRY_TITLE,
|
||||
data={
|
||||
"auth_implementation": DOMAIN,
|
||||
"token": {
|
||||
"access_token": "mock-access-token",
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"expires_at": 9_999_999_999,
|
||||
"scope": " ".join(OAUTH2_SCOPES),
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock]:
|
||||
"""Override async_setup_entry."""
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.dropbox.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_dropbox_client(account_info: SimpleNamespace) -> Generator[MagicMock]:
|
||||
"""Patch DropboxAPIClient to exercise auth while mocking API calls."""
|
||||
|
||||
client = MagicMock()
|
||||
client.list_folder = AsyncMock(return_value=[])
|
||||
client.download_file = MagicMock()
|
||||
client.upload_file = AsyncMock()
|
||||
client.delete_file = AsyncMock()
|
||||
|
||||
captured_auth = None
|
||||
|
||||
def capture_auth(auth):
|
||||
nonlocal captured_auth
|
||||
captured_auth = auth
|
||||
return client
|
||||
|
||||
async def get_account_info_with_auth():
|
||||
await captured_auth.async_get_access_token()
|
||||
return client.get_account_info.return_value
|
||||
|
||||
client.get_account_info = AsyncMock(
|
||||
side_effect=get_account_info_with_auth,
|
||||
return_value=account_info,
|
||||
)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.dropbox.config_flow.DropboxAPIClient",
|
||||
side_effect=capture_auth,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.dropbox.DropboxAPIClient",
|
||||
side_effect=capture_auth,
|
||||
),
|
||||
):
|
||||
yield client
|
||||
577
tests/components/dropbox/test_backup.py
Normal file
577
tests/components/dropbox/test_backup.py
Normal file
@@ -0,0 +1,577 @@
|
||||
"""Test the Dropbox backup platform."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from io import StringIO
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
from python_dropbox_api import DropboxAuthException
|
||||
|
||||
from homeassistant.components.backup import (
|
||||
DOMAIN as BACKUP_DOMAIN,
|
||||
AddonInfo,
|
||||
AgentBackup,
|
||||
suggested_filename,
|
||||
)
|
||||
from homeassistant.components.dropbox.backup import (
|
||||
DropboxFileOrFolderNotFoundException,
|
||||
DropboxUnknownException,
|
||||
async_register_backup_agents_listener,
|
||||
)
|
||||
from homeassistant.components.dropbox.const import DATA_BACKUP_AGENT_LISTENERS, DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_REAUTH
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from .conftest import CONFIG_ENTRY_TITLE, TEST_AGENT_ID
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.test_util.aiohttp import mock_stream
|
||||
from tests.typing import ClientSessionGenerator, WebSocketGenerator
|
||||
|
||||
TEST_AGENT_BACKUP = AgentBackup(
|
||||
addons=[AddonInfo(name="Test", slug="test", version="1.0.0")],
|
||||
backup_id="dropbox-backup",
|
||||
database_included=True,
|
||||
date="2025-01-01T00:00:00.000Z",
|
||||
extra_metadata={"with_automatic_settings": False},
|
||||
folders=[],
|
||||
homeassistant_included=True,
|
||||
homeassistant_version="2024.12.0",
|
||||
name="Dropbox backup",
|
||||
protected=False,
|
||||
size=2048,
|
||||
)
|
||||
|
||||
TEST_AGENT_BACKUP_RESULT = {
|
||||
"addons": [{"name": "Test", "slug": "test", "version": "1.0.0"}],
|
||||
"agents": {TEST_AGENT_ID: {"protected": False, "size": 2048}},
|
||||
"backup_id": TEST_AGENT_BACKUP.backup_id,
|
||||
"database_included": True,
|
||||
"date": TEST_AGENT_BACKUP.date,
|
||||
"extra_metadata": {"with_automatic_settings": False},
|
||||
"failed_addons": [],
|
||||
"failed_agent_ids": [],
|
||||
"failed_folders": [],
|
||||
"folders": [],
|
||||
"homeassistant_included": True,
|
||||
"homeassistant_version": TEST_AGENT_BACKUP.homeassistant_version,
|
||||
"name": TEST_AGENT_BACKUP.name,
|
||||
"with_automatic_settings": None,
|
||||
}
|
||||
|
||||
|
||||
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"
|
||||
|
||||
|
||||
async def _mock_metadata_stream(backup: AgentBackup) -> AsyncIterator[bytes]:
|
||||
"""Create a mock metadata download stream."""
|
||||
yield json.dumps(backup.as_dict()).encode()
|
||||
|
||||
|
||||
def _setup_list_folder_with_backup(
|
||||
mock_dropbox_client: Mock,
|
||||
backup: AgentBackup,
|
||||
) -> None:
|
||||
"""Set up mock to return a backup in list_folder and download_file."""
|
||||
tar_name, metadata_name = _suggested_filenames(backup)
|
||||
mock_dropbox_client.list_folder = AsyncMock(
|
||||
return_value=[
|
||||
SimpleNamespace(name=tar_name),
|
||||
SimpleNamespace(name=metadata_name),
|
||||
]
|
||||
)
|
||||
mock_dropbox_client.download_file = Mock(return_value=_mock_metadata_stream(backup))
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def setup_integration(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_dropbox_client,
|
||||
) -> None:
|
||||
"""Set up the Dropbox and Backup integrations for testing."""
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
assert await async_setup_component(hass, BACKUP_DOMAIN, {BACKUP_DOMAIN: {}})
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
mock_dropbox_client.reset_mock()
|
||||
|
||||
|
||||
async def test_agents_info(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test listing available backup agents."""
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
|
||||
await client.send_json_auto_id({"type": "backup/agents/info"})
|
||||
response = await client.receive_json()
|
||||
|
||||
assert response["success"]
|
||||
assert response["result"] == {
|
||||
"agents": [
|
||||
{"agent_id": "backup.local", "name": "local"},
|
||||
{"agent_id": TEST_AGENT_ID, "name": CONFIG_ENTRY_TITLE},
|
||||
]
|
||||
}
|
||||
|
||||
await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await client.send_json_auto_id({"type": "backup/agents/info"})
|
||||
response = await client.receive_json()
|
||||
|
||||
assert response["success"]
|
||||
assert response["result"] == {
|
||||
"agents": [{"agent_id": "backup.local", "name": "local"}]
|
||||
}
|
||||
|
||||
|
||||
async def test_agents_list_backups(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test listing backups via the Dropbox agent."""
|
||||
|
||||
_setup_list_folder_with_backup(mock_dropbox_client, TEST_AGENT_BACKUP)
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json_auto_id({"type": "backup/info"})
|
||||
response = await client.receive_json()
|
||||
|
||||
assert response["success"]
|
||||
assert response["result"]["agent_errors"] == {}
|
||||
assert response["result"]["backups"] == [TEST_AGENT_BACKUP_RESULT]
|
||||
mock_dropbox_client.list_folder.assert_awaited()
|
||||
|
||||
|
||||
async def test_agents_list_backups_metadata_without_tar(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test that orphaned metadata files are skipped with a warning."""
|
||||
|
||||
mock_dropbox_client.list_folder = AsyncMock(
|
||||
return_value=[SimpleNamespace(name="orphan.metadata.json")]
|
||||
)
|
||||
|
||||
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"] == []
|
||||
assert "without matching backup file" in caplog.text
|
||||
|
||||
|
||||
async def test_agents_list_backups_invalid_metadata(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test that invalid metadata files are skipped with a warning."""
|
||||
|
||||
async def _invalid_stream() -> AsyncIterator[bytes]:
|
||||
yield b"not valid json"
|
||||
|
||||
mock_dropbox_client.list_folder = AsyncMock(
|
||||
return_value=[
|
||||
SimpleNamespace(name="backup.tar"),
|
||||
SimpleNamespace(name="backup.metadata.json"),
|
||||
]
|
||||
)
|
||||
mock_dropbox_client.download_file = Mock(return_value=_invalid_stream())
|
||||
|
||||
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"] == []
|
||||
assert "Skipping invalid metadata file" in caplog.text
|
||||
|
||||
|
||||
async def test_agents_list_backups_fail(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test handling list backups failures."""
|
||||
|
||||
mock_dropbox_client.list_folder = AsyncMock(
|
||||
side_effect=DropboxUnknownException("boom")
|
||||
)
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json_auto_id({"type": "backup/info"})
|
||||
response = await client.receive_json()
|
||||
|
||||
assert response["success"]
|
||||
assert response["result"]["backups"] == []
|
||||
assert response["result"]["agent_errors"] == {
|
||||
TEST_AGENT_ID: "Failed to list backups"
|
||||
}
|
||||
|
||||
|
||||
async def test_agents_list_backups_reauth(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test reauthentication is triggered on auth error."""
|
||||
|
||||
mock_dropbox_client.list_folder = AsyncMock(
|
||||
side_effect=DropboxAuthException("auth failed")
|
||||
)
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json_auto_id({"type": "backup/info"})
|
||||
response = await client.receive_json()
|
||||
|
||||
assert response["success"]
|
||||
assert response["result"]["backups"] == []
|
||||
assert response["result"]["agent_errors"] == {TEST_AGENT_ID: "Authentication error"}
|
||||
|
||||
await hass.async_block_till_done()
|
||||
flows = hass.config_entries.flow.async_progress()
|
||||
assert len(flows) == 1
|
||||
|
||||
flow = flows[0]
|
||||
assert flow["step_id"] == "reauth_confirm"
|
||||
assert flow["handler"] == DOMAIN
|
||||
assert flow["context"]["source"] == SOURCE_REAUTH
|
||||
assert flow["context"]["entry_id"] == mock_config_entry.entry_id
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"backup_id",
|
||||
[TEST_AGENT_BACKUP.backup_id, "other-backup"],
|
||||
ids=["found", "not_found"],
|
||||
)
|
||||
async def test_agents_get_backup(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
backup_id: str,
|
||||
) -> None:
|
||||
"""Test retrieving a backup's metadata."""
|
||||
|
||||
_setup_list_folder_with_backup(mock_dropbox_client, TEST_AGENT_BACKUP)
|
||||
|
||||
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"] == {}
|
||||
if backup_id == TEST_AGENT_BACKUP.backup_id:
|
||||
assert response["result"]["backup"] == TEST_AGENT_BACKUP_RESULT
|
||||
else:
|
||||
assert response["result"]["backup"] is None
|
||||
|
||||
|
||||
async def test_agents_download(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test downloading a backup file."""
|
||||
|
||||
tar_name, metadata_name = _suggested_filenames(TEST_AGENT_BACKUP)
|
||||
|
||||
mock_dropbox_client.list_folder = AsyncMock(
|
||||
return_value=[
|
||||
SimpleNamespace(name=tar_name),
|
||||
SimpleNamespace(name=metadata_name),
|
||||
]
|
||||
)
|
||||
|
||||
def download_side_effect(path: str) -> AsyncIterator[bytes]:
|
||||
if path == f"/{tar_name}":
|
||||
return mock_stream(b"backup data")
|
||||
return _mock_metadata_stream(TEST_AGENT_BACKUP)
|
||||
|
||||
mock_dropbox_client.download_file = Mock(side_effect=download_side_effect)
|
||||
|
||||
client = await hass_client()
|
||||
resp = await client.get(
|
||||
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
|
||||
)
|
||||
|
||||
assert resp.status == 200
|
||||
assert await resp.content.read() == b"backup data"
|
||||
|
||||
|
||||
async def test_agents_download_fail(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test handling download failures."""
|
||||
|
||||
mock_dropbox_client.list_folder = AsyncMock(
|
||||
side_effect=DropboxUnknownException("boom")
|
||||
)
|
||||
|
||||
client = await hass_client()
|
||||
resp = await client.get(
|
||||
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
|
||||
)
|
||||
|
||||
assert resp.status == 500
|
||||
body = await resp.content.read()
|
||||
assert b"Failed to get backup" in body
|
||||
|
||||
|
||||
async def test_agents_download_not_found(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test download when backup disappears between get and download."""
|
||||
|
||||
tar_name, metadata_name = _suggested_filenames(TEST_AGENT_BACKUP)
|
||||
files = [
|
||||
SimpleNamespace(name=tar_name),
|
||||
SimpleNamespace(name=metadata_name),
|
||||
]
|
||||
|
||||
# First list_folder call (async_get_backup) returns the backup;
|
||||
# second call (async_download_backup) returns empty, simulating deletion.
|
||||
mock_dropbox_client.list_folder = AsyncMock(side_effect=[files, []])
|
||||
mock_dropbox_client.download_file = Mock(
|
||||
return_value=_mock_metadata_stream(TEST_AGENT_BACKUP)
|
||||
)
|
||||
|
||||
client = await hass_client()
|
||||
resp = await client.get(
|
||||
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
|
||||
)
|
||||
|
||||
assert resp.status == 404
|
||||
assert await resp.content.read() == b""
|
||||
|
||||
|
||||
async def test_agents_download_file_not_found(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test download when Dropbox file is not found returns 404."""
|
||||
|
||||
mock_dropbox_client.list_folder = AsyncMock(
|
||||
side_effect=DropboxFileOrFolderNotFoundException("not found")
|
||||
)
|
||||
|
||||
client = await hass_client()
|
||||
resp = await client.get(
|
||||
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
|
||||
)
|
||||
|
||||
assert resp.status == 404
|
||||
|
||||
|
||||
async def test_agents_download_metadata_not_found(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test download when metadata lookup fails."""
|
||||
|
||||
mock_dropbox_client.list_folder = AsyncMock(return_value=[])
|
||||
|
||||
client = await hass_client()
|
||||
resp = await client.get(
|
||||
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
|
||||
)
|
||||
|
||||
assert resp.status == 404
|
||||
assert await resp.content.read() == b""
|
||||
|
||||
|
||||
async def test_agents_upload(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test uploading a backup to Dropbox."""
|
||||
|
||||
mock_dropbox_client.upload_file = AsyncMock(return_value=None)
|
||||
|
||||
client = await hass_client()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
|
||||
return_value=TEST_AGENT_BACKUP,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.backup.manager.read_backup",
|
||||
return_value=TEST_AGENT_BACKUP,
|
||||
),
|
||||
patch("pathlib.Path.open") as mocked_open,
|
||||
):
|
||||
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
|
||||
resp = await client.post(
|
||||
f"/api/backup/upload?agent_id={TEST_AGENT_ID}",
|
||||
data={"file": StringIO("test")},
|
||||
)
|
||||
|
||||
assert resp.status == 201
|
||||
assert f"Uploading backup {TEST_AGENT_BACKUP.backup_id} to agents" in caplog.text
|
||||
assert mock_dropbox_client.upload_file.await_count == 2
|
||||
|
||||
|
||||
async def test_agents_upload_fail(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test that backup tar is cleaned up when metadata upload fails."""
|
||||
|
||||
call_count = 0
|
||||
|
||||
async def upload_side_effect(path: str, stream: AsyncIterator[bytes]) -> None:
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
async for _ in stream:
|
||||
pass
|
||||
if call_count == 2:
|
||||
raise DropboxUnknownException("metadata upload failed")
|
||||
|
||||
mock_dropbox_client.upload_file = AsyncMock(side_effect=upload_side_effect)
|
||||
mock_dropbox_client.delete_file = AsyncMock()
|
||||
|
||||
client = await hass_client()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
|
||||
return_value=TEST_AGENT_BACKUP,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.backup.manager.read_backup",
|
||||
return_value=TEST_AGENT_BACKUP,
|
||||
),
|
||||
patch("pathlib.Path.open") as mocked_open,
|
||||
):
|
||||
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
|
||||
resp = await client.post(
|
||||
f"/api/backup/upload?agent_id={TEST_AGENT_ID}",
|
||||
data={"file": StringIO("test")},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert resp.status == 201
|
||||
assert "Failed to upload backup" in caplog.text
|
||||
mock_dropbox_client.delete_file.assert_awaited_once()
|
||||
|
||||
|
||||
async def test_agents_delete(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test deleting a backup."""
|
||||
|
||||
_setup_list_folder_with_backup(mock_dropbox_client, TEST_AGENT_BACKUP)
|
||||
mock_dropbox_client.delete_file = AsyncMock(return_value=None)
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json_auto_id(
|
||||
{
|
||||
"type": "backup/delete",
|
||||
"backup_id": TEST_AGENT_BACKUP.backup_id,
|
||||
}
|
||||
)
|
||||
response = await client.receive_json()
|
||||
|
||||
assert response["success"]
|
||||
assert response["result"] == {"agent_errors": {}}
|
||||
assert mock_dropbox_client.delete_file.await_count == 2
|
||||
|
||||
|
||||
async def test_agents_delete_fail(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test error handling when delete fails."""
|
||||
|
||||
mock_dropbox_client.list_folder = AsyncMock(
|
||||
side_effect=DropboxUnknownException("boom")
|
||||
)
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json_auto_id(
|
||||
{
|
||||
"type": "backup/delete",
|
||||
"backup_id": TEST_AGENT_BACKUP.backup_id,
|
||||
}
|
||||
)
|
||||
response = await client.receive_json()
|
||||
|
||||
assert response["success"]
|
||||
assert response["result"] == {
|
||||
"agent_errors": {TEST_AGENT_ID: "Failed to delete backup"}
|
||||
}
|
||||
|
||||
|
||||
async def test_agents_delete_not_found(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
mock_dropbox_client: Mock,
|
||||
) -> None:
|
||||
"""Test deleting a backup that does not exist."""
|
||||
|
||||
mock_dropbox_client.list_folder = AsyncMock(return_value=[])
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json_auto_id(
|
||||
{
|
||||
"type": "backup/delete",
|
||||
"backup_id": TEST_AGENT_BACKUP.backup_id,
|
||||
}
|
||||
)
|
||||
response = await client.receive_json()
|
||||
|
||||
assert response["success"]
|
||||
assert response["result"] == {"agent_errors": {}}
|
||||
|
||||
|
||||
async def test_remove_backup_agents_listener(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test removing a backup agent listener."""
|
||||
listener = Mock()
|
||||
remove = async_register_backup_agents_listener(hass, listener=listener)
|
||||
|
||||
assert DATA_BACKUP_AGENT_LISTENERS in hass.data
|
||||
assert listener in hass.data[DATA_BACKUP_AGENT_LISTENERS]
|
||||
|
||||
# Remove all other listeners to test the cleanup path
|
||||
hass.data[DATA_BACKUP_AGENT_LISTENERS] = [listener]
|
||||
|
||||
remove()
|
||||
|
||||
assert DATA_BACKUP_AGENT_LISTENERS not in hass.data
|
||||
210
tests/components/dropbox/test_config_flow.py
Normal file
210
tests/components/dropbox/test_config_flow.py
Normal file
@@ -0,0 +1,210 @@
|
||||
"""Test the Dropbox config flow."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from yarl import URL
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.dropbox.const import (
|
||||
DOMAIN,
|
||||
OAUTH2_AUTHORIZE,
|
||||
OAUTH2_SCOPES,
|
||||
OAUTH2_TOKEN,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
from homeassistant.helpers import config_entry_oauth2_flow
|
||||
|
||||
from .conftest import ACCOUNT_EMAIL, ACCOUNT_ID, CLIENT_ID
|
||||
|
||||
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||
from tests.typing import ClientSessionGenerator
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("current_request_with_host")
|
||||
async def test_full_flow(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
mock_dropbox_client,
|
||||
mock_setup_entry: AsyncMock,
|
||||
) -> None:
|
||||
"""Test creating a new config entry through the OAuth flow."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
state = config_entry_oauth2_flow._encode_jwt(
|
||||
hass,
|
||||
{
|
||||
"flow_id": result["flow_id"],
|
||||
"redirect_uri": "https://example.com/auth/external/callback",
|
||||
},
|
||||
)
|
||||
|
||||
result_url = URL(result["url"])
|
||||
assert f"{result_url.origin()}{result_url.path}" == OAUTH2_AUTHORIZE
|
||||
assert result_url.query["response_type"] == "code"
|
||||
assert result_url.query["client_id"] == CLIENT_ID
|
||||
assert (
|
||||
result_url.query["redirect_uri"] == "https://example.com/auth/external/callback"
|
||||
)
|
||||
assert result_url.query["state"] == state
|
||||
assert result_url.query["scope"] == " ".join(OAUTH2_SCOPES)
|
||||
assert result_url.query["token_access_type"] == "offline"
|
||||
assert result_url.query["code_challenge"]
|
||||
assert result_url.query["code_challenge_method"] == "S256"
|
||||
|
||||
client = await hass_client_no_auth()
|
||||
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
|
||||
assert resp.status == 200
|
||||
assert resp.headers["content-type"] == "text/html; charset=utf-8"
|
||||
|
||||
aioclient_mock.post(
|
||||
OAUTH2_TOKEN,
|
||||
json={
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"access_token": "mock-access-token",
|
||||
"token_type": "Bearer",
|
||||
"expires_in": 60,
|
||||
},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == ACCOUNT_EMAIL
|
||||
assert result["data"]["token"]["access_token"] == "mock-access-token"
|
||||
assert result["result"].unique_id == ACCOUNT_ID
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("current_request_with_host")
|
||||
async def test_already_configured(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
mock_config_entry,
|
||||
mock_dropbox_client,
|
||||
) -> None:
|
||||
"""Test aborting when the account is already configured."""
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
state = config_entry_oauth2_flow._encode_jwt(
|
||||
hass,
|
||||
{
|
||||
"flow_id": result["flow_id"],
|
||||
"redirect_uri": "https://example.com/auth/external/callback",
|
||||
},
|
||||
)
|
||||
|
||||
client = await hass_client_no_auth()
|
||||
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
|
||||
assert resp.status == 200
|
||||
|
||||
aioclient_mock.post(
|
||||
OAUTH2_TOKEN,
|
||||
json={
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"access_token": "mock-access-token",
|
||||
"token_type": "Bearer",
|
||||
"expires_in": 60,
|
||||
},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("current_request_with_host")
|
||||
@pytest.mark.parametrize(
|
||||
(
|
||||
"new_account_info",
|
||||
"expected_reason",
|
||||
"expected_setup_calls",
|
||||
"expected_access_token",
|
||||
),
|
||||
[
|
||||
(
|
||||
SimpleNamespace(account_id=ACCOUNT_ID, email=ACCOUNT_EMAIL),
|
||||
"reauth_successful",
|
||||
1,
|
||||
"updated-access-token",
|
||||
),
|
||||
(
|
||||
SimpleNamespace(account_id="dbid:different", email="other@example.com"),
|
||||
"wrong_account",
|
||||
0,
|
||||
"mock-access-token",
|
||||
),
|
||||
],
|
||||
ids=["success", "wrong_account"],
|
||||
)
|
||||
async def test_reauth_flow(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
mock_config_entry,
|
||||
mock_dropbox_client,
|
||||
mock_setup_entry: AsyncMock,
|
||||
new_account_info: SimpleNamespace,
|
||||
expected_reason: str,
|
||||
expected_setup_calls: int,
|
||||
expected_access_token: str,
|
||||
) -> None:
|
||||
"""Test reauthentication flow outcomes."""
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
mock_dropbox_client.get_account_info.return_value = new_account_info
|
||||
|
||||
result = await mock_config_entry.start_reauth_flow(hass)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "reauth_confirm"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
|
||||
|
||||
state = config_entry_oauth2_flow._encode_jwt(
|
||||
hass,
|
||||
{
|
||||
"flow_id": result["flow_id"],
|
||||
"redirect_uri": "https://example.com/auth/external/callback",
|
||||
},
|
||||
)
|
||||
|
||||
client = await hass_client_no_auth()
|
||||
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
|
||||
assert resp.status == 200
|
||||
|
||||
aioclient_mock.post(
|
||||
OAUTH2_TOKEN,
|
||||
json={
|
||||
"refresh_token": "mock-refresh-token",
|
||||
"access_token": "updated-access-token",
|
||||
"token_type": "Bearer",
|
||||
"expires_in": 120,
|
||||
},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(result["flow_id"])
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == expected_reason
|
||||
assert mock_setup_entry.await_count == expected_setup_calls
|
||||
|
||||
assert mock_config_entry.data["token"]["access_token"] == expected_access_token
|
||||
100
tests/components/dropbox/test_init.py
Normal file
100
tests/components/dropbox/test_init.py
Normal file
@@ -0,0 +1,100 @@
|
||||
"""Test the Dropbox integration setup."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from python_dropbox_api import DropboxAuthException, DropboxUnknownException
|
||||
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.config_entry_oauth2_flow import (
|
||||
ImplementationUnavailableError,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_dropbox_client")
|
||||
async def test_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test successful setup of a config entry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
|
||||
async def test_setup_entry_auth_failed(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_dropbox_client: AsyncMock,
|
||||
) -> None:
|
||||
"""Test setup failure when authentication fails."""
|
||||
mock_dropbox_client.get_account_info.side_effect = DropboxAuthException(
|
||||
"Invalid token"
|
||||
)
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"side_effect",
|
||||
[DropboxUnknownException("Unknown error"), TimeoutError("Connection timed out")],
|
||||
ids=["unknown_exception", "timeout_error"],
|
||||
)
|
||||
async def test_setup_entry_not_ready(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_dropbox_client: AsyncMock,
|
||||
side_effect: Exception,
|
||||
) -> None:
|
||||
"""Test setup retry when the service is temporarily unavailable."""
|
||||
mock_dropbox_client.get_account_info.side_effect = side_effect
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
async def test_setup_entry_implementation_unavailable(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test setup retry when OAuth implementation is unavailable."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.dropbox.async_get_config_entry_implementation",
|
||||
side_effect=ImplementationUnavailableError,
|
||||
):
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_dropbox_client")
|
||||
async def test_unload_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test unloading a config entry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
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
|
||||
@@ -70,6 +70,7 @@ FIXTURES = [
|
||||
"mock_microwave_oven",
|
||||
"mock_mounted_dimmable_load_control_fixture",
|
||||
"mock_occupancy_sensor",
|
||||
"mock_occupancy_sensor_pir",
|
||||
"mock_on_off_plugin_unit",
|
||||
"mock_onoff_light",
|
||||
"mock_onoff_light_alt_name",
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
{
|
||||
"node_id": 98,
|
||||
"date_commissioned": "2022-11-29T21:23:48.485051",
|
||||
"last_interview": "2022-11-29T21:23:48.485057",
|
||||
"interview_version": 2,
|
||||
"attributes": {
|
||||
"0/29/0": [
|
||||
{
|
||||
"0": 22,
|
||||
"1": 1
|
||||
}
|
||||
],
|
||||
"0/29/1": [
|
||||
4, 29, 31, 40, 42, 43, 44, 48, 49, 50, 51, 52, 53, 54, 55, 59, 60, 62, 63,
|
||||
64, 65
|
||||
],
|
||||
"0/29/2": [41],
|
||||
"0/29/3": [1],
|
||||
"0/29/65532": 0,
|
||||
"0/29/65533": 1,
|
||||
"0/29/65528": [],
|
||||
"0/29/65529": [],
|
||||
"0/29/65531": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
|
||||
"0/40/0": 1,
|
||||
"0/40/1": "Nabu Casa",
|
||||
"0/40/2": 65521,
|
||||
"0/40/3": "Mock PIR Occupancy Sensor",
|
||||
"0/40/4": 32768,
|
||||
"0/40/5": "Mock PIR Occupancy Sensor",
|
||||
"0/40/6": "XX",
|
||||
"0/40/7": 0,
|
||||
"0/40/8": "v1.0",
|
||||
"0/40/9": 1,
|
||||
"0/40/10": "v1.0",
|
||||
"0/40/11": "20260206",
|
||||
"0/40/12": "",
|
||||
"0/40/13": "",
|
||||
"0/40/14": "",
|
||||
"0/40/15": "TEST_SN",
|
||||
"0/40/16": false,
|
||||
"0/40/17": true,
|
||||
"0/40/18": "mock-pir-occupancy-sensor",
|
||||
"0/40/19": {
|
||||
"0": 3,
|
||||
"1": 3
|
||||
},
|
||||
"0/40/65532": 0,
|
||||
"0/40/65533": 1,
|
||||
"0/40/65528": [],
|
||||
"0/40/65529": [],
|
||||
"0/40/65531": [
|
||||
0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19,
|
||||
65528, 65529, 65531, 65532, 65533
|
||||
],
|
||||
"1/3/0": 0,
|
||||
"1/3/1": 2,
|
||||
"1/3/65532": 0,
|
||||
"1/3/65533": 4,
|
||||
"1/3/65528": [],
|
||||
"1/3/65529": [0, 64],
|
||||
"1/3/65531": [0, 1, 65528, 65529, 65531, 65532, 65533],
|
||||
"1/29/0": [
|
||||
{
|
||||
"0": 263,
|
||||
"1": 1
|
||||
}
|
||||
],
|
||||
"1/29/1": [
|
||||
3, 4, 5, 6, 7, 8, 15, 29, 30, 37, 47, 59, 64, 65, 69, 80, 257, 258, 259,
|
||||
512, 513, 514, 516, 768, 1024, 1026, 1027, 1028, 1029, 1030, 1283, 1284,
|
||||
1285, 1286, 1287, 1288, 1289, 1290, 1291, 1292, 1293, 1294, 2820,
|
||||
4294048773
|
||||
],
|
||||
"1/29/2": [],
|
||||
"1/29/3": [],
|
||||
"1/29/65532": 0,
|
||||
"1/29/65533": 1,
|
||||
"1/29/65528": [],
|
||||
"1/29/65529": [],
|
||||
"1/29/65531": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
|
||||
"1/30/0": [],
|
||||
"1/30/65532": 0,
|
||||
"1/30/65533": 1,
|
||||
"1/30/65528": [],
|
||||
"1/30/65529": [],
|
||||
"1/30/65531": [0, 65528, 65529, 65531, 65532, 65533],
|
||||
"1/1030/0": 1,
|
||||
"1/1030/1": 0,
|
||||
"1/1030/2": 1,
|
||||
"1/1030/3": 10,
|
||||
"1/1030/4": {
|
||||
"0": 1,
|
||||
"1": 65534,
|
||||
"2": 10
|
||||
},
|
||||
"1/1030/17": 10,
|
||||
"1/1030/18": 1,
|
||||
"1/1030/65532": 2,
|
||||
"1/1030/65533": 5,
|
||||
"1/1030/65528": [],
|
||||
"1/1030/65529": [],
|
||||
"1/1030/65531": [0, 1, 2, 3, 4, 17, 18, 65528, 65529, 65531, 65532, 65533]
|
||||
},
|
||||
"available": true,
|
||||
"attribute_subscriptions": []
|
||||
}
|
||||
@@ -1623,6 +1623,57 @@
|
||||
'state': 'on',
|
||||
})
|
||||
# ---
|
||||
# name: test_binary_sensors[mock_occupancy_sensor_pir][binary_sensor.mock_pir_occupancy_sensor_occupancy-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'area_id': None,
|
||||
'capabilities': None,
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'binary_sensor',
|
||||
'entity_category': None,
|
||||
'entity_id': 'binary_sensor.mock_pir_occupancy_sensor_occupancy',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': 'Occupancy',
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': <BinarySensorDeviceClass.OCCUPANCY: 'occupancy'>,
|
||||
'original_icon': None,
|
||||
'original_name': 'Occupancy',
|
||||
'platform': 'matter',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': None,
|
||||
'unique_id': '00000000000004D2-0000000000000062-MatterNodeDevice-1-OccupancySensor-1030-0',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_binary_sensors[mock_occupancy_sensor_pir][binary_sensor.mock_pir_occupancy_sensor_occupancy-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'device_class': 'occupancy',
|
||||
'friendly_name': 'Mock PIR Occupancy Sensor Occupancy',
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'binary_sensor.mock_pir_occupancy_sensor_occupancy',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'on',
|
||||
})
|
||||
# ---
|
||||
# name: test_binary_sensors[mock_onoff_light_alt_name][binary_sensor.mock_onoff_light_occupancy-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
|
||||
@@ -3147,6 +3147,57 @@
|
||||
'state': 'unknown',
|
||||
})
|
||||
# ---
|
||||
# name: test_buttons[mock_occupancy_sensor_pir][button.mock_pir_occupancy_sensor_identify-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'area_id': None,
|
||||
'capabilities': None,
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'button',
|
||||
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
|
||||
'entity_id': 'button.mock_pir_occupancy_sensor_identify',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': 'Identify',
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': <ButtonDeviceClass.IDENTIFY: 'identify'>,
|
||||
'original_icon': None,
|
||||
'original_name': 'Identify',
|
||||
'platform': 'matter',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': None,
|
||||
'unique_id': '00000000000004D2-0000000000000062-MatterNodeDevice-1-IdentifyButton-3-1',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_buttons[mock_occupancy_sensor_pir][button.mock_pir_occupancy_sensor_identify-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'device_class': 'identify',
|
||||
'friendly_name': 'Mock PIR Occupancy Sensor Identify',
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'button.mock_pir_occupancy_sensor_identify',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'unknown',
|
||||
})
|
||||
# ---
|
||||
# name: test_buttons[mock_on_off_plugin_unit][button.mock_onoffpluginunit_identify-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
|
||||
@@ -3040,6 +3040,185 @@
|
||||
'state': 'unavailable',
|
||||
})
|
||||
# ---
|
||||
# name: test_numbers[mock_occupancy_sensor_pir][number.mock_pir_occupancy_sensor_detection_delay-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'area_id': None,
|
||||
'capabilities': dict({
|
||||
'max': 65534,
|
||||
'min': 0,
|
||||
'mode': <NumberMode.BOX: 'box'>,
|
||||
'step': 1.0,
|
||||
}),
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'number',
|
||||
'entity_category': <EntityCategory.CONFIG: 'config'>,
|
||||
'entity_id': 'number.mock_pir_occupancy_sensor_detection_delay',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': 'Detection delay',
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': None,
|
||||
'original_icon': None,
|
||||
'original_name': 'Detection delay',
|
||||
'platform': 'matter',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': 'detection_delay',
|
||||
'unique_id': '00000000000004D2-0000000000000062-MatterNodeDevice-1-OccupancySensingPIRUnoccupiedToOccupiedDelay-1030-17',
|
||||
'unit_of_measurement': <UnitOfTime.SECONDS: 's'>,
|
||||
})
|
||||
# ---
|
||||
# name: test_numbers[mock_occupancy_sensor_pir][number.mock_pir_occupancy_sensor_detection_delay-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'friendly_name': 'Mock PIR Occupancy Sensor Detection delay',
|
||||
'max': 65534,
|
||||
'min': 0,
|
||||
'mode': <NumberMode.BOX: 'box'>,
|
||||
'step': 1.0,
|
||||
'unit_of_measurement': <UnitOfTime.SECONDS: 's'>,
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'number.mock_pir_occupancy_sensor_detection_delay',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': '10',
|
||||
})
|
||||
# ---
|
||||
# name: test_numbers[mock_occupancy_sensor_pir][number.mock_pir_occupancy_sensor_detection_threshold-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'area_id': None,
|
||||
'capabilities': dict({
|
||||
'max': 254,
|
||||
'min': 1,
|
||||
'mode': <NumberMode.BOX: 'box'>,
|
||||
'step': 1.0,
|
||||
}),
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'number',
|
||||
'entity_category': <EntityCategory.CONFIG: 'config'>,
|
||||
'entity_id': 'number.mock_pir_occupancy_sensor_detection_threshold',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': 'Detection threshold',
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': None,
|
||||
'original_icon': None,
|
||||
'original_name': 'Detection threshold',
|
||||
'platform': 'matter',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': 'detection_threshold',
|
||||
'unique_id': '00000000000004D2-0000000000000062-MatterNodeDevice-1-OccupancySensingPIRUnoccupiedToOccupiedThreshold-1030-18',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_numbers[mock_occupancy_sensor_pir][number.mock_pir_occupancy_sensor_detection_threshold-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'friendly_name': 'Mock PIR Occupancy Sensor Detection threshold',
|
||||
'max': 254,
|
||||
'min': 1,
|
||||
'mode': <NumberMode.BOX: 'box'>,
|
||||
'step': 1.0,
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'number.mock_pir_occupancy_sensor_detection_threshold',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': '1',
|
||||
})
|
||||
# ---
|
||||
# name: test_numbers[mock_occupancy_sensor_pir][number.mock_pir_occupancy_sensor_hold_time-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'area_id': None,
|
||||
'capabilities': dict({
|
||||
'max': 65534,
|
||||
'min': 1,
|
||||
'mode': <NumberMode.BOX: 'box'>,
|
||||
'step': 1.0,
|
||||
}),
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'number',
|
||||
'entity_category': <EntityCategory.CONFIG: 'config'>,
|
||||
'entity_id': 'number.mock_pir_occupancy_sensor_hold_time',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': 'Hold time',
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': None,
|
||||
'original_icon': None,
|
||||
'original_name': 'Hold time',
|
||||
'platform': 'matter',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': 'hold_time',
|
||||
'unique_id': '00000000000004D2-0000000000000062-MatterNodeDevice-1-OccupancySensingHoldTime-1030-3',
|
||||
'unit_of_measurement': <UnitOfTime.SECONDS: 's'>,
|
||||
})
|
||||
# ---
|
||||
# name: test_numbers[mock_occupancy_sensor_pir][number.mock_pir_occupancy_sensor_hold_time-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'friendly_name': 'Mock PIR Occupancy Sensor Hold time',
|
||||
'max': 65534,
|
||||
'min': 1,
|
||||
'mode': <NumberMode.BOX: 'box'>,
|
||||
'step': 1.0,
|
||||
'unit_of_measurement': <UnitOfTime.SECONDS: 's'>,
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'number.mock_pir_occupancy_sensor_hold_time',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': '10',
|
||||
})
|
||||
# ---
|
||||
# name: test_numbers[mock_on_off_plugin_unit][number.mock_onoffpluginunit_off_transition_time-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
|
||||
@@ -395,3 +395,79 @@ async def test_matter_exception_on_door_lock_write_attribute(
|
||||
)
|
||||
|
||||
assert str(exc_info.value) == "Boom!"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("node_fixture", ["mock_occupancy_sensor_pir"])
|
||||
async def test_occupancy_sensing_pir_attributes(
|
||||
hass: HomeAssistant,
|
||||
matter_client: MagicMock,
|
||||
matter_node: MatterNode,
|
||||
) -> None:
|
||||
"""Test PIR occupancy sensor attributes."""
|
||||
# PIRUnoccupiedToOccupiedDelay
|
||||
state = hass.states.get("number.mock_pir_occupancy_sensor_detection_delay")
|
||||
assert state
|
||||
assert state.state == "10"
|
||||
assert state.attributes["min"] == 0
|
||||
assert state.attributes["max"] == 65534
|
||||
assert state.attributes["unit_of_measurement"] == "s"
|
||||
|
||||
set_node_attribute(matter_node, 1, 1030, 17, 20)
|
||||
await trigger_subscription_callback(hass, matter_client)
|
||||
state = hass.states.get("number.mock_pir_occupancy_sensor_detection_delay")
|
||||
assert state
|
||||
assert state.state == "20"
|
||||
|
||||
# PIRUnoccupiedToOccupiedThreshold
|
||||
state = hass.states.get("number.mock_pir_occupancy_sensor_detection_threshold")
|
||||
assert state
|
||||
assert state.state == "1"
|
||||
assert state.attributes["min"] == 1
|
||||
assert state.attributes["max"] == 254
|
||||
|
||||
set_node_attribute(matter_node, 1, 1030, 18, 5)
|
||||
await trigger_subscription_callback(hass, matter_client)
|
||||
state = hass.states.get("number.mock_pir_occupancy_sensor_detection_threshold")
|
||||
assert state
|
||||
assert state.state == "5"
|
||||
|
||||
# Test set value for delay
|
||||
await hass.services.async_call(
|
||||
"number",
|
||||
"set_value",
|
||||
{
|
||||
"entity_id": "number.mock_pir_occupancy_sensor_detection_delay",
|
||||
"value": 15,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
assert matter_client.write_attribute.call_count == 1
|
||||
assert matter_client.write_attribute.call_args_list[0] == call(
|
||||
node_id=matter_node.node_id,
|
||||
attribute_path=create_attribute_path_from_attribute(
|
||||
endpoint_id=1,
|
||||
attribute=clusters.OccupancySensing.Attributes.PIRUnoccupiedToOccupiedDelay,
|
||||
),
|
||||
value=15,
|
||||
)
|
||||
|
||||
# Test set value for threshold
|
||||
matter_client.write_attribute.reset_mock()
|
||||
await hass.services.async_call(
|
||||
"number",
|
||||
"set_value",
|
||||
{
|
||||
"entity_id": "number.mock_pir_occupancy_sensor_detection_threshold",
|
||||
"value": 3,
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
assert matter_client.write_attribute.call_count == 1
|
||||
assert matter_client.write_attribute.call_args_list[0] == call(
|
||||
node_id=matter_node.node_id,
|
||||
attribute_path=create_attribute_path_from_attribute(
|
||||
endpoint_id=1,
|
||||
attribute=clusters.OccupancySensing.Attributes.PIRUnoccupiedToOccupiedThreshold,
|
||||
),
|
||||
value=3,
|
||||
)
|
||||
|
||||
@@ -883,6 +883,11 @@ def test_time_selector_schema(schema, valid_selections, invalid_selections) -> N
|
||||
(),
|
||||
(),
|
||||
),
|
||||
(
|
||||
{"attribute": "best_attribute"},
|
||||
(),
|
||||
(),
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_state_selector_schema(schema, valid_selections, invalid_selections) -> None:
|
||||
|
||||
Reference in New Issue
Block a user