Add diagnostics for Axle Energy (#183818)

This commit is contained in:
Anthony
2026-10-01 21:03:02 +02:00
committed by GitHub
parent b14e6f2b55
commit ba0f70cec6
4 changed files with 154 additions and 1 deletions
@@ -0,0 +1,19 @@
"""Diagnostics support for Axle Energy."""
from dataclasses import asdict
from typing import Any
from homeassistant.core import HomeAssistant
from .coordinator import AxleConfigEntry
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: AxleConfigEntry
) -> dict[str, Any]:
"""Return cached event data without requesting an update."""
coordinator = entry.runtime_data
return {
"last_update_success": coordinator.last_update_success,
"data": asdict(coordinator.data) if coordinator.data is not None else None,
}
@@ -42,7 +42,7 @@ rules:
reauthentication-flow: done
test-coverage: done
devices: done
diagnostics: todo
diagnostics: done
discovery:
status: exempt
comment: The household event feed is a cloud service with no local endpoint.
@@ -0,0 +1,37 @@
# serializer version: 1
# name: test_diagnostics[healthy-empty]
dict({
'data': None,
'last_update_success': True,
})
# ---
# name: test_diagnostics[healthy-scheduled]
dict({
'data': dict({
'direction': 'export',
'end': '2026-09-11T18:00:00+00:00',
'opted_out': False,
'start': '2026-09-11T17:00:00+00:00',
'updated_at': '2026-09-11T08:00:00+00:00',
}),
'last_update_success': True,
})
# ---
# name: test_diagnostics[unavailable-empty]
dict({
'data': None,
'last_update_success': False,
})
# ---
# name: test_diagnostics[unavailable-scheduled]
dict({
'data': dict({
'direction': 'export',
'end': '2026-09-11T18:00:00+00:00',
'opted_out': False,
'start': '2026-09-11T17:00:00+00:00',
'updated_at': '2026-09-11T08:00:00+00:00',
}),
'last_update_success': False,
})
# ---
@@ -0,0 +1,97 @@
"""Test Axle Energy diagnostics."""
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock
from aioaxlevpp import AxleConnectionError, GridEvent
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.axle_energy.const import DOMAIN
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, async_fire_time_changed
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
@pytest.mark.parametrize(
"event",
[
pytest.param(
GridEvent(
start=datetime(2026, 9, 11, 17, tzinfo=UTC),
end=datetime(2026, 9, 11, 18, tzinfo=UTC),
direction="export",
updated_at=datetime(2026, 9, 11, 8, tzinfo=UTC),
),
id="scheduled",
),
pytest.param(None, id="empty"),
],
)
@pytest.mark.parametrize(
"error",
[
pytest.param(None, id="healthy"),
pytest.param(AxleConnectionError(), id="unavailable"),
],
)
async def test_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_config_entry: MockConfigEntry,
mock_client: AsyncMock,
freezer: FrozenDateTimeFactory,
snapshot: SnapshotAssertion,
event: GridEvent | None,
error: AxleConnectionError | None,
) -> None:
"""Exclude the key and retain cached data without another API request."""
mock_client.get_event.return_value = event
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mock_client.get_event.side_effect = error
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
mock_client.get_event.reset_mock()
assert (
await get_diagnostics_for_config_entry(hass, hass_client, mock_config_entry)
== snapshot
)
mock_client.get_event.assert_not_awaited()
assert mock_config_entry.data[CONF_API_KEY] == "test-token"
async def test_diagnostics_multiple_entries(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_config_entry: MockConfigEntry,
mock_client: AsyncMock,
) -> None:
"""Download only the selected feed's cached event without exposing either key."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mock_client.get_event.return_value = None
other_entry = MockConfigEntry(domain=DOMAIN, data={CONF_API_KEY: "other-token"})
other_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(other_entry.entry_id)
await hass.async_block_till_done()
mock_client.get_event.reset_mock()
first = await get_diagnostics_for_config_entry(hass, hass_client, mock_config_entry)
second = await get_diagnostics_for_config_entry(hass, hass_client, other_entry)
assert first["data"]["direction"] == "export"
assert second["data"] is None
assert "entry_data" not in first
assert "entry_data" not in second
mock_client.get_event.assert_not_awaited()
assert mock_config_entry.data[CONF_API_KEY] == "test-token"
assert other_entry.data[CONF_API_KEY] == "other-token"