Use entity description class for Garages Amsterdam (#131672)

This commit is contained in:
Klaas Schoute 2024-11-27 08:42:19 +01:00 committed by GitHub
parent 8bb0fab732
commit 67ba44c3fa
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 528 additions and 85 deletions

View File

@ -2,19 +2,39 @@
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from odp_amsterdam import Garage
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import GaragesAmsterdamConfigEntry
from .coordinator import GaragesAmsterdamDataUpdateCoordinator
from .entity import GaragesAmsterdamEntity
BINARY_SENSORS = {
"state",
}
@dataclass(frozen=True, kw_only=True)
class GaragesAmsterdamBinarySensorEntityDescription(BinarySensorEntityDescription):
"""Class describing Garages Amsterdam binary sensor entity."""
is_on: Callable[[Garage], bool]
BINARY_SENSORS: tuple[GaragesAmsterdamBinarySensorEntityDescription, ...] = (
GaragesAmsterdamBinarySensorEntityDescription(
key="state",
translation_key="state",
device_class=BinarySensorDeviceClass.PROBLEM,
is_on=lambda garage: garage.state != "ok",
),
)
async def async_setup_entry(
@ -26,20 +46,33 @@ async def async_setup_entry(
coordinator = entry.runtime_data
async_add_entities(
GaragesAmsterdamBinarySensor(coordinator, entry.data["garage_name"], info_type)
for info_type in BINARY_SENSORS
GaragesAmsterdamBinarySensor(
coordinator=coordinator,
garage_name=entry.data["garage_name"],
description=description,
)
for description in BINARY_SENSORS
)
class GaragesAmsterdamBinarySensor(GaragesAmsterdamEntity, BinarySensorEntity):
"""Binary Sensor representing garages amsterdam data."""
_attr_device_class = BinarySensorDeviceClass.PROBLEM
_attr_name = None
entity_description: GaragesAmsterdamBinarySensorEntityDescription
def __init__(
self,
*,
coordinator: GaragesAmsterdamDataUpdateCoordinator,
garage_name: str,
description: GaragesAmsterdamBinarySensorEntityDescription,
) -> None:
"""Initialize garages amsterdam binary sensor."""
super().__init__(coordinator, garage_name)
self.entity_description = description
self._attr_unique_id = f"{garage_name}-{description.key}"
@property
def is_on(self) -> bool:
"""If the binary sensor is currently on or off."""
return (
getattr(self.coordinator.data[self._garage_name], self._info_type) != "ok"
)
return self.entity_description.is_on(self.coordinator.data[self._garage_name])

View File

@ -7,7 +7,7 @@ import logging
from typing import Final
DOMAIN: Final = "garages_amsterdam"
ATTRIBUTION = f'{"Data provided by municipality of Amsterdam"}'
ATTRIBUTION = "Data provided by municipality of Amsterdam"
LOGGER = logging.getLogger(__package__)
SCAN_INTERVAL = timedelta(minutes=10)

View File

@ -19,13 +19,10 @@ class GaragesAmsterdamEntity(CoordinatorEntity[GaragesAmsterdamDataUpdateCoordin
self,
coordinator: GaragesAmsterdamDataUpdateCoordinator,
garage_name: str,
info_type: str,
) -> None:
"""Initialize garages amsterdam entity."""
super().__init__(coordinator)
self._attr_unique_id = f"{garage_name}-{info_type}"
self._garage_name = garage_name
self._info_type = info_type
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, garage_name)},
name=garage_name,

View File

@ -2,20 +2,56 @@
from __future__ import annotations
from homeassistant.components.sensor import SensorEntity
from collections.abc import Callable
from dataclasses import dataclass
from odp_amsterdam import Garage
from homeassistant.components.sensor import (
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from . import GaragesAmsterdamConfigEntry
from .coordinator import GaragesAmsterdamDataUpdateCoordinator
from .entity import GaragesAmsterdamEntity
SENSORS = {
"free_space_short",
"free_space_long",
"short_capacity",
"long_capacity",
}
@dataclass(frozen=True, kw_only=True)
class GaragesAmsterdamSensorEntityDescription(SensorEntityDescription):
"""Class describing Garages Amsterdam sensor entity."""
value_fn: Callable[[Garage], StateType]
SENSORS: tuple[GaragesAmsterdamSensorEntityDescription, ...] = (
GaragesAmsterdamSensorEntityDescription(
key="free_space_short",
translation_key="free_space_short",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda garage: garage.free_space_short,
),
GaragesAmsterdamSensorEntityDescription(
key="free_space_long",
translation_key="free_space_long",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda garage: garage.free_space_long,
),
GaragesAmsterdamSensorEntityDescription(
key="short_capacity",
translation_key="short_capacity",
value_fn=lambda garage: garage.short_capacity,
),
GaragesAmsterdamSensorEntityDescription(
key="long_capacity",
translation_key="long_capacity",
value_fn=lambda garage: garage.long_capacity,
),
)
async def async_setup_entry(
@ -27,26 +63,32 @@ async def async_setup_entry(
coordinator = entry.runtime_data
async_add_entities(
GaragesAmsterdamSensor(coordinator, entry.data["garage_name"], info_type)
for info_type in SENSORS
if getattr(coordinator.data[entry.data["garage_name"]], info_type) is not None
GaragesAmsterdamSensor(
coordinator=coordinator,
garage_name=entry.data["garage_name"],
description=description,
)
for description in SENSORS
if description.value_fn(coordinator.data[entry.data["garage_name"]]) is not None
)
class GaragesAmsterdamSensor(GaragesAmsterdamEntity, SensorEntity):
"""Sensor representing garages amsterdam data."""
_attr_native_unit_of_measurement = "cars"
entity_description: GaragesAmsterdamSensorEntityDescription
def __init__(
self,
*,
coordinator: GaragesAmsterdamDataUpdateCoordinator,
garage_name: str,
info_type: str,
description: GaragesAmsterdamSensorEntityDescription,
) -> None:
"""Initialize garages amsterdam sensor."""
super().__init__(coordinator, garage_name, info_type)
self._attr_translation_key = info_type
super().__init__(coordinator, garage_name)
self.entity_description = description
self._attr_unique_id = f"{garage_name}-{description.key}"
@property
def available(self) -> bool:
@ -56,6 +98,8 @@ class GaragesAmsterdamSensor(GaragesAmsterdamEntity, SensorEntity):
)
@property
def native_value(self) -> str:
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return getattr(self.coordinator.data[self._garage_name], self._info_type)
return self.entity_description.value_fn(
self.coordinator.data[self._garage_name]
)

View File

@ -3,8 +3,13 @@
"config": {
"step": {
"user": {
"title": "Pick a garage to monitor",
"data": { "garage_name": "Garage name" }
"description": "Select a garage from the list",
"data": {
"garage_name": "Garage name"
},
"data_description": {
"garage_name": "The name of the garage you want to monitor."
}
}
},
"abort": {
@ -16,16 +21,25 @@
"entity": {
"sensor": {
"free_space_short": {
"name": "Short parking free space"
"name": "Short parking free space",
"unit_of_measurement": "cars"
},
"free_space_long": {
"name": "Long parking free space"
"name": "Long parking free space",
"unit_of_measurement": "cars"
},
"short_capacity": {
"name": "Short parking capacity"
"name": "Short parking capacity",
"unit_of_measurement": "cars"
},
"long_capacity": {
"name": "Long parking capacity"
"name": "Long parking capacity",
"unit_of_measurement": "cars"
}
},
"binary_sensor": {
"state": {
"name": "State"
}
}
}

View File

@ -1 +1,12 @@
"""Tests for the Garages Amsterdam integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Fixture for setting up the integration."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)

View File

@ -1,7 +1,10 @@
"""Fixtures for Garages Amsterdam integration tests."""
from unittest.mock import Mock, patch
from collections.abc import Generator
from datetime import UTC, datetime
from unittest.mock import AsyncMock, patch
from odp_amsterdam import Garage, GarageCategory, VehicleType
import pytest
from homeassistant.components.garages_amsterdam.const import DOMAIN
@ -9,40 +12,74 @@ from homeassistant.components.garages_amsterdam.const import DOMAIN
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override setup entry."""
with patch(
"homeassistant.components.garages_amsterdam.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_garages_amsterdam() -> Generator[AsyncMock]:
"""Mock garages_amsterdam garages."""
with (
patch(
"homeassistant.components.garages_amsterdam.ODPAmsterdam",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.garages_amsterdam.config_flow.ODPAmsterdam",
new=mock_client,
),
):
client = mock_client.return_value
client.all_garages.return_value = [
Garage(
garage_id="test-id-1",
garage_name="IJDok",
vehicle=VehicleType.CAR,
category=GarageCategory.GARAGE,
state="ok",
free_space_short=100,
free_space_long=10,
short_capacity=120,
long_capacity=60,
availability_pct=50.5,
longitude=1.111111,
latitude=2.222222,
updated_at=datetime(2023, 2, 23, 13, 44, 48, tzinfo=UTC),
),
Garage(
garage_id="test-id-2",
garage_name="Arena",
vehicle=VehicleType.CAR,
category=GarageCategory.GARAGE,
state="error",
free_space_short=200,
free_space_long=None,
short_capacity=240,
long_capacity=None,
availability_pct=83.3,
longitude=3.333333,
latitude=4.444444,
updated_at=datetime(2023, 2, 23, 13, 44, 48, tzinfo=UTC),
),
]
yield client
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="monitor",
domain=DOMAIN,
data={},
data={
"garage_name": "IJDok",
},
unique_id="unique_thingy",
version=1,
)
@pytest.fixture(autouse=True)
def mock_garages_amsterdam():
"""Mock garages_amsterdam garages."""
with patch(
"odp_amsterdam.ODPAmsterdam.all_garages",
return_value=[
Mock(
garage_name="IJDok",
free_space_short=100,
free_space_long=10,
short_capacity=120,
long_capacity=60,
state="ok",
),
Mock(
garage_name="Arena",
free_space_short=200,
free_space_long=20,
short_capacity=240,
long_capacity=80,
state="error",
),
],
) as mock_get_garages:
yield mock_get_garages

View File

@ -0,0 +1,49 @@
# serializer version: 1
# name: test_all_binary_sensors[binary_sensor.ijdok_state-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.ijdok_state',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.PROBLEM: 'problem'>,
'original_icon': None,
'original_name': 'State',
'platform': 'garages_amsterdam',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'state',
'unique_id': 'IJDok-state',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.ijdok_state-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by municipality of Amsterdam',
'device_class': 'problem',
'friendly_name': 'IJDok State',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.ijdok_state',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---

View File

@ -0,0 +1,195 @@
# serializer version: 1
# name: test_all_sensors[sensor.ijdok_long_parking_capacity-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ijdok_long_parking_capacity',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Long parking capacity',
'platform': 'garages_amsterdam',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'long_capacity',
'unique_id': 'IJDok-long_capacity',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.ijdok_long_parking_capacity-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by municipality of Amsterdam',
'friendly_name': 'IJDok Long parking capacity',
}),
'context': <ANY>,
'entity_id': 'sensor.ijdok_long_parking_capacity',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '60',
})
# ---
# name: test_all_sensors[sensor.ijdok_long_parking_free_space-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ijdok_long_parking_free_space',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Long parking free space',
'platform': 'garages_amsterdam',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'free_space_long',
'unique_id': 'IJDok-free_space_long',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.ijdok_long_parking_free_space-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by municipality of Amsterdam',
'friendly_name': 'IJDok Long parking free space',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.ijdok_long_parking_free_space',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '10',
})
# ---
# name: test_all_sensors[sensor.ijdok_short_parking_capacity-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ijdok_short_parking_capacity',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Short parking capacity',
'platform': 'garages_amsterdam',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'short_capacity',
'unique_id': 'IJDok-short_capacity',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.ijdok_short_parking_capacity-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by municipality of Amsterdam',
'friendly_name': 'IJDok Short parking capacity',
}),
'context': <ANY>,
'entity_id': 'sensor.ijdok_short_parking_capacity',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '120',
})
# ---
# name: test_all_sensors[sensor.ijdok_short_parking_free_space-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ijdok_short_parking_free_space',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Short parking free space',
'platform': 'garages_amsterdam',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'free_space_short',
'unique_id': 'IJDok-free_space_short',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.ijdok_short_parking_free_space-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'Data provided by municipality of Amsterdam',
'friendly_name': 'IJDok Short parking free space',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.ijdok_short_parking_free_space',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '100',
})
# ---

