Compare commits

...

42 Commits

Author SHA1 Message Date
Ludovic BOUÉ
9cadf32e36 Update snapshot aliases from set to list for consistency in test_binary_sensor, test_button, and test_number 2026-03-19 06:57:04 +00:00
Ludovic BOUÉ
d44387e36b Merge branch 'dev' into PIRUnoccupiedToOccupiedDelay 2026-03-18 18:45:18 +01:00
Erik Montnemery
538b817bf1 Adjust inheritance tree of EntitySelectorConfig (#165915) 2026-03-18 18:32:44 +01:00
Brandon Rothweiler
7efa2d3cac Add Dropbox backup integration (#155644) 2026-03-18 17:58:57 +01:00
Erik Montnemery
3f872fd196 Allow specifying attribute in state selector (#165928) 2026-03-18 17:54:36 +01:00
Ludovic BOUÉ
1824ef12bb Merge branch 'dev' into PIRUnoccupiedToOccupiedDelay 2026-03-14 08:34:44 +01:00
Ludovic BOUÉ
c706e8a5b8 Merge branch 'dev' into PIRUnoccupiedToOccupiedDelay 2026-03-13 18:56:24 +01:00
Ludovic BOUÉ
5bd9742eb3 Potential fix for pull request finding
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
2026-03-13 18:47:52 +01:00
Ludovic BOUÉ
26f3eb5f6d Update snapshots 2026-03-13 17:41:22 +00:00
Ludovic BOUÉ
7a34d4f881 Merge branch 'dev' into PIRUnoccupiedToOccupiedDelay 2026-03-13 18:33:41 +01:00
Ludovic BOUÉ
e0a37a5eeb Merge branch 'dev' into PIRUnoccupiedToOccupiedDelay 2026-03-11 11:52:51 +01:00
Ludovic BOUÉ
ec3d1fd72c Merge branch 'dev' into PIRUnoccupiedToOccupiedDelay 2026-03-07 18:07:37 +01:00
Ludovic BOUÉ
4edea21cb7 Update unique_id in snapshots 2026-03-06 16:13:57 +00:00
Ludovic BOUÉ
7f065c1942 Add allow_multi attribute to occupancy sensing discovery schemas 2026-03-06 15:58:04 +00:00
Ludovic BOUÉ
46ce07a9a1 Update mock occupancy sensor fixture OccupancySensing revision attribute 2026-03-06 15:49:17 +00:00
Ludovic BOUÉ
5807db2c60 Add HoldTime and HoldTimeLimits attributes to mock occupancy sensor fixture for conformance 2026-03-06 15:46:12 +00:00
Ludovic BOUÉ
85732543b2 Update node_id in mock occupancy sensor fixture to match expected value 2026-03-06 15:33:35 +00:00
Ludovic BOUÉ
054c61d73f Merge branch 'dev' into PIRUnoccupiedToOccupiedDelay 2026-03-06 16:20:39 +01:00
Ludovic BOUÉ
be2c20c624 Add HoldTime attribute to occupancy sensing discovery schema 2026-03-06 15:19:28 +00:00
Ludovic BOUÉ
706127c9ea Add mock_occupancy_sensor_pir to common.py 2026-02-12 11:13:19 +01:00
Ludovic BOUÉ
b163829970 Merge branch 'dev' into PIRUnoccupiedToOccupiedDelay 2026-02-12 10:49:35 +01:00
Ludovic BOUÉ
7a93eb779c Merge branch 'dev' into PIRUnoccupiedToOccupiedDelay 2026-02-11 16:30:17 +01:00
Ludovic BOUÉ
7d673cd9c4 Update occupancy sensing PIR attributes for detection delay and threshold 2026-02-07 09:49:56 +00:00
Ludovic BOUÉ
44bc11580d Rename occupancy sensor attributes for clarity and update tests 2026-02-07 09:49:14 +00:00
Ludovic BOUÉ
c23795fe14 Rename occupancy sensing keys to include PIR prefix for clarity 2026-02-07 09:45:46 +00:00
Ludovic BOUÉ
bf6f9a011b Rename occupancy sensing translation keys and add new entries for detection delay and threshold 2026-02-07 09:44:30 +00:00
Ludovic BOUÉ
1cdbe596fe Update snapshots 2026-02-06 17:29:30 +00:00
Ludovic BOUÉ
a9d52bfbe7 Remove feature map attribute from occupancy sensing discovery schema 2026-02-06 17:24:51 +00:00
Ludovic BOUÉ
6eed1f9961 Update snapshots 2026-02-06 17:06:27 +00:00
Ludovic BOUÉ
149607ab17 Refactor strings.json: Remove duplicate unoccupied to occupied delay entries and standardize casing for threshold name 2026-02-06 17:04:42 +00:00
Ludovic BOUÉ
279b5be357 Add assertions for min, max, and unit_of_measurement in occupancy sensor tests 2026-02-06 17:02:19 +00:00
Ludovic BOUÉ
82b93e788b Update snapshots 2026-02-06 16:58:16 +00:00
Ludovic BOUÉ
555813f84f Move PIRUnoccupiedToOccupiedDelay before 2026-02-06 16:58:16 +00:00
Ludovic BOUÉ
ecf1b4e591 Fix occupancy sensor threshold test assertion to match updated mock data 2026-02-06 16:58:15 +00:00
Ludovic BOUÉ
e17a9f12a1 Rename occupancy sensor state and entity IDs for clarity in PIR tests 2026-02-06 16:58:15 +00:00
Ludovic BOUÉ
e8f05f5291 Update snapshots 2026-02-06 16:58:15 +00:00
Ludovic BOUÉ
a5a76e9268 Add mock occupancy sensor JSON fixture 2026-02-06 16:58:15 +00:00
Ludovic BOUÉ
edc3fb47b2 Réorganiser les chaînes pour le délai et le seuil de passage de l'état inoccupé à occupé 2026-02-06 16:58:15 +00:00
Ludovic BOUÉ
f1e514a70a Update homeassistant/components/matter/strings.json
Co-authored-by: Norbert Rittel <norbert@rittel.de>
2026-02-06 17:52:35 +01:00
Ludovic BOUÉ
5632baca5b Merge branch 'dev' into PIRUnoccupiedToOccupiedDelay 2026-02-06 17:08:24 +01:00
Ludovic BOUÉ
78f9bad706 PIRUnoccupiedToOccupiedDelay attribute 2026-02-06 16:00:18 +00:00
Ludovic BOUÉ
3fdaaecd0f PIRUnoccupiedToOccupied attributes 2026-02-04 13:01:13 +00:00
32 changed files with 2180 additions and 6 deletions

View File

@@ -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
View File

@@ -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

View 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

View 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

View 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

View 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")

View 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()

View 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"
)

View 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"]
}

View 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

View 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%]"
}
}
}

View File

@@ -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,

View File

@@ -214,6 +214,12 @@
"cook_time": {
"name": "Cooking time"
},
"detection_delay": {
"name": "Detection delay"
},
"detection_threshold": {
"name": "Detection threshold"
},
"hold_time": {
"name": "Hold time"
},

View File

@@ -6,6 +6,7 @@ To update, run python3 -m script.hassfest
APPLICATION_CREDENTIALS = [
"aladdin_connect",
"august",
"dropbox",
"ekeybionyx",
"electric_kiwi",
"fitbit",

View File

@@ -158,6 +158,7 @@ FLOWS = {
"downloader",
"dremel_3d_printer",
"drop_connect",
"dropbox",
"droplet",
"dsmr",
"dsmr_reader",

View File

@@ -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",

View File

@@ -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
View File

@@ -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
View File

@@ -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

View File

@@ -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

View File

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

View 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

View 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

View 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

View 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

View File

@@ -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",

View File

@@ -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": []
}

View File

@@ -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([

View File

@@ -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([

View File

@@ -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([

View File

@@ -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,
)

View File

@@ -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: