starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-28 10:16:02 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add firmware update platform to WattWächter Plus (#180203)

Browse Source
This commit is contained in:
smartcircuits
2026-08-26 17:44:14 +02:00
committed by GitHub
parent 9feefc0ebe
commit 2fcb86ffc8
8 changed files with 379 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/wattwaechter/__init__.py
+1 -1
View File
@@ -10,7 +10,7 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
from .coordinator import WattwaechterConfigEntry, WattwaechterCoordinator
PLATFORMS = [Platform.SENSOR]
PLATFORMS = [Platform.SENSOR, Platform.UPDATE]
async def async_setup_entry(
homeassistant/components/wattwaechter/coordinator.py
+12 -5
View File
@@ -12,7 +12,7 @@ from aio_wattwaechter import (
WattwaechterError,
WattwaechterNoDataError,
)
from aio_wattwaechter.models import MeterData, SystemInfo
from aio_wattwaechter.models import MeterData, OtaData, SystemInfo
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DEVICE_ID, CONF_HOST, CONF_MAC, CONF_MODEL
@@ -33,6 +33,7 @@ class WattwaechterData:
meter: MeterData
system: SystemInfo | None
ota: OtaData | None
class WattwaechterCoordinator(DataUpdateCoordinator[WattwaechterData]):
@@ -93,13 +94,19 @@ class WattwaechterCoordinator(DataUpdateCoordinator[WattwaechterData]):
translation_placeholders={"host": self.host},
)
# System info is fetched best-effort: a failure here must not take the
# meter sensors unavailable, so the diagnostic sensors just report
# unknown until the next successful poll.
# System info and OTA status are fetched best-effort: a failure here
# must not take the meter sensors unavailable, so their entities just
# report unknown until the next successful poll.
system: SystemInfo | None
try:
system = await self.client.system_info()
except WattwaechterError:
system = None
return WattwaechterData(meter=data, system=system)
ota: OtaData | None
try:
ota = (await self.client.ota_check()).data
except WattwaechterError:
ota = None
return WattwaechterData(meter=data, system=system, ota=ota)
homeassistant/components/wattwaechter/strings.json
+3
View File
@@ -85,6 +85,9 @@
"no_meter_data": {
"message": "No meter data available from {host} yet"
},
"ota_failed": {
"message": "Failed to start the firmware update."
},
"update_failed": {
"message": "Failed to fetch data: {error}"
}
homeassistant/components/wattwaechter/update.py
+87
View File
@@ -0,0 +1,87 @@
"""Update platform for the WattWächter Plus integration."""
from typing import Any, override
from aio_wattwaechter import WattwaechterError
from homeassistant.components.update import (
UpdateDeviceClass,
UpdateEntity,
UpdateEntityFeature,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import WattwaechterConfigEntry, WattwaechterCoordinator
from .entity import WattwaechterEntity
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
entry: WattwaechterConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the WattWächter firmware update entity."""
async_add_entities([WattwaechterUpdateEntity(entry.runtime_data)])
class WattwaechterUpdateEntity(WattwaechterEntity, UpdateEntity):
"""Firmware update entity for WattWächter Plus."""
_attr_device_class = UpdateDeviceClass.FIRMWARE
_attr_supported_features = (
UpdateEntityFeature.INSTALL | UpdateEntityFeature.RELEASE_NOTES
)
def __init__(self, coordinator: WattwaechterCoordinator) -> None:
"""Initialize the update entity."""
super().__init__(coordinator)
self._attr_unique_id = coordinator.device_id
@property
@override
def installed_version(self) -> str | None:
"""Return the currently installed firmware version."""
system = self.coordinator.data.system
if system is not None and (version := system.get_value("esp", "os_version")):
return version
return self.coordinator.fw_version
@property
@override
def latest_version(self) -> str | None:
"""Return the latest available firmware version."""
ota = self.coordinator.data.ota
if ota is None:
return None
if ota.update_available:
return ota.version
return self.installed_version
@override
def release_notes(self) -> str | None:
"""Return the release notes for the available firmware."""
ota = self.coordinator.data.ota
if ota is None:
return None
return ota.release_note_en or None
@override
async def async_install(
self, version: str | None, backup: bool, **kwargs: Any
) -> None:
"""Install the available firmware update."""
try:
success = await self.coordinator.client.ota_start()
except WattwaechterError as err:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="ota_failed"
) from err
if not success:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="ota_failed"
)
tests/components/wattwaechter/conftest.py
+24 -1
View File
@@ -3,7 +3,13 @@
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
from aio_wattwaechter.models import AliveResponse, _parse_meter_data, _parse_system_info
from aio_wattwaechter.models import (
AliveResponse,
OtaCheckResponse,
OtaData,
_parse_meter_data,
_parse_system_info,
)
import pytest
from homeassistant.components.wattwaechter.const import CONF_FW_VERSION, DOMAIN
@@ -49,6 +55,21 @@ MOCK_METER_DATA_MINIMAL = _parse_meter_data(
load_json_object_fixture("meter_data_minimal.json", DOMAIN)
)
MOCK_OTA_CHECK = OtaCheckResponse(
ok=True,
data=OtaData(
update_available=True,
version="1.3.0",
tag="v1.3.0",
release_date="2026-07-01",
release_note_de="Fehlerbehebungen und Verbesserungen",
release_note_en="Bug fixes and improvements",
last_checked=1720000000,
url="https://example.com/firmware/1.3.0.bin",
md5="0123456789abcdef",
),
)
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
@@ -78,6 +99,8 @@ def mock_client() -> Generator[AsyncMock]:
client.system_info = AsyncMock(return_value=MOCK_SYSTEM_INFO)
client.settings = AsyncMock(return_value=MOCK_SETTINGS)
client.meter_data = AsyncMock(return_value=MOCK_METER_DATA)
client.ota_check = AsyncMock(return_value=MOCK_OTA_CHECK)
client.ota_start = AsyncMock(return_value=True)
yield client
tests/components/wattwaechter/snapshots/test_update.ambr
+64
View File
@@ -0,0 +1,64 @@
# serializer version: 1
# name: test_all_entities[update.haushalt_test_firmware-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': 'update',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'update.haushalt_test_firmware',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Firmware',
'options': dict({
}),
'original_device_class': <UpdateDeviceClass.FIRMWARE: 'firmware'>,
'original_icon': None,
'original_name': 'Firmware',
'platform': 'wattwaechter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <UpdateEntityFeature: 17>,
'translation_key': None,
'unique_id': 'ABC123',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[update.haushalt_test_firmware-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<UpdateEntityStateAttribute.AUTO_UPDATE: 'auto_update'>: False,
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'firmware',
<UpdateEntityStateAttribute.DISPLAY_PRECISION: 'display_precision'>: 0,
<EntityStateAttribute.ENTITY_PICTURE: 'entity_picture'>: '/api/brands/integration/wattwaechter/icon.png',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Haushalt Test Firmware',
<UpdateEntityStateAttribute.IN_PROGRESS: 'in_progress'>: False,
<UpdateEntityStateAttribute.INSTALLED_VERSION: 'installed_version'>: '1.2.3',
<UpdateEntityStateAttribute.LATEST_VERSION: 'latest_version'>: '1.3.0',
<UpdateEntityStateAttribute.RELEASE_SUMMARY: 'release_summary'>: None,
<UpdateEntityStateAttribute.RELEASE_URL: 'release_url'>: None,
<UpdateEntityStateAttribute.SKIPPED_VERSION: 'skipped_version'>: None,
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <UpdateEntityFeature: 17>,
<UpdateEntityStateAttribute.TITLE: 'title'>: None,
<UpdateEntityStateAttribute.UPDATE_PERCENTAGE: 'update_percentage'>: None,
}),
'context': <ANY>,
'entity_id': 'update.haushalt_test_firmware',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
tests/components/wattwaechter/test_sensor.py
+5 -4
View File
@@ -3,7 +3,7 @@
from __future__ import annotations
from datetime import timedelta
from unittest.mock import AsyncMock
from unittest.mock import AsyncMock, patch
from aio_wattwaechter import (
WattwaechterConnectionError,
@@ -16,7 +16,7 @@ import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.wattwaechter.const import DEFAULT_SCAN_INTERVAL, DOMAIN
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
@@ -34,8 +34,9 @@ async def test_all_entities(
mock_client: AsyncMock,
) -> None:
"""Test all sensor entities created from a full OBIS payload."""
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
with patch("homeassistant.components.wattwaechter.PLATFORMS", [Platform.SENSOR]):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
tests/components/wattwaechter/test_update.py
+183
View File
@@ -0,0 +1,183 @@
"""Tests for the WattWächter Plus update platform."""
from dataclasses import replace
from unittest.mock import AsyncMock, patch
from aio_wattwaechter import WattwaechterConnectionError
from aio_wattwaechter.models import OtaCheckResponse
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.update import DOMAIN as UPDATE_DOMAIN, SERVICE_INSTALL
from homeassistant.components.wattwaechter.const import DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from .conftest import MOCK_DEVICE_ID, MOCK_FW_VERSION, MOCK_OTA_CHECK
from tests.common import MockConfigEntry, snapshot_platform
from tests.typing import WebSocketGenerator
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
mock_client: AsyncMock,
) -> None:
"""Test the firmware update entity."""
with patch("homeassistant.components.wattwaechter.PLATFORMS", [Platform.UPDATE]):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_install(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: AsyncMock,
entity_registry: er.EntityRegistry,
) -> None:
"""Test installing a firmware update triggers ota_start."""
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id("update", DOMAIN, MOCK_DEVICE_ID)
assert entity_id is not None
await hass.services.async_call(
UPDATE_DOMAIN,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_client.ota_start.assert_awaited_once()
@pytest.mark.parametrize(
("ota_start_return", "ota_start_side_effect"),
[
pytest.param(False, None, id="rejected"),
pytest.param(True, WattwaechterConnectionError("offline"), id="error"),
],
)
async def test_install_fails(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: AsyncMock,
entity_registry: er.EntityRegistry,
ota_start_return: bool,
ota_start_side_effect: Exception | None,
) -> None:
"""Test a rejected or failed firmware update raises an error."""
mock_client.ota_start.return_value = ota_start_return
mock_client.ota_start.side_effect = ota_start_side_effect
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id("update", DOMAIN, MOCK_DEVICE_ID)
assert entity_id is not None
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
UPDATE_DOMAIN,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
async def test_installed_version_falls_back_without_system_info(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: AsyncMock,
entity_registry: er.EntityRegistry,
) -> None:
"""Test the installed version falls back to the stored version."""
mock_client.system_info.side_effect = WattwaechterConnectionError("offline")
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id("update", DOMAIN, MOCK_DEVICE_ID)
assert entity_id is not None
assert hass.states.get(entity_id).attributes["installed_version"] == MOCK_FW_VERSION
@pytest.mark.parametrize(
("ota_check_return", "ota_check_side_effect", "expected_latest"),
[
pytest.param(
OtaCheckResponse(
ok=True, data=replace(MOCK_OTA_CHECK.data, update_available=False)
),
None,
MOCK_FW_VERSION,
id="up_to_date",
),
pytest.param(
MOCK_OTA_CHECK,
WattwaechterConnectionError("offline"),
None,
id="ota_unavailable",
),
],
)
async def test_latest_version(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_client: AsyncMock,
entity_registry: er.EntityRegistry,
ota_check_return: OtaCheckResponse,
ota_check_side_effect: Exception | None,
expected_latest: str | None,
) -> None:
"""Test the latest version reflects the OTA status."""
mock_client.ota_check.return_value = ota_check_return
mock_client.ota_check.side_effect = ota_check_side_effect
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id("update", DOMAIN, MOCK_DEVICE_ID)
assert entity_id is not None
assert hass.states.get(entity_id).attributes["latest_version"] == expected_latest
@pytest.mark.parametrize(
("ota_check_side_effect", "expected_notes"),
[
pytest.param(None, "Bug fixes and improvements", id="available"),
pytest.param(WattwaechterConnectionError("offline"), None, id="unavailable"),
],
)
async def test_release_notes(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_config_entry: MockConfigEntry,
mock_client: AsyncMock,
entity_registry: er.EntityRegistry,
ota_check_side_effect: Exception | None,
expected_notes: str | None,
) -> None:
"""Test the firmware release notes are exposed."""
mock_client.ota_check.side_effect = ota_check_side_effect
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id("update", DOMAIN, MOCK_DEVICE_ID)
assert entity_id is not None
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{"type": "update/release_notes", "entity_id": entity_id}
)
result = await client.receive_json()
assert result["result"] == expected_notes