diff --git a/homeassistant/components/openevse/diagnostics.py b/homeassistant/components/openevse/diagnostics.py new file mode 100644 index 0000000000000..e0ba563ad37c6 --- /dev/null +++ b/homeassistant/components/openevse/diagnostics.py @@ -0,0 +1,166 @@ +"""Provide diagnostics for OpenEVSE.""" + +import asyncio +from datetime import date, datetime +from enum import Enum +from typing import Any + +from homeassistant.components.diagnostics import async_redact_data +from homeassistant.const import CONF_PASSWORD, CONF_USERNAME +from homeassistant.core import HomeAssistant + +from .coordinator import OpenEVSEConfigEntry + +REDACT_CONFIG_DATA = {CONF_PASSWORD, CONF_USERNAME} + +MAX_JSON_DEPTH = 20 + +CHARGER_PROPERTIES = ( + "status", + "vehicle", + "mode", + "charge_mode", + "divertmode", + "manual_override", + "ota_update", + "service_level", + "charge_time_elapsed", + "vehicle_eta", + "charging_current", + "charging_voltage", + "charging_power", + "current_power", + "current_capacity", + "max_current", + "min_amps", + "max_amps", + "max_current_soft", + "available_current", + "smoothed_available_current", + "charge_rate", + "ambient_temperature", + "ir_temperature", + "rtc_temperature", + "esp_temperature", + "usage_session", + "usage_total", + "total_day", + "total_week", + "total_month", + "total_year", + "vehicle_soc", + "vehicle_range", + "wifi_signal", + "shaper_live_power", + "shaper_available_current", + "shaper_max_power", + "gfi_trip_count", + "no_gnd_trip_count", + "stuck_relay_trip_count", + "uptime", + "freeram", + "wifi_firmware", + "openevse_firmware", +) + + +def _to_json_safe(val: Any, seen: set[int] | None = None, depth: int = 0) -> Any: + """Coerce value to be JSON-serializable. + + Top-level callables on the charger object are skipped entirely in the main + diagnostics loop. For nested structures (lists, dicts, tuples, sets), any + encountered callable elements are coerced to None here to preserve the + structure while remaining JSON-safe. + """ + if isinstance(val, (str, int, float, bool)) or val is None: + return val + + # Limit the maximum allowed depth value (0-based) to MAX_JSON_DEPTH. + # At depth = MAX_JSON_DEPTH, we truncate and return a placeholder. + if depth >= MAX_JSON_DEPTH: + return f"" + + if seen is None: + seen = set() + + val_id = id(val) + if val_id in seen: + return f"" + + if isinstance(val, (datetime, date)): + return val.isoformat() + if isinstance(val, Enum): + return _to_json_safe(val.value, seen, depth + 1) + if isinstance(val, (set, frozenset)): + seen.add(val_id) + try: + coerced_vals = [_to_json_safe(v, seen, depth + 1) for v in val] + return sorted(coerced_vals, key=str) + finally: + seen.remove(val_id) + if isinstance(val, (list, tuple)): + seen.add(val_id) + try: + return [_to_json_safe(v, seen, depth + 1) for v in val] + finally: + seen.remove(val_id) + if isinstance(val, dict): + seen.add(val_id) + try: + key_mappings = [] + for k in val: + if isinstance(k, str): + key_str = k + elif isinstance(k, (int, float, bool)) or k is None: + key_str = f"<{type(k).__name__}: {k}>" + elif isinstance(k, Enum): + key_str = f"{type(k).__name__}.{k.name}" + else: + key_str = f"<{type(k).__name__}>" + key_mappings.append((key_str, k)) + + # Sort key_str pairs deterministically, avoiding comparisons on original objects/keys + key_mappings.sort(key=lambda item: item[0]) + + res = {} + for key_str, k in key_mappings: + res[key_str] = _to_json_safe(val[k], seen, depth + 1) + return res + finally: + seen.remove(val_id) + if callable(val): + return None + return f"<{type(val).__name__} object>" + + +async def async_get_config_entry_diagnostics( + _hass: HomeAssistant, config_entry: OpenEVSEConfigEntry +) -> dict[str, Any]: + """Return diagnostics for a config entry.""" + coordinator = config_entry.runtime_data + charger = coordinator.charger + + charger_data: dict[str, Any] = {} + + for prop in CHARGER_PROPERTIES: + try: + val = getattr(charger, prop) + except AttributeError: + continue + except asyncio.CancelledError: + raise + except Exception as err: # noqa: BLE001 + charger_data[prop] = f"Error: {type(err).__name__}" + continue + + # Top-level callables on the charger object are omitted from diagnostics. + # Any nested callables within collections are coerced to None by _to_json_safe. + if callable(val): + continue + + charger_data[prop] = _to_json_safe(val) + + return { + "config_entry": async_redact_data(config_entry.as_dict(), REDACT_CONFIG_DATA), + "charger": charger_data, + } diff --git a/tests/components/openevse/test_diagnostics.py b/tests/components/openevse/test_diagnostics.py new file mode 100644 index 0000000000000..7ed06a3eb8c13 --- /dev/null +++ b/tests/components/openevse/test_diagnostics.py @@ -0,0 +1,275 @@ +"""Test OpenEVSE diagnostics.""" + +import asyncio +from datetime import datetime +from enum import Enum +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from homeassistant.components.openevse.diagnostics import ( + MAX_JSON_DEPTH, + async_get_config_entry_diagnostics, +) +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_entry_diagnostics( + hass: HomeAssistant, + hass_client: ClientSessionGenerator, + mock_config_entry: MockConfigEntry, + mock_charger: MagicMock, +) -> None: + """Test OpenEVSE diagnostics.""" + 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() + + diagnostics = await get_diagnostics_for_config_entry( + hass, hass_client, mock_config_entry + ) + + assert diagnostics["config_entry"]["data"] == { + "host": "192.168.1.100", + } + assert diagnostics["charger"]["status"] == "Charging" + assert diagnostics["charger"]["charging_voltage"] == 240 + assert diagnostics["charger"]["charging_current"] == 32000.0 + + +async def test_entry_diagnostics_redact( + hass: HomeAssistant, + hass_client: ClientSessionGenerator, + mock_charger: MagicMock, +) -> None: + """Test OpenEVSE diagnostics with auth data redacted.""" + entry = MockConfigEntry( + title="openevse_mock_config", + domain="openevse", + data={ + "host": "192.168.1.100", + "username": "my_username", + "password": "my_password", + }, + entry_id="FAKE_AUTH", + unique_id="deadbeeffeed", + ) + entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + diagnostics = await get_diagnostics_for_config_entry(hass, hass_client, entry) + + assert diagnostics["config_entry"]["data"] == { + "host": "192.168.1.100", + "username": "**REDACTED**", + "password": "**REDACTED**", + } + + +async def test_entry_diagnostics_exceptions( + hass: HomeAssistant, + hass_client: ClientSessionGenerator, + mock_config_entry: MockConfigEntry, + mock_charger: MagicMock, +) -> None: + """Test OpenEVSE diagnostics handles exceptions and JSON coercion correctly.""" + + class MockEnum(Enum): + TEST = "test_value" + + class FakeCharger: + """Fake charger to raise exceptions and return custom values for properties.""" + + def __init__(self, original_charger: MagicMock) -> None: + self._original_charger = original_charger + # Copy other properties from the original mock for realism + excluded = { + "status", + "charging_voltage", + "vehicle_eta", + "mode", + "divertmode", + "manual_override", + "ota_update", + "service_level", + "uptime", + "wifi_firmware", + "openevse_firmware", + "wifi_signal", + "freeram", + } + # Copy all mock attributes except the overridden ones + for key, val in original_charger.__dict__.items(): + if key not in excluded: + self.__dict__[key] = val + + @property + def charging_voltage(self) -> int: + raise ValueError("Connection error") + + @property + def vehicle_eta(self) -> datetime: + return datetime(2000, 1, 1, 12, 0, 0) + + @property + def mode(self) -> MockEnum: + return MockEnum.TEST + + @property + def divertmode(self) -> set[str]: + return {"solar", "eco"} + + @property + def manual_override(self) -> frozenset[str]: + return frozenset({"override"}) + + @property + def ota_update(self) -> tuple[Any, ...]: + return ("v1.0", lambda: "nested_callable") + + @property + def service_level(self) -> dict[MockEnum, str]: + return {MockEnum.TEST: "level_2"} + + @property + def uptime(self) -> object: + class CustomObj: + def __str__(self) -> str: + return "custom_str" + + return CustomObj() + + @property + def custom_key_obj(self) -> object: + class CustomKeyObj: + pass + + return CustomKeyObj() + + @property + def wifi_firmware(self) -> Any: + return lambda: "callable_value" + + @property + def openevse_firmware(self) -> list[Any]: + cyclic: list[Any] = [] + cyclic.append(cyclic) + return cyclic + + @property + def wifi_signal(self) -> list[Any]: + nested: list[Any] = [] + curr = nested + for _ in range(MAX_JSON_DEPTH + 2): + new_list: list[Any] = [] + curr.append(new_list) + curr = new_list + return nested + + @property + def freeram(self) -> dict[Any, Any]: + return { + "simple": "val", + 123: "int_key", + MockEnum.TEST: "enum_key", + self.custom_key_obj: "obj_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() + + # Inject the FakeCharger into the coordinator to isolate side effects + coordinator = mock_config_entry.runtime_data + coordinator.charger = FakeCharger(mock_charger) + + diagnostics = await get_diagnostics_for_config_entry( + hass, hass_client, mock_config_entry + ) + + # status should be omitted because the attribute is not present + assert "status" not in diagnostics["charger"] + + # charging_voltage should show the recorded error type only + assert diagnostics["charger"]["charging_voltage"] == "Error: ValueError" + + # vehicle_eta should be coerced to ISO format string + assert diagnostics["charger"]["vehicle_eta"] == "2000-01-01T12:00:00" + + # mode should be coerced to Enum raw value + assert diagnostics["charger"]["mode"] == "test_value" + + # divertmode should be sorted and coerced to list + assert diagnostics["charger"]["divertmode"] == ["eco", "solar"] + + # manual_override should be sorted and coerced to list + assert diagnostics["charger"]["manual_override"] == ["override"] + + # ota_update should be coerced to list, with callable elements coerced to None + assert diagnostics["charger"]["ota_update"] == ["v1.0", None] + + # service_level should have keys coerced to str + assert diagnostics["charger"]["service_level"] == {"MockEnum.TEST": "level_2"} + + # uptime should fallback to type name representation + assert diagnostics["charger"]["uptime"] == "" + + # wifi_firmware should be omitted because it is callable + assert "wifi_firmware" not in diagnostics["charger"] + + # wifi_signal has a deeply nested list exceeding the limit + expected_wifi_signal: list[Any] = [] + curr = expected_wifi_signal + for _ in range(MAX_JSON_DEPTH - 1): + new_list = [] + curr.append(new_list) + curr = new_list + curr.append("") + assert diagnostics["charger"]["wifi_signal"] == expected_wifi_signal + + # freeram key types and deterministic serialization + assert diagnostics["charger"]["freeram"] == { + "": "int_key", + "MockEnum.TEST": "enum_key", + "": "obj_key", + "simple": "val", + } + + # openevse_firmware contains circular reference + assert diagnostics["charger"]["openevse_firmware"] == [ + "" + ] + + +async def test_entry_diagnostics_cancelled_error( + hass: HomeAssistant, + hass_client: ClientSessionGenerator, + mock_config_entry: MockConfigEntry, + mock_charger: MagicMock, +) -> None: + """Test OpenEVSE diagnostics handles asyncio.CancelledError correctly.""" + + class CancelledFakeCharger: + @property + def status(self) -> str: + raise asyncio.CancelledError + + @property + def websocket(self) -> Any: + return None + + 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() + + coordinator = mock_config_entry.runtime_data + coordinator.charger = CancelledFakeCharger() + + with pytest.raises(asyncio.CancelledError): + await async_get_config_entry_diagnostics(hass, mock_config_entry)