View File

@ -0,0 +1,31 @@
"""Tests the binary sensors provided by the Garages Amsterdam integration."""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
from syrupy import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import snapshot_platform
async def test_all_binary_sensors(
hass: HomeAssistant,
mock_garages_amsterdam: AsyncMock,
mock_config_entry: AsyncMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all binary sensors."""
with patch(
"homeassistant.components.garages_amsterdam.PLATFORMS", [Platform.BINARY_SENSOR]
):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)

View File

@ -1,39 +1,40 @@
"""Test the Garages Amsterdam config flow."""
from http import HTTPStatus
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
from aiohttp import ClientResponseError
import pytest
from homeassistant import config_entries
from homeassistant.components.garages_amsterdam.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
async def test_full_user_flow(hass: HomeAssistant) -> None:
"""Test we get the form."""
async def test_full_user_flow(
hass: HomeAssistant,
mock_garages_amsterdam: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test the full user configuration flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": SOURCE_USER}
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "user"
assert not result.get("errors")
with patch(
"homeassistant.components.garages_amsterdam.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"garage_name": "IJDok"},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"garage_name": "IJDok"},
)
assert result2.get("type") is FlowResultType.CREATE_ENTRY
assert result2.get("title") == "IJDok"
assert "result" in result2
assert result2["result"].unique_id == "IJDok"
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "IJDok"
assert result.get("data") == {"garage_name": "IJDok"}
assert len(mock_garages_amsterdam.all_garages.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@ -50,14 +51,14 @@ async def test_full_user_flow(hass: HomeAssistant) -> None:
async def test_error_handling(
side_effect: Exception, reason: str, hass: HomeAssistant
) -> None:
"""Test we get the form."""
"""Test error handling in the config flow."""
with patch(
"homeassistant.components.garages_amsterdam.config_flow.ODPAmsterdam.all_garages",
side_effect=side_effect,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": SOURCE_USER}
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == reason

View File

@ -0,0 +1,31 @@
"""Tests the sensors provided by the Garages Amsterdam integration."""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
from syrupy import SnapshotAssertion
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import snapshot_platform
async def test_all_sensors(
hass: HomeAssistant,
mock_garages_amsterdam: AsyncMock,
mock_config_entry: AsyncMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all sensors."""
with patch(
"homeassistant.components.garages_amsterdam.PLATFORMS", [Platform.SENSOR]
):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)