starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-24 15:31:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add iZone config entry diagnostics via dump_state (#177255)

Browse Source
This commit is contained in:
Penny Wood
2026-07-27 14:34:29 +02:00
committed by GitHub
parent 79ec3dd3af
commit 0e4895b232
4 changed files with 388 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/izone/diagnostics.py
+68
View File
@@ -0,0 +1,68 @@
"""Diagnostics support for iZone."""
from collections.abc import Mapping
from typing import Any
from homeassistant.components.diagnostics import REDACTED
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from .coordinator import IZoneConfigEntry
from .discovery import DATA_DISCOVERY_SERVICE
TO_REDACT = {
CONF_HOST,
"device_ip",
"source_ip",
}
def _redact_data(data: Any) -> Any:
"""Redact host keys and scrub those host values from sibling strings."""
if isinstance(data, list):
return [_redact_data(item) for item in data]
if isinstance(data, Mapping):
hosts = {
value
for key, value in data.items()
if key in TO_REDACT and isinstance(value, str)
}
result: dict[Any, Any] = {}
for key, value in data.items():
if key in TO_REDACT:
result[key] = REDACTED
continue
redacted = _redact_data(value)
if hosts and isinstance(redacted, str):
for host in hosts:
redacted = redacted.replace(host, REDACTED)
result[key] = redacted
return result
return data
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: IZoneConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
controller = entry.runtime_data.controller
discovery_slot = hass.data.get(DATA_DISCOVERY_SERVICE)
if discovery_slot is not None and discovery_slot.runtime is not None:
discovery: dict[str, Any] = {
"running": True,
**discovery_slot.runtime.service.dump_state(),
}
else:
discovery = {"running": False}
return _redact_data(
{
"entry": {
**dict(entry.data),
"unique_id": entry.unique_id,
},
"discovery": discovery,
"controller": controller.dump_state(),
}
)
tests/components/izone/conftest.py
+96
View File
@@ -51,6 +51,9 @@ def create_mock_controller(
controller.free_air = free_air
controller.is_on = is_on
controller.connected = True
controller.bridge_connected = True
controller.is_v2 = False
controller.is_ipower = False
controller.mode = Controller.Mode.COOL
controller.temp_setpoint = 24.0
controller.temp_return = 22.0
@@ -65,6 +68,29 @@ def create_mock_controller(
Controller.Fan.AUTO,
]
controller.zones = []
controller._system_settings = {
"AirStreamDeviceUId": device_uid,
"DeviceType": "ASH",
"SysOn": "on" if is_on else "off",
"SysMode": "cool",
"SysFan": "med",
"SleepTimer": 0,
"Supply": "16.0",
"Setpoint": "24.0",
"Temp": "22.0",
"RAS": ras_mode,
"CtrlZone": zone_ctrl,
"EcoLock": "true",
"EcoMax": "30.0",
"EcoMin": "15.0",
"NoOfConst": 0,
"NoOfZones": zones_total,
"SysType": sys_type,
"FreeAir": "disabled"
if not free_air_enabled
else ("on" if free_air else "off"),
"FanAuto": "3-speed",
}
controller.refresh_all = AsyncMock()
controller.close = AsyncMock()
controller.set_temp_setpoint = AsyncMock()
@@ -72,6 +98,25 @@ def create_mock_controller(
controller.set_on = AsyncMock()
controller.set_mode = AsyncMock()
controller.set_free_air = AsyncMock()
def dump_state() -> dict:
return {
"device_uid": controller.device_uid,
"device_ip": controller.device_ip,
"connected": controller.connected,
"bridge_connected": controller.bridge_connected,
"is_v2": controller.is_v2,
"is_ipower": controller.is_ipower,
"fan_modes": [
fan.value if hasattr(fan, "value") else fan
for fan in controller.fan_modes
],
"system_settings": dict(controller._system_settings),
"zones": [zone.dump_state() for zone in controller.zones],
"power": None,
}
controller.dump_state = dump_state
return controller
@@ -105,10 +150,32 @@ def create_mock_zone(
zone.airflow_min = 0
zone.airflow_max = 100
zone.is_on = True
zone._zone_data = {
"AirStreamDeviceUId": "000000001",
"Id": 0,
"Index": index,
"Name": name,
"Type": "auto",
"Mode": "auto",
"SetPoint": temp_setpoint,
"Temp": temp_current if temp_current is not None else 0,
"MaxAir": 100,
"MinAir": 0,
"Const": 255,
"ConstA": "false",
"DmpFlt": "false",
"Master": "false",
"iSense": "false",
}
zone.set_airflow_min = AsyncMock()
zone.set_airflow_max = AsyncMock()
zone.set_temp_setpoint = AsyncMock()
zone.set_mode = AsyncMock()
def dump_state() -> dict:
return dict(zone._zone_data)
zone.dump_state = dump_state
return zone
@@ -212,6 +279,35 @@ def mock_discovery_service(mock_controller: Mock) -> Mock:
return_value=endpoint_from_controller(mock_controller)
)
service.create_controller = AsyncMock(return_value=mock_controller)
def dump_state() -> dict:
return {
"closed": False,
"udp_bound": True,
"claimed": [
{
"uid": mock_controller.device_uid,
"host": mock_controller.device_ip,
}
],
"known": [],
"recent_udp": [
{
"source_ip": mock_controller.device_ip,
"source_port": 12107,
"received_at": "2026-07-25T14:37:00.257385+00:00",
"message": (
f"ASPort_12107,Mac_{mock_controller.device_uid},"
f"IP_{mock_controller.device_ip},iZoneV2"
),
"uid": mock_controller.device_uid,
"host": mock_controller.device_ip,
"tags": ["iZoneV2"],
}
],
}
service.dump_state = dump_state
return service
tests/components/izone/snapshots/test_diagnostics.ambr
+90
View File
@@ -0,0 +1,90 @@
# serializer version: 1
# name: test_config_entry_diagnostics
dict({
'controller': dict({
'bridge_connected': True,
'connected': True,
'device_ip': '**REDACTED**',
'device_uid': '000000001',
'fan_modes': list([
'low',
'med',
'high',
'auto',
]),
'is_ipower': False,
'is_v2': False,
'power': None,
'system_settings': dict({
'AirStreamDeviceUId': '000000001',
'CtrlZone': 1,
'DeviceType': 'ASH',
'EcoLock': 'true',
'EcoMax': '30.0',
'EcoMin': '15.0',
'FanAuto': '3-speed',
'FreeAir': 'disabled',
'NoOfConst': 0,
'NoOfZones': 1,
'RAS': 'master',
'Setpoint': '24.0',
'SleepTimer': 0,
'Supply': '16.0',
'SysFan': 'med',
'SysMode': 'cool',
'SysOn': 'on',
'SysType': 'iZone310',
'Temp': '22.0',
}),
'zones': list([
dict({
'AirStreamDeviceUId': '000000001',
'Const': 255,
'ConstA': 'false',
'DmpFlt': 'false',
'Id': 0,
'Index': 0,
'Master': 'false',
'MaxAir': 100,
'MinAir': 0,
'Mode': 'auto',
'Name': 'Living Room',
'SetPoint': 24.0,
'Temp': 22.5,
'Type': 'auto',
'iSense': 'false',
}),
]),
}),
'discovery': dict({
'claimed': list([
dict({
'host': '**REDACTED**',
'uid': '000000001',
}),
]),
'closed': False,
'known': list([
]),
'recent_udp': list([
dict({
'host': '**REDACTED**',
'message': 'ASPort_12107,Mac_000000001,IP_**REDACTED**,iZoneV2',
'received_at': '2026-07-25T14:37:00.257385+00:00',
'source_ip': '**REDACTED**',
'source_port': 12107,
'tags': list([
'iZoneV2',
]),
'uid': '000000001',
}),
]),
'running': True,
'udp_bound': True,
}),
'entry': dict({
'host': '**REDACTED**',
'unique_id': '000000001',
}),
})
# ---
tests/components/izone/test_diagnostics.py
+134
View File
@@ -0,0 +1,134 @@
"""Test the iZone diagnostics."""
from collections.abc import Callable
from unittest.mock import Mock
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.diagnostics import REDACTED
from homeassistant.components.izone.discovery import DATA_DISCOVERY_SERVICE
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
async def test_config_entry_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
init_integration: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Config entry diagnostics include discovery and controller dump_state."""
entry = init_integration
result = await get_diagnostics_for_config_entry(hass, hass_client, entry)
# Diagnostics JSON round-trips REDACTED to its string form.
assert result["entry"][CONF_HOST] == str(REDACTED)
assert result["entry"]["unique_id"] == "000000001"
assert result["discovery"]["running"] is True
assert result["discovery"]["udp_bound"] is True
assert result["discovery"]["claimed"][0]["host"] == str(REDACTED)
assert result["discovery"]["claimed"][0]["uid"] == "000000001"
udp = result["discovery"]["recent_udp"][0]
assert udp["host"] == str(REDACTED)
assert udp["source_ip"] == str(REDACTED)
assert entry.runtime_data.controller.device_ip not in udp["message"]
assert str(REDACTED) in udp["message"]
assert result["controller"]["device_ip"] == str(REDACTED)
assert result["controller"]["device_uid"] == "000000001"
assert result["controller"]["system_settings"]["RAS"] == "master"
assert result["controller"]["system_settings"]["CtrlZone"] == 1
assert result["controller"]["zones"][0]["Name"] == "Living Room"
assert result["controller"]["connected"] is True
assert result["controller"]["fan_modes"] == ["low", "med", "high", "auto"]
assert result == snapshot
async def test_config_entry_diagnostics_redacts_discovery_peer_hosts(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
init_integration: MockConfigEntry,
mock_discovery_service: Mock,
) -> None:
"""Hosts from discovery peers are redacted in UDP message payloads."""
peer_ip = "198.51.100.9"
def dump_state() -> dict:
return {
"closed": False,
"udp_bound": True,
"claimed": [
{
"uid": "000000001",
"host": "192.0.2.1",
}
],
"known": [
{
"uid": "000000009",
"host": peer_ip,
}
],
"recent_udp": [
{
"source_ip": peer_ip,
"source_port": 12107,
"received_at": "2026-07-25T14:37:00.257385+00:00",
"message": f"ASPort_12107,Mac_000000009,IP_{peer_ip},iZone",
"uid": "000000009",
"host": peer_ip,
"tags": ["iZone"],
}
],
}
mock_discovery_service.dump_state = dump_state
result = await get_diagnostics_for_config_entry(hass, hass_client, init_integration)
assert result["discovery"]["known"][0]["host"] == str(REDACTED)
udp = result["discovery"]["recent_udp"][0]
assert udp["host"] == str(REDACTED)
assert udp["source_ip"] == str(REDACTED)
assert peer_ip not in udp["message"]
assert str(REDACTED) in udp["message"]
def _pop_discovery_slot(hass: HomeAssistant) -> None:
"""Remove the shared discovery slot entirely."""
hass.data.pop(DATA_DISCOVERY_SERVICE, None)
def _clear_discovery_runtime(hass: HomeAssistant) -> None:
"""Leave the slot present but with discovery stopped (runtime cleared)."""
hass.data[DATA_DISCOVERY_SERVICE].runtime = None
@pytest.mark.parametrize(
"prepare_discovery",
[
pytest.param(_pop_discovery_slot, id="slot_absent"),
pytest.param(_clear_discovery_runtime, id="runtime_stopped"),
],
)
async def test_config_entry_diagnostics_without_discovery(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
init_integration: MockConfigEntry,
prepare_discovery: Callable[[HomeAssistant], None],
) -> None:
"""Diagnostics report discovery stopped when discovery is not running."""
entry = init_integration
prepare_discovery(hass)
result = await get_diagnostics_for_config_entry(hass, hass_client, entry)
assert result["discovery"] == {"running": False}
assert result["entry"][CONF_HOST] == str(REDACTED)
assert result["entry"]["unique_id"] == "000000001"
assert result["controller"]["device_uid"] == "000000001"
assert result["controller"]["device_ip"] == str(REDACTED)