Add automatic backup event entity to Home Assistant Backup system (#145350)

* add automatic backup event entity

* add tests

* fix test

* Apply suggestions from code review

Co-authored-by: Josef Zweck <josef@zweck.dev>

* implement _handle_coordinator_update

* add translations for event attributes

* simplify condition

* Apply suggestions from code review

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

---------

Co-authored-by: Josef Zweck <josef@zweck.dev>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Michael 2025-05-23 17:01:57 +02:00 committed by GitHub
parent 5048d1512c
commit 086e97821f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 257 additions and 5 deletions

View File

@ -81,7 +81,7 @@ __all__ = [
"suggested_filename_from_name_date",
]
PLATFORMS = [Platform.SENSOR]
PLATFORMS = [Platform.EVENT, Platform.SENSOR]
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)

View File

@ -33,6 +33,7 @@ class BackupCoordinatorData:
last_attempted_automatic_backup: datetime | None
last_successful_automatic_backup: datetime | None
next_scheduled_automatic_backup: datetime | None
last_event: ManagerStateEvent | BackupPlatformEvent | None
class BackupDataUpdateCoordinator(DataUpdateCoordinator[BackupCoordinatorData]):
@ -60,11 +61,13 @@ class BackupDataUpdateCoordinator(DataUpdateCoordinator[BackupCoordinatorData]):
]
self.backup_manager = backup_manager
self._last_event: ManagerStateEvent | BackupPlatformEvent | None = None
@callback
def _on_event(self, event: ManagerStateEvent | BackupPlatformEvent) -> None:
"""Handle new event."""
LOGGER.debug("Received backup event: %s", event)
self._last_event = event
self.config_entry.async_create_task(self.hass, self.async_refresh())
async def _async_update_data(self) -> BackupCoordinatorData:
@ -74,6 +77,7 @@ class BackupDataUpdateCoordinator(DataUpdateCoordinator[BackupCoordinatorData]):
self.backup_manager.config.data.last_attempted_automatic_backup,
self.backup_manager.config.data.last_completed_automatic_backup,
self.backup_manager.config.data.schedule.next_automatic_backup,
self._last_event,
)
@callback

View File

@ -11,7 +11,7 @@ from .const import DOMAIN
from .coordinator import BackupDataUpdateCoordinator
class BackupManagerEntity(CoordinatorEntity[BackupDataUpdateCoordinator]):
class BackupManagerBaseEntity(CoordinatorEntity[BackupDataUpdateCoordinator]):
"""Base entity for backup manager."""
_attr_has_entity_name = True
@ -19,12 +19,9 @@ class BackupManagerEntity(CoordinatorEntity[BackupDataUpdateCoordinator]):
def __init__(
self,
coordinator: BackupDataUpdateCoordinator,
entity_description: EntityDescription,
) -> None:
"""Initialize base entity."""
super().__init__(coordinator)
self.entity_description = entity_description
self._attr_unique_id = entity_description.key
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, "backup_manager")},
manufacturer="Home Assistant",
@ -34,3 +31,17 @@ class BackupManagerEntity(CoordinatorEntity[BackupDataUpdateCoordinator]):
entry_type=DeviceEntryType.SERVICE,
configuration_url="homeassistant://config/backup",
)
class BackupManagerEntity(BackupManagerBaseEntity):
"""Entity for backup manager."""
def __init__(
self,
coordinator: BackupDataUpdateCoordinator,
entity_description: EntityDescription,
) -> None:
"""Initialize entity."""
super().__init__(coordinator)
self.entity_description = entity_description
self._attr_unique_id = entity_description.key

View File

