Electric-Special/ha-core
1
0
Fork 0
mirror of https://github.com/Electric-Special/ha-core.git synced 2026-03-21 02:03:27 +01:00
Code Issues Packages Projects Releases Wiki Activity

Netatmo doortag binary sensor addition (#160608)

Browse Source
This commit is contained in:
Zoltán Farkasdi
2026-02-17 10:34:23 +01:00
committed by GitHub
parent 9b1812858b
commit 307c6a4ce2
8 changed files with 791 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

350
homeassistant/components/netatmo/binary_sensor.py
View File

@@ -1,8 +1,12 @@
"""Support for Netatmo binary sensors."""
from collections.abc import Callable
from dataclasses import dataclass
from functools import partial
import logging
from typing import Final, cast
from typing import Any, Final, cast
from pyatmo.modules.device_types import DeviceCategory as NetatmoDeviceCategory
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@@ -13,34 +17,166 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import NETATMO_CREATE_WEATHER_BINARY_SENSOR
from .data_handler import NetatmoDevice
from .entity import NetatmoWeatherModuleEntity
from .const import (
CONF_URL_SECURITY,
DOORTAG_CATEGORY_DOOR,
DOORTAG_CATEGORY_FURNITURE,
DOORTAG_CATEGORY_GARAGE,
DOORTAG_CATEGORY_GATE,
DOORTAG_CATEGORY_OTHER,
DOORTAG_CATEGORY_WINDOW,
DOORTAG_STATUS_CALIBRATING,
DOORTAG_STATUS_CALIBRATION_FAILED,
DOORTAG_STATUS_CLOSED,
DOORTAG_STATUS_MAINTENANCE,
DOORTAG_STATUS_NO_NEWS,
DOORTAG_STATUS_OPEN,
DOORTAG_STATUS_UNDEFINED,
DOORTAG_STATUS_WEAK_SIGNAL,
NETATMO_CREATE_CONNECTIVITY_BINARY_SENSOR,
NETATMO_CREATE_OPENING_BINARY_SENSOR,
NETATMO_CREATE_WEATHER_BINARY_SENSOR,
)
from .data_handler import SIGNAL_NAME, NetatmoDevice
from .entity import NetatmoModuleEntity, NetatmoWeatherModuleEntity
_LOGGER = logging.getLogger(__name__)
DEFAULT_OPENING_SENSOR_KEY = "opening_sensor"
OPENING_STATUS_TO_BINARY_SENSOR_STATE: Final[dict[str, bool | None]] = {
DOORTAG_STATUS_NO_NEWS: None,
DOORTAG_STATUS_CALIBRATING: None,
DOORTAG_STATUS_UNDEFINED: None,
DOORTAG_STATUS_CLOSED: False,
DOORTAG_STATUS_OPEN: True,
DOORTAG_STATUS_CALIBRATION_FAILED: None,
DOORTAG_STATUS_MAINTENANCE: None,
DOORTAG_STATUS_WEAK_SIGNAL: None,
}
OPENING_CATEGORY_TO_DEVICE_CLASS: Final[dict[str | None, BinarySensorDeviceClass]] = {
DOORTAG_CATEGORY_DOOR: BinarySensorDeviceClass.DOOR,
DOORTAG_CATEGORY_FURNITURE: BinarySensorDeviceClass.OPENING,
DOORTAG_CATEGORY_GARAGE: BinarySensorDeviceClass.GARAGE_DOOR,
DOORTAG_CATEGORY_GATE: BinarySensorDeviceClass.OPENING,
DOORTAG_CATEGORY_OTHER: BinarySensorDeviceClass.OPENING,
DOORTAG_CATEGORY_WINDOW: BinarySensorDeviceClass.WINDOW,
}
def get_opening_category(netatmo_device: NetatmoDevice) -> str:
"""Helper function to get opening category from Netatmo API raw data."""
# Iterate through each home in the raw data.
for home in netatmo_device.data_handler.account.raw_data["homes"]:
# Check if the modules list exists for the current home.
if "modules" in home:
# Iterate through each module to find a matching ID.
for module in home["modules"]:
if module["id"] == netatmo_device.device.entity_id:
# We found the matching device. Get its category.
if module.get("category") is not None:
return cast(str, module["category"])
raise ValueError(
f"Device {netatmo_device.device.entity_id} found, "
"but 'category' is missing in raw data."
)
raise ValueError(
f"Device {netatmo_device.device.entity_id} not found in Netatmo raw data."
)
OPENING_CATEGORY_TO_KEY: Final[dict[str, str | None]] = {
DOORTAG_CATEGORY_DOOR: None,
DOORTAG_CATEGORY_FURNITURE: DOORTAG_CATEGORY_FURNITURE,
DOORTAG_CATEGORY_GARAGE: None,
DOORTAG_CATEGORY_GATE: DOORTAG_CATEGORY_GATE,
DOORTAG_CATEGORY_OTHER: DEFAULT_OPENING_SENSOR_KEY,
DOORTAG_CATEGORY_WINDOW: None,
}
@dataclass(frozen=True, kw_only=True)
class NetatmoBinarySensorEntityDescription(BinarySensorEntityDescription):
"""Describes Netatmo binary sensor entity."""
name: str | None = None # The default name of the sensor
netatmo_name: str # The name used by Netatmo API for this sensor
netatmo_name: str | None = (
None # The name used by Netatmo API for this sensor (exposed feature as attribute) if different than key
)
value_fn: Callable[[str], str | bool | None] = lambda x: x
NETATMO_WEATHER_BINARY_SENSOR_DESCRIPTIONS: Final[
NETATMO_CONNECTIVITY_BINARY_SENSOR_DESCRIPTIONS: Final[
list[NetatmoBinarySensorEntityDescription]
] = [
NetatmoBinarySensorEntityDescription(
key="reachable",
name="Connectivity",
netatmo_name="reachable",
device_class=BinarySensorDeviceClass.CONNECTIVITY,
),
]
# Assuming a Module object with the following attributes:
# {'battery_level': 5780,
# 'battery_percent': None,
# 'battery_state': 'full',
# 'bridge': 'XX:XX:XX:XX:XX:XX',
# 'device_category': <DeviceCategory.opening: 'opening'>,
# 'device_type': <DeviceType.NACamDoorTag: 'NACamDoorTag'>,
# 'entity_id': 'NN:NN:NN:NN:NN:NN',
# 'features': {'status', 'battery', 'rf_strength', 'reachable'},
# 'firmware_name': None,
# 'firmware_revision': 58,
# 'history_features': set(),
# 'history_features_values': {},
# 'home': <pyatmo.home.Home object at 0x790e5c3ea660>,
# 'modules': None,
# 'name': 'YYYYYY',
# 'reachable': True,
# 'rf_strength': 74,
# 'room_id': 'ZZZZZZZZ',
# 'status': 'open'}
NETATMO_OPENING_BINARY_SENSOR_DESCRIPTIONS: Final[
list[NetatmoBinarySensorEntityDescription]
] = [
NetatmoBinarySensorEntityDescription(
key="opening",
netatmo_name="status",
value_fn=OPENING_STATUS_TO_BINARY_SENSOR_STATE.get,
),
]
DEVICE_CATEGORY_BINARY_URLS: Final[dict[NetatmoDeviceCategory, str]] = {
NetatmoDeviceCategory.opening: CONF_URL_SECURITY,
}
DEVICE_CATEGORY_WEATHER_BINARY_SENSORS: Final[
dict[NetatmoDeviceCategory, list[NetatmoBinarySensorEntityDescription]]
] = {
NetatmoDeviceCategory.air_care: NETATMO_CONNECTIVITY_BINARY_SENSOR_DESCRIPTIONS,
NetatmoDeviceCategory.weather: NETATMO_CONNECTIVITY_BINARY_SENSOR_DESCRIPTIONS,
}
DEVICE_CATEGORY_CONNECTIVITY_BINARY_SENSORS: Final[
dict[NetatmoDeviceCategory, list[NetatmoBinarySensorEntityDescription]]
] = {
NetatmoDeviceCategory.opening: NETATMO_CONNECTIVITY_BINARY_SENSOR_DESCRIPTIONS,
}
DEVICE_CATEGORY_OPENING_BINARY_SENSORS: Final[
dict[NetatmoDeviceCategory, list[NetatmoBinarySensorEntityDescription]]
] = {
NetatmoDeviceCategory.opening: NETATMO_OPENING_BINARY_SENSOR_DESCRIPTIONS,
}
DEVICE_CATEGORY_BINARY_PUBLISHERS: Final[list[NetatmoDeviceCategory]] = [
NetatmoDeviceCategory.opening,
]
async def async_setup_entry(
hass: HomeAssistant,
@@ -50,25 +186,47 @@ async def async_setup_entry(
"""Set up Netatmo weather binary sensors based on a config entry."""
@callback
def _create_weather_binary_sensor_entity(netatmo_device: NetatmoDevice) -> None:
"""Create weather binary sensor entities for a Netatmo weather device."""
def _create_binary_sensor_entity(
binarySensorClass: type[
NetatmoWeatherBinarySensor
| NetatmoOpeningBinarySensor
| NetatmoConnectivityBinarySensor
],
descriptions: dict[
NetatmoDeviceCategory, list[NetatmoBinarySensorEntityDescription]
],
netatmo_device: NetatmoDevice,
) -> None:
"""Create binary sensor entities for a Netatmo device."""
descriptions_to_add = NETATMO_WEATHER_BINARY_SENSOR_DESCRIPTIONS
if netatmo_device.device.device_category is None:
return
entities: list[NetatmoWeatherBinarySensor] = []
descriptions_to_add = descriptions.get(
netatmo_device.device.device_category, []
)
entities: list[
NetatmoWeatherBinarySensor
| NetatmoOpeningBinarySensor
| NetatmoConnectivityBinarySensor
] = []
# Create binary sensors for module
for description in descriptions_to_add:
# Actual check is simple for reachable
feature_check = description.key
if description.netatmo_name is None:
feature_check = description.key
else:
feature_check = description.netatmo_name
if feature_check in netatmo_device.device.features:
_LOGGER.debug(
'Adding "%s" weather binary sensor for device %s',
'Adding "%s" (native: "%s") binary sensor for device %s',
description.key,
feature_check,
netatmo_device.device.name,
)
entities.append(
NetatmoWeatherBinarySensor(
binarySensorClass(
netatmo_device,
description,
)
@@ -81,12 +239,100 @@ async def async_setup_entry(
async_dispatcher_connect(
hass,
NETATMO_CREATE_WEATHER_BINARY_SENSOR,
_create_weather_binary_sensor_entity,
partial(
_create_binary_sensor_entity,
NetatmoWeatherBinarySensor,
DEVICE_CATEGORY_WEATHER_BINARY_SENSORS,
),
)
)
entry.async_on_unload(
async_dispatcher_connect(
hass,
NETATMO_CREATE_OPENING_BINARY_SENSOR,
partial(
_create_binary_sensor_entity,
NetatmoOpeningBinarySensor,
DEVICE_CATEGORY_OPENING_BINARY_SENSORS,
),
)
)
entry.async_on_unload(
async_dispatcher_connect(
hass,
NETATMO_CREATE_CONNECTIVITY_BINARY_SENSOR,
partial(
_create_binary_sensor_entity,
NetatmoConnectivityBinarySensor,
DEVICE_CATEGORY_CONNECTIVITY_BINARY_SENSORS,
),
)
)
class NetatmoWeatherBinarySensor(NetatmoWeatherModuleEntity, BinarySensorEntity):
class NetatmoBinarySensor(NetatmoModuleEntity, BinarySensorEntity):
"""Implementation of a Netatmo binary sensor."""
entity_description: NetatmoBinarySensorEntityDescription
_attr_has_entity_name = True
def __init__(
self,
netatmo_device: NetatmoDevice,
description: NetatmoBinarySensorEntityDescription,
**kwargs: Any, # Add this to capture extra args from super()
) -> None:
"""Initialize a Netatmo binary sensor."""
# To prevent exception about missing URL we need to set it explicitly
if netatmo_device.device.device_category is not None:
if (
DEVICE_CATEGORY_BINARY_URLS.get(netatmo_device.device.device_category)
is not None
):
self._attr_configuration_url = DEVICE_CATEGORY_BINARY_URLS[
netatmo_device.device.device_category
]
super().__init__(netatmo_device, **kwargs)
self.entity_description = description
self._attr_unique_id = f"{self.device.entity_id}-{description.key}"
# Register publishers for the entity if needed (not already done in parent class - weather and air_care)
# We need to keep this here because we have two classes depending on it and we want to avoid adding publishers for all binary sensors
if self.device.device_category in DEVICE_CATEGORY_BINARY_PUBLISHERS:
self._publishers.extend(
[
{
"name": self.home.entity_id,
"home_id": self.home.entity_id,
SIGNAL_NAME: netatmo_device.signal_name,
},
]
)
@callback
def async_update_callback(self) -> None:
"""Update the entity's state."""
# Should be the connectivity (reachable) sensor only here as we have update for opening in its class
# Setting reachable sensor, so we just get it directly (backward compatibility to weather binary sensor)
value = getattr(self.device, self.entity_description.key, None)
if value is None:
self._attr_available = False
self._attr_is_on = False
else:
self._attr_available = True
self._attr_is_on = cast(bool, value)
self.async_write_ha_state()
class NetatmoWeatherBinarySensor(NetatmoWeatherModuleEntity, NetatmoBinarySensor):
"""Implementation of a Netatmo weather binary sensor."""
entity_description: NetatmoBinarySensorEntityDescription
@@ -98,24 +344,68 @@ class NetatmoWeatherBinarySensor(NetatmoWeatherModuleEntity, BinarySensorEntity)
) -> None:
"""Initialize a Netatmo weather binary sensor."""
super().__init__(netatmo_device)
super().__init__(netatmo_device, description=description)
self.entity_description = description
self._attr_unique_id = f"{self.device.entity_id}-{description.key}"
class NetatmoOpeningBinarySensor(NetatmoBinarySensor):
"""Implementation of a Netatmo opening binary sensor."""
entity_description: NetatmoBinarySensorEntityDescription
_attr_has_entity_name = True
def __init__(
self,
netatmo_device: NetatmoDevice,
description: NetatmoBinarySensorEntityDescription,
) -> None:
"""Initialize a Netatmo binary sensor."""
super().__init__(netatmo_device, description)
# Apply Dynamic Device Class override
self._attr_device_class = OPENING_CATEGORY_TO_DEVICE_CLASS.get(
get_opening_category(netatmo_device), BinarySensorDeviceClass.OPENING
)
# Apply Dynamic Translation Key override if needed
translation_key = OPENING_CATEGORY_TO_KEY.get(
get_opening_category(netatmo_device), DEFAULT_OPENING_SENSOR_KEY
)
if translation_key is not None:
self._attr_translation_key = translation_key
@callback
def async_update_callback(self) -> None:
"""Update the entity's state."""
value: StateType | None = None
value = getattr(self.device, self.entity_description.netatmo_name, None)
if value is None:
if not self.device.reachable:
# If reachable is None or False we set availability to False
self._attr_available = False
self._attr_is_on = False
self._attr_is_on = None
else:
# If reachable is True, we get the actual value
if self.entity_description.netatmo_name is None:
raw_value = getattr(self.device, self.entity_description.key, None)
else:
raw_value = getattr(
self.device, self.entity_description.netatmo_name, None
)
if raw_value is not None:
value = self.entity_description.value_fn(raw_value)
else:
value = None
# Set sensor state
self._attr_available = True
self._attr_is_on = cast(bool, value)
self._attr_is_on = cast(bool, value) if value is not None else None
self.async_write_ha_state()
class NetatmoConnectivityBinarySensor(NetatmoBinarySensor):
"""Implementation of a Netatmo connectivity binary sensor."""
entity_description: NetatmoBinarySensorEntityDescription
_attr_has_entity_name = True

19
homeassistant/components/netatmo/const.py
View File

@@ -46,9 +46,11 @@ NETATMO_CREATE_CAMERA = "netatmo_create_camera"
NETATMO_CREATE_CAMERA_LIGHT = "netatmo_create_camera_light"
NETATMO_CREATE_CLIMATE = "netatmo_create_climate"
NETATMO_CREATE_COVER = "netatmo_create_cover"
NETATMO_CREATE_CONNECTIVITY_BINARY_SENSOR = "netatmo_create_connectivity_binary_sensor"
NETATMO_CREATE_BUTTON = "netatmo_create_button"
NETATMO_CREATE_FAN = "netatmo_create_fan"
NETATMO_CREATE_LIGHT = "netatmo_create_light"
NETATMO_CREATE_OPENING_BINARY_SENSOR = "netatmo_create_opening_binary_sensor"
NETATMO_CREATE_ROOM_SENSOR = "netatmo_create_room_sensor"
NETATMO_CREATE_SELECT = "netatmo_create_select"
NETATMO_CREATE_SENSOR = "netatmo_create_sensor"
@@ -191,6 +193,23 @@ MODE_LIGHT_OFF = "off"
MODE_LIGHT_ON = "on"
CAMERA_LIGHT_MODES = [MODE_LIGHT_ON, MODE_LIGHT_OFF, MODE_LIGHT_AUTO]
# Door tag categories
DOORTAG_CATEGORY_DOOR = "door"
DOORTAG_CATEGORY_FURNITURE = "furniture"
DOORTAG_CATEGORY_GARAGE = "garage"
DOORTAG_CATEGORY_GATE = "gate"
DOORTAG_CATEGORY_OTHER = "other"
DOORTAG_CATEGORY_WINDOW = "window"
# Door tag statuses
DOORTAG_STATUS_CALIBRATING = "calibrating"
DOORTAG_STATUS_CALIBRATION_FAILED = "calibration_failed"
DOORTAG_STATUS_CLOSED = "closed"
DOORTAG_STATUS_MAINTENANCE = "maintenance"
DOORTAG_STATUS_NO_NEWS = "no_news"
DOORTAG_STATUS_OPEN = "open"
DOORTAG_STATUS_UNDEFINED = "undefined"
DOORTAG_STATUS_WEAK_SIGNAL = "weak_signal"
# Webhook push_types MUST follow exactly Netatmo's naming on products!
# See https://dev.netatmo.com/apidocumentation
# e.g. cameras: NACamera, NOC, etc.

6
homeassistant/components/netatmo/data_handler.py
View File

@@ -38,9 +38,11 @@ from .const import (
NETATMO_CREATE_CAMERA,
NETATMO_CREATE_CAMERA_LIGHT,
NETATMO_CREATE_CLIMATE,
NETATMO_CREATE_CONNECTIVITY_BINARY_SENSOR,
NETATMO_CREATE_COVER,
NETATMO_CREATE_FAN,
NETATMO_CREATE_LIGHT,
NETATMO_CREATE_OPENING_BINARY_SENSOR,
NETATMO_CREATE_ROOM_SENSOR,
NETATMO_CREATE_SELECT,
NETATMO_CREATE_SENSOR,
@@ -367,6 +369,10 @@ class NetatmoDataHandler:
],
NetatmoDeviceCategory.meter: [NETATMO_CREATE_SENSOR],
NetatmoDeviceCategory.fan: [NETATMO_CREATE_FAN],
NetatmoDeviceCategory.opening: [
NETATMO_CREATE_CONNECTIVITY_BINARY_SENSOR,
NETATMO_CREATE_OPENING_BINARY_SENSOR,
],
}
for module in home.modules.values():
if not module.device_category:

14
homeassistant/components/netatmo/entity.py
View File

@@ -93,9 +93,11 @@ class NetatmoBaseEntity(Entity):
class NetatmoDeviceEntity(NetatmoBaseEntity):
"""Netatmo entity base class."""
def __init__(self, data_handler: NetatmoDataHandler, device: NetatmoBase) -> None:
def __init__(
self, data_handler: NetatmoDataHandler, device: NetatmoBase, **kwargs: Any
) -> None:
"""Set up Netatmo entity base."""
super().__init__(data_handler)
super().__init__(data_handler, **kwargs)
self.device = device
@property
@@ -153,9 +155,9 @@ class NetatmoModuleEntity(NetatmoDeviceEntity):
device: Module
_attr_configuration_url: str
def __init__(self, device: NetatmoDevice) -> None:
def __init__(self, device: NetatmoDevice, **kwargs: Any) -> None:
"""Set up a Netatmo module entity."""
super().__init__(device.data_handler, device.device)
super().__init__(device.data_handler, device.device, **kwargs)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device.device.entity_id)},
name=device.device.name,
@@ -175,9 +177,9 @@ class NetatmoWeatherModuleEntity(NetatmoModuleEntity):
_attr_configuration_url = CONF_URL_WEATHER
def __init__(self, device: NetatmoDevice) -> None:
def __init__(self, device: NetatmoDevice, **kwargs: Any) -> None:
"""Set up a Netatmo weather module entity."""
super().__init__(device)
super().__init__(device, **kwargs)
assert self.device.device_category
category = self.device.device_category.name
self._publishers.extend(

11
homeassistant/components/netatmo/strings.json
View File

@@ -54,6 +54,17 @@
}
},
"entity": {
"binary_sensor": {
"furniture": {
"name": "Furniture"
},
"gate": {
"name": "Gate"
},
"opening_sensor": {
"name": "Opening"
}
},
"button": {
"preferred_position": {
"name": "Preferred position"

4
tests/components/netatmo/common.py
View File

@@ -83,6 +83,10 @@ async def fake_post_request(hass: HomeAssistant, *args: Any, **kwargs: Any):
else:
payload = json.loads(await async_load_fixture(hass, f"{endpoint}.json", DOMAIN))
# Apply test-specific modifications to the payload
if "msg_callback" in kwargs:
kwargs["msg_callback"](payload)
return AiohttpClientMockResponse(
method="POST",
url=kwargs["endpoint"],

102
tests/components/netatmo/snapshots/test_binary_sensor.ambr
View File

@@ -572,3 +572,105 @@
'state': 'on',
})
# ---
# name: test_entity[binary_sensor.window_hall_connectivity-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.window_hall_connectivity',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Connectivity',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.CONNECTIVITY: 'connectivity'>,
'original_icon': None,
'original_name': 'Connectivity',
'platform': 'netatmo',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '12:34:56:00:86:99-reachable',
'unit_of_measurement': None,
})
# ---
# name: test_entity[binary_sensor.window_hall_connectivity-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Netatmo',
'device_class': 'connectivity',
'friendly_name': 'Window Hall Connectivity',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.window_hall_connectivity',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_entity[binary_sensor.window_hall_window-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.window_hall_window',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Window',
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.WINDOW: 'window'>,
'original_icon': None,
'original_name': 'Window',
'platform': 'netatmo',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '12:34:56:00:86:99-opening',
'unit_of_measurement': None,
})
# ---
# name: test_entity[binary_sensor.window_hall_window-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by Netatmo',
'device_class': 'window',
'friendly_name': 'Window Hall Window',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.window_hall_window',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unavailable',
})
# ---