@ -0,0 +1,59 @@
"""Event platform for Home Assistant Backup integration."""
from __future__ import annotations
from typing import Final
from homeassistant.components.event import EventEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import BackupConfigEntry, BackupDataUpdateCoordinator
from .entity import BackupManagerBaseEntity
from .manager import CreateBackupEvent, CreateBackupState
ATTR_BACKUP_STAGE: Final[str] = "backup_stage"
ATTR_FAILED_REASON: Final[str] = "failed_reason"
async def async_setup_entry(
hass: HomeAssistant,
config_entry: BackupConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Event set up for backup config entry."""
coordinator = config_entry.runtime_data
async_add_entities([AutomaticBackupEvent(coordinator)])
class AutomaticBackupEvent(BackupManagerBaseEntity, EventEntity):
"""Representation of an automatic backup event."""
_attr_event_types = [s.value for s in CreateBackupState]
_unrecorded_attributes = frozenset({ATTR_FAILED_REASON, ATTR_BACKUP_STAGE})
coordinator: BackupDataUpdateCoordinator
def __init__(self, coordinator: BackupDataUpdateCoordinator) -> None:
"""Initialize the automatic backup event."""
super().__init__(coordinator)
self._attr_unique_id = "automatic_backup_event"
self._attr_translation_key = "automatic_backup_event"
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
if (
not (data := self.coordinator.data)
or (event := data.last_event) is None
or not isinstance(event, CreateBackupEvent)
):
return
self._trigger_event(
event.state,
{
ATTR_BACKUP_STAGE: event.stage,
ATTR_FAILED_REASON: event.reason,
},
)
self.async_write_ha_state()

View File

@ -1,4 +1,11 @@
{
"entity": {
"event": {
"automatic_backup_event": {
"default": "mdi:database"
}
}
},
"services": {
"create": {
"service": "mdi:cloud-upload"

View File

@ -36,6 +36,22 @@
}
},
"entity": {
"event": {
"automatic_backup_event": {
"name": "Automatic backup",
"state_attributes": {
"event_type": {
"state": {
"completed": "Completed successfully",
"failed": "Failed",
"in_progress": "In progress"
}
},
"backup_stage": { "name": "Backup stage" },
"failed_reason": { "name": "Failure reason" }
}
}
},
"sensor": {
"backup_manager_state": {
"name": "Backup Manager state",

View File

@ -0,0 +1,60 @@
# serializer version: 1
# name: test_event_entity[event.backup_automatic_backup-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'event_types': list([
'completed',
'failed',
'in_progress',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.backup_automatic_backup',
'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': 'Automatic backup',
'platform': 'backup',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'automatic_backup_event',
'unique_id': 'automatic_backup_event',
'unit_of_measurement': None,
})
# ---
# name: test_event_entity[event.backup_automatic_backup-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'event_type': None,
'event_types': list([
'completed',
'failed',
'in_progress',
]),
'friendly_name': 'Backup Automatic backup',
}),
'context': <ANY>,
'entity_id': 'event.backup_automatic_backup',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

View File

@ -0,0 +1,95 @@
"""The tests for the Backup event entity."""
from unittest.mock import AsyncMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.backup.const import DOMAIN
from homeassistant.components.backup.event import ATTR_BACKUP_STAGE, ATTR_FAILED_REASON
from homeassistant.components.event import ATTR_EVENT_TYPE
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .common import setup_backup_integration
from tests.common import snapshot_platform
from tests.typing import WebSocketGenerator
@pytest.mark.usefixtures("mock_backup_generation")
async def test_event_entity(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test automatic backup event entity."""
with patch("homeassistant.components.backup.PLATFORMS", [Platform.EVENT]):
await setup_backup_integration(hass, with_hassio=False)
await hass.async_block_till_done(wait_background_tasks=True)
entry = hass.config_entries.async_entries(DOMAIN)[0]
await snapshot_platform(hass, entity_registry, snapshot, entry.entry_id)
@pytest.mark.usefixtures("mock_backup_generation")
async def test_event_entity_backup_completed(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test completed automatic backup event."""
with patch("homeassistant.components.backup.PLATFORMS", [Platform.EVENT]):
await setup_backup_integration(hass, with_hassio=False)
await hass.async_block_till_done(wait_background_tasks=True)
state = hass.states.get("event.backup_automatic_backup")
assert state.attributes[ATTR_EVENT_TYPE] is None
client = await hass_ws_client(hass)
await hass.async_block_till_done()
await client.send_json_auto_id(
{"type": "backup/generate", "agent_ids": ["backup.local"]}
)
assert await client.receive_json()
state = hass.states.get("event.backup_automatic_backup")
assert state.attributes[ATTR_EVENT_TYPE] == "in_progress"
assert state.attributes[ATTR_BACKUP_STAGE] is not None
assert state.attributes[ATTR_FAILED_REASON] is None
await hass.async_block_till_done(wait_background_tasks=True)
state = hass.states.get("event.backup_automatic_backup")
assert state.attributes[ATTR_EVENT_TYPE] == "completed"
assert state.attributes[ATTR_BACKUP_STAGE] is None
assert state.attributes[ATTR_FAILED_REASON] is None
@pytest.mark.usefixtures("mock_backup_generation")
async def test_event_entity_backup_failed(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
create_backup: AsyncMock,
) -> None:
"""Test failed automatic backup event."""
with patch("homeassistant.components.backup.PLATFORMS", [Platform.EVENT]):
await setup_backup_integration(hass, with_hassio=False)
await hass.async_block_till_done(wait_background_tasks=True)
state = hass.states.get("event.backup_automatic_backup")
assert state.attributes[ATTR_EVENT_TYPE] is None
create_backup.side_effect = Exception("Boom!")
client = await hass_ws_client(hass)
await hass.async_block_till_done()
await client.send_json_auto_id(
{"type": "backup/generate", "agent_ids": ["backup.local"]}
)
assert await client.receive_json()
state = hass.states.get("event.backup_automatic_backup")
assert state.attributes[ATTR_EVENT_TYPE] == "failed"
assert state.attributes[ATTR_BACKUP_STAGE] is None
assert state.attributes[ATTR_FAILED_REASON] == "unknown_error"