325
tests/components/netatmo/test_binary_sensor.py
View File

@@ -1,17 +1,27 @@
"""Support for Netatmo binary sensors."""
from unittest.mock import AsyncMock
from datetime import timedelta
from typing import Any
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.const import CONF_WEBHOOK_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
import homeassistant.util.dt as dt_util
from .common import snapshot_platform_entities
from .common import (
FAKE_WEBHOOK_ACTIVATION,
fake_post_request,
simulate_webhook,
snapshot_platform_entities,
)
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
@@ -30,3 +40,310 @@ async def test_entity(
entity_registry,
snapshot,
)
async def test_doortag_setup(
hass: HomeAssistant, config_entry: MockConfigEntry, netatmo_auth: AsyncMock
) -> None:
"""Test doortag setup."""
fake_post_hits = 0
async def fake_post(*args: Any, **kwargs: Any):
"""Fake error during requesting backend data."""
nonlocal fake_post_hits
fake_post_hits += 1
return await fake_post_request(hass, *args, **kwargs)
with (
patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth,
patch(
"homeassistant.components.netatmo.data_handler.PLATFORMS",
["camera", "binary_sensor"],
),
patch(
"homeassistant.components.netatmo.async_get_config_entry_implementation",
return_value=AsyncMock(),
),
patch(
"homeassistant.components.netatmo.webhook_generate_url",
) as mock_webhook,
):
mock_auth.return_value.async_post_api_request.side_effect = fake_post
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
mock_webhook.return_value = "https://example.com"
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
# Fake webhook activation
await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK_ACTIVATION)
await hass.async_block_till_done()
# Define variables for the test
_doortag_entity = "window_hall"
_doortag_entity_opening = f"binary_sensor.{_doortag_entity}_window"
_doortag_entity_connectivity = f"binary_sensor.{_doortag_entity}_connectivity"
# Check opening creation
assert hass.states.get(_doortag_entity_opening) is not None
# Check connectivity creation
assert hass.states.get(_doortag_entity_connectivity) is not None
# Check opening initial state
assert hass.states.get(_doortag_entity_opening).state == "unavailable"
# Check connectivity initial state
assert hass.states.get(_doortag_entity_connectivity).state == "off"
@pytest.mark.parametrize(
("doortag_status", "expected"),
[
("no_news", "unknown"),
("calibrating", "unknown"),
("undefined", "unknown"),
("closed", "off"),
("open", "on"),
("calibration_failed", "unknown"),
("maintenance", "unknown"),
("weak_signal", "unknown"),
("invalid_value", "unknown"),
],
)
async def test_doortag_opening_status_change(
hass: HomeAssistant,
config_entry: MockConfigEntry,
netatmo_auth: AsyncMock,
freezer: FrozenDateTimeFactory,
doortag_status: str,
expected: str,
) -> None:
"""Test doortag opening status changes."""
fake_post_hits = 0
# Repeatedly used variables for the test and initial value from fixture
# Use nonexistent ID to prevent matching during initial setup
doortag_entity_id = "aa:bb:cc:dd:ee:ff"
doortag_connectivity = False
doortag_opening = "no_news"
doortag_timestamp = None
def tag_modifier(payload):
"""This function will be called by common.py during ANY homestatus call."""
nonlocal doortag_connectivity, doortag_opening, doortag_timestamp
if doortag_timestamp is not None:
payload["time_server"] = doortag_timestamp
body = payload.get("body", {})
# Handle both structures: {"home": {...}} AND {"homes": [{...}]}
homes_to_check = []
if "home" in body and isinstance(body["home"], dict):
homes_to_check.append(body["home"])
elif "homes" in body and isinstance(body["homes"], list):
homes_to_check.extend(body["homes"])
for home_data in homes_to_check:
# Safety check: ensure home_data is actually a dictionary
if not isinstance(home_data, dict):
continue
modules = home_data.get("modules", [])
for module in modules:
if isinstance(module, dict) and module.get("id") == doortag_entity_id:
module["reachable"] = doortag_connectivity
module["status"] = doortag_opening
if doortag_timestamp is not None:
module["last_seen"] = doortag_timestamp
break
async def fake_tag_post(*args, **kwargs):
"""Fake tag status during requesting backend data."""
nonlocal fake_post_hits
fake_post_hits += 1
return await fake_post_request(hass, *args, msg_callback=tag_modifier, **kwargs)
with (
patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth,
patch(
"homeassistant.components.netatmo.data_handler.PLATFORMS",
["camera", "binary_sensor"],
),
patch(
"homeassistant.components.netatmo.async_get_config_entry_implementation",
return_value=AsyncMock(),
),
patch(
"homeassistant.components.netatmo.webhook_generate_url",
) as mock_webhook,
):
mock_auth.return_value.async_post_api_request.side_effect = fake_tag_post
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
mock_webhook.return_value = "https://example.com"
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
# Fake webhook activation
await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK_ACTIVATION)
await hass.async_block_till_done()
# Define the variables for the test
_doortag_entity = "window_hall"
_doortag_entity_opening = f"binary_sensor.{_doortag_entity}_window"
_doortag_entity_connectivity = f"binary_sensor.{_doortag_entity}_connectivity"
# Check connectivity creation
assert hass.states.get(_doortag_entity_connectivity) is not None
# Check opening creation
assert hass.states.get(_doortag_entity_opening) is not None
# Check connectivity initial state
assert hass.states.get(_doortag_entity_connectivity).state == "off"
# Check opening initial state
assert hass.states.get(_doortag_entity_opening).state == "unavailable"
# Trigger some polling cycle to let API throttling work
for _ in range(11):
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
await hass.async_block_till_done()
await hass.async_block_till_done()
await hass.async_block_till_done()
# Change mocked status
doortag_entity_id = "12:34:56:00:86:99"
doortag_connectivity = True
doortag_opening = doortag_status
doortag_timestamp = int(dt_util.utcnow().timestamp())
# Trigger some polling cycle to let status change be picked up
for _ in range(11):
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
await hass.async_block_till_done()
await hass.async_block_till_done()
await hass.async_block_till_done()
# Check connectivity mocked state
assert hass.states.get(_doortag_entity_connectivity).state == "on"
# Check opening mocked state
assert hass.states.get(_doortag_entity_opening).state == expected
@pytest.mark.parametrize(
("doortag_category", "expected_key", "expected_class"),
[
("door", "door", BinarySensorDeviceClass.DOOR),
("furniture", "furniture", BinarySensorDeviceClass.OPENING),
("garage", "garage_door", BinarySensorDeviceClass.GARAGE_DOOR),
("gate", "gate", BinarySensorDeviceClass.OPENING),
("other", "opening", BinarySensorDeviceClass.OPENING),
("window", "window", BinarySensorDeviceClass.WINDOW),
("invalid_value", "opening", BinarySensorDeviceClass.OPENING),
],
)
async def test_doortag_opening_category(
hass: HomeAssistant,
config_entry: MockConfigEntry,
netatmo_auth: AsyncMock,
doortag_category: str,
expected_key: str,
expected_class: BinarySensorDeviceClass,
) -> None:
"""Test doortag opening status changes."""
fake_post_hits = 0
# Repeatedly used variables for the test and initial value from fixture
doortag_entity_id = "12:34:56:00:86:99"
doortag_connectivity = False
doortag_opening = "no_news"
def tag_modifier(payload):
"""This function will be called by common.py during ANY homestatus call."""
nonlocal doortag_connectivity, doortag_opening
payload["time_server"] = int(dt_util.utcnow().timestamp())
body = payload.get("body", {})
# Handle both structures: {"home": {...}} AND {"homes": [{...}]}
homes_to_check = []
if "home" in body and isinstance(body["home"], dict):
homes_to_check.append(body["home"])
elif "homes" in body and isinstance(body["homes"], list):
homes_to_check.extend(body["homes"])
for home_data in homes_to_check:
# Safety check: ensure home_data is actually a dictionary
if not isinstance(home_data, dict):
continue
modules = home_data.get("modules", [])
for module in modules:
if isinstance(module, dict) and module.get("id") == doortag_entity_id:
module["category"] = doortag_category
break
async def fake_tag_post(*args, **kwargs):
"""Fake tag status during requesting backend data."""
nonlocal fake_post_hits
fake_post_hits += 1
return await fake_post_request(hass, *args, msg_callback=tag_modifier, **kwargs)
with (
patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth"
) as mock_auth,
patch(
"homeassistant.components.netatmo.data_handler.PLATFORMS",
["camera", "binary_sensor"],
),
patch(
"homeassistant.components.netatmo.async_get_config_entry_implementation",
return_value=AsyncMock(),
),
patch(
"homeassistant.components.netatmo.webhook_generate_url",
) as mock_webhook,
):
mock_auth.return_value.async_post_api_request.side_effect = fake_tag_post
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
mock_webhook.return_value = "https://example.com"
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
webhook_id = config_entry.data[CONF_WEBHOOK_ID]
# Fake webhook activation
await simulate_webhook(hass, webhook_id, FAKE_WEBHOOK_ACTIVATION)
await hass.async_block_till_done()
# Define the variables for the test
_doortag_entity = "window_hall"
_doortag_entity_opening = f"binary_sensor.{_doortag_entity}_{expected_key}"
# Check opening creation with right key
assert hass.states.get(_doortag_entity_opening) is not None
# Check opening device class
assert (
hass.states.get(_doortag_entity_opening).attributes.get("device_class")
== expected_class.value
)
# Check opening device name
assert (
hass.states.get(_doortag_entity_opening).attributes.get("friendly_name")
== _doortag_entity.replace("_", " ").title()
+ " "
+ expected_key.replace("_", " ").capitalize()
)