jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add option to ESPHome to subscribe to logs (#139073)

Browse Source
This commit is contained in:
J. Nick Koston 2025-02-25 20:56:39 +00:00 committed by GitHub
parent fe348e17a3
commit 81db3dea41
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 189 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
homeassistant/components/esphome/config_flow.py
View File

@ -41,6 +41,7 @@ from .const import (
CONF_ALLOW_SERVICE_CALLS,
CONF_DEVICE_NAME,
CONF_NOISE_PSK,
CONF_SUBSCRIBE_LOGS,
DEFAULT_ALLOW_SERVICE_CALLS,
DEFAULT_NEW_CONFIG_ALLOW_ALLOW_SERVICE_CALLS,
DOMAIN,
@ -508,6 +509,10 @@ class OptionsFlowHandler(OptionsFlow):
CONF_ALLOW_SERVICE_CALLS, DEFAULT_ALLOW_SERVICE_CALLS
),
): bool,
vol.Required(
CONF_SUBSCRIBE_LOGS,
default=self.config_entry.options.get(CONF_SUBSCRIBE_LOGS, False),
): bool,
}
)
return self.async_show_form(step_id="init", data_schema=data_schema)

1
homeassistant/components/esphome/const.py
View File

@ -5,6 +5,7 @@ from awesomeversion import AwesomeVersion
DOMAIN = "esphome"
CONF_ALLOW_SERVICE_CALLS = "allow_service_calls"
CONF_SUBSCRIBE_LOGS = "subscribe_logs"
CONF_DEVICE_NAME = "device_name"
CONF_NOISE_PSK = "noise_psk"

39
homeassistant/components/esphome/manager.py
View File

@ -5,6 +5,7 @@ from __future__ import annotations
import asyncio
from functools import partial
import logging
import re
from typing import TYPE_CHECKING, Any, NamedTuple
from aioesphomeapi import (
@ -16,6 +17,7 @@ from aioesphomeapi import (
HomeassistantServiceCall,
InvalidAuthAPIError,
InvalidEncryptionKeyAPIError,
LogLevel,
ReconnectLogic,
RequiresEncryptionAPIError,
UserService,
@ -61,6 +63,7 @@ from .bluetooth import async_connect_scanner
from .const import (
CONF_ALLOW_SERVICE_CALLS,
CONF_DEVICE_NAME,
CONF_SUBSCRIBE_LOGS,
DEFAULT_ALLOW_SERVICE_CALLS,
DEFAULT_URL,
DOMAIN,
@ -74,8 +77,30 @@ from .domain_data import DomainData
# Import config flow so that it's added to the registry
from .entry_data import ESPHomeConfigEntry, RuntimeEntryData
if TYPE_CHECKING:
from aioesphomeapi.api_pb2 import ( # type: ignore[attr-defined]
SubscribeLogsResponse,
)
_LOGGER = logging.getLogger(__name__)
LOG_LEVEL_TO_LOGGER = {
LogLevel.LOG_LEVEL_NONE: logging.DEBUG,
LogLevel.LOG_LEVEL_ERROR: logging.ERROR,
LogLevel.LOG_LEVEL_WARN: logging.WARNING,
LogLevel.LOG_LEVEL_INFO: logging.INFO,
LogLevel.LOG_LEVEL_CONFIG: logging.INFO,
LogLevel.LOG_LEVEL_DEBUG: logging.DEBUG,
LogLevel.LOG_LEVEL_VERBOSE: logging.DEBUG,
LogLevel.LOG_LEVEL_VERY_VERBOSE: logging.DEBUG,
}
# 7-bit and 8-bit C1 ANSI sequences
# https://stackoverflow.com/questions/14693701/how-can-i-remove-the-ansi-escape-sequences-from-a-string-in-python
ANSI_ESCAPE_78BIT = re.compile(
rb"(?:\x1B[@-Z\\-_]|[\x80-\x9A\x9C-\x9F]|(?:\x1B\[|\x9B)[0-?]*[ -/]*[@-~])"
)
@callback
def _async_check_firmware_version(
@ -341,6 +366,18 @@ class ESPHomeManager:
# Re-connection logic will trigger after this
await self.cli.disconnect()
def _async_on_log(self, msg: SubscribeLogsResponse) -> None:
"""Handle a log message from the API."""
logger_level = LOG_LEVEL_TO_LOGGER.get(msg.level, logging.DEBUG)
if _LOGGER.isEnabledFor(logger_level):
log: bytes = msg.message
_LOGGER.log(
logger_level,
"%s: %s",
self.entry.title,
ANSI_ESCAPE_78BIT.sub(b"", log).decode("utf-8", "backslashreplace"),
)
async def _on_connnect(self) -> None:
"""Subscribe to states and list entities on successful API login."""
entry = self.entry
@ -352,6 +389,8 @@ class ESPHomeManager:
cli = self.cli
stored_device_name = entry.data.get(CONF_DEVICE_NAME)
unique_id_is_mac_address = unique_id and ":" in unique_id
if entry.options.get(CONF_SUBSCRIBE_LOGS):
cli.subscribe_logs(self._async_on_log, LogLevel.LOG_LEVEL_VERY_VERBOSE)
results = await asyncio.gather(
create_eager_task(cli.device_info()),
create_eager_task(cli.list_entities_services()),

3
homeassistant/components/esphome/strings.json
View File

@ -54,7 +54,8 @@
"step": {
"init": {
"data": {
"allow_service_calls": "Allow the device to perform Home Assistant actions."
"allow_service_calls": "Allow the device to perform Home Assistant actions.",
"subscribe_logs": "Subscribe to logs from the device. When enabled, the device will send logs to Home Assistant and you can view them in the logs panel."
}
}
}

26
tests/components/esphome/conftest.py
View File

@ -6,7 +6,7 @@ import asyncio
from asyncio import Event
from collections.abc import AsyncGenerator, Awaitable, Callable, Coroutine
from pathlib import Path
from typing import Any
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, MagicMock, Mock, patch
from aioesphomeapi import (
@ -17,6 +17,7 @@ from aioesphomeapi import (
EntityInfo,
EntityState,
HomeassistantServiceCall,
LogLevel,
ReconnectLogic,
UserService,
VoiceAssistantAnnounceFinished,
@ -42,6 +43,10 @@ from . import DASHBOARD_HOST, DASHBOARD_PORT, DASHBOARD_SLUG
from tests.common import MockConfigEntry
if TYPE_CHECKING:
from aioesphomeapi.api_pb2 import SubscribeLogsResponse
_ONE_SECOND = 16000 * 2 # 16Khz 16-bit
@ -154,6 +159,7 @@ def mock_client(mock_device_info) -> APIClient:
mock_client.device_info = AsyncMock(return_value=mock_device_info)
mock_client.connect = AsyncMock()
mock_client.disconnect = AsyncMock()
mock_client.subscribe_logs = Mock()
mock_client.list_entities_services = AsyncMock(return_value=([], []))
mock_client.address = "127.0.0.1"
mock_client.api_version = APIVersion(99, 99)
@ -222,6 +228,7 @@ class MockESPHomeDevice:
]
| None
)
self.on_log_message: Callable[[SubscribeLogsResponse], None]
self.device_info = device_info
def set_state_callback(self, state_callback: Callable[[EntityState], None]) -> None:
@ -250,6 +257,16 @@ class MockESPHomeDevice:
"""Mock disconnecting."""
await self.on_disconnect(expected_disconnect)
def set_on_log_message(
self, on_log_message: Callable[[SubscribeLogsResponse], None]
) -> None:
"""Set the log message callback."""
self.on_log_message = on_log_message
def mock_on_log_message(self, log_message: SubscribeLogsResponse) -> None:
"""Mock on log message."""
self.on_log_message(log_message)
def set_on_connect(self, on_connect: Callable[[], None]) -> None:
"""Set the connect callback."""
self.on_connect = on_connect
@ -413,6 +430,12 @@ async def _mock_generic_device_entry(
on_state_sub, on_state_request
)
def _subscribe_logs(
on_log_message: Callable[[SubscribeLogsResponse], None], log_level: LogLevel
) -> None:
"""Subscribe to log messages."""
mock_device.set_on_log_message(on_log_message)
def _subscribe_voice_assistant(
*,
handle_start: Callable[
@ -453,6 +476,7 @@ async def _mock_generic_device_entry(
mock_client.subscribe_states = _subscribe_states
mock_client.subscribe_service_calls = _subscribe_service_calls
mock_client.subscribe_home_assistant_states = _subscribe_home_assistant_states
mock_client.subscribe_logs = _subscribe_logs
try_connect_done = Event()

66
tests/components/esphome/test_config_flow.py
View File

@ -23,6 +23,7 @@ from homeassistant.components.esphome.const import (
CONF_ALLOW_SERVICE_CALLS,
CONF_DEVICE_NAME,
CONF_NOISE_PSK,
CONF_SUBSCRIBE_LOGS,
DEFAULT_NEW_CONFIG_ALLOW_ALLOW_SERVICE_CALLS,
DOMAIN,
)
@ -1295,14 +1296,57 @@ async def test_zeroconf_no_encryption_key_via_dashboard(
assert result["step_id"] == "encryption_key"
@pytest.mark.parametrize("option_value", [True, False])
async def test_option_flow(
async def test_option_flow_allow_service_calls(
hass: HomeAssistant,
option_value: bool,
mock_client: APIClient,
mock_generic_device_entry,
) -> None:
"""Test config flow options."""
"""Test config flow options for allow service calls."""
entry = await mock_generic_device_entry(
mock_client=mock_client,
entity_info=[],
user_service=[],
states=[],
)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
assert result["data_schema"]({}) == {
CONF_ALLOW_SERVICE_CALLS: DEFAULT_NEW_CONFIG_ALLOW_ALLOW_SERVICE_CALLS,
CONF_SUBSCRIBE_LOGS: False,
}
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
assert result["data_schema"]({}) == {
CONF_ALLOW_SERVICE_CALLS: DEFAULT_NEW_CONFIG_ALLOW_ALLOW_SERVICE_CALLS,
CONF_SUBSCRIBE_LOGS: False,
}
with patch(
"homeassistant.components.esphome.async_setup_entry", return_value=True
) as mock_reload:
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_ALLOW_SERVICE_CALLS: True, CONF_SUBSCRIBE_LOGS: False},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_ALLOW_SERVICE_CALLS: True,
CONF_SUBSCRIBE_LOGS: False,
}
assert len(mock_reload.mock_calls) == 1
async def test_option_flow_subscribe_logs(
hass: HomeAssistant,
mock_client: APIClient,
mock_generic_device_entry,
) -> None:
"""Test config flow options with subscribe logs."""
entry = await mock_generic_device_entry(
mock_client=mock_client,
entity_info=[],
@ -1315,7 +1359,8 @@ async def test_option_flow(
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
assert result["data_schema"]({}) == {
CONF_ALLOW_SERVICE_CALLS: DEFAULT_NEW_CONFIG_ALLOW_ALLOW_SERVICE_CALLS
CONF_ALLOW_SERVICE_CALLS: DEFAULT_NEW_CONFIG_ALLOW_ALLOW_SERVICE_CALLS,
CONF_SUBSCRIBE_LOGS: False,
}
with patch(
@ -1323,15 +1368,16 @@ async def test_option_flow(
) as mock_reload:
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_ALLOW_SERVICE_CALLS: option_value,
},
user_input={CONF_ALLOW_SERVICE_CALLS: False, CONF_SUBSCRIBE_LOGS: True},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {CONF_ALLOW_SERVICE_CALLS: option_value}
assert len(mock_reload.mock_calls) == int(option_value)
assert result["data"] == {
CONF_ALLOW_SERVICE_CALLS: False,
CONF_SUBSCRIBE_LOGS: True,
}
assert len(mock_reload.mock_calls) == 1
@pytest.mark.usefixtures("mock_zeroconf")

62
tests/components/esphome/test_manager.py
View File

@ -2,7 +2,8 @@
import asyncio
from collections.abc import Awaitable, Callable
from unittest.mock import AsyncMock, call
import logging
from unittest.mock import AsyncMock, Mock, call
from aioesphomeapi import (
APIClient,
@ -13,6 +14,7 @@ from aioesphomeapi import (
HomeassistantServiceCall,
InvalidAuthAPIError,
InvalidEncryptionKeyAPIError,
LogLevel,
RequiresEncryptionAPIError,
UserService,
UserServiceArg,
@ -24,6 +26,7 @@ from homeassistant import config_entries
from homeassistant.components.esphome.const import (
CONF_ALLOW_SERVICE_CALLS,
CONF_DEVICE_NAME,
CONF_SUBSCRIBE_LOGS,
DOMAIN,
STABLE_BLE_VERSION_STR,
)
@ -44,6 +47,63 @@ from .conftest import MockESPHomeDevice
from tests.common import MockConfigEntry, async_capture_events, async_mock_service
async def test_esphome_device_subscribe_logs(
hass: HomeAssistant,
mock_client: APIClient,
mock_esphome_device: Callable[
[APIClient, list[EntityInfo], list[UserService], list[EntityState]],
Awaitable[MockESPHomeDevice],
],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test configuring a device to subscribe to logs."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_HOST: "fe80::1",
CONF_PORT: 6053,
CONF_PASSWORD: "",
},
options={CONF_SUBSCRIBE_LOGS: True},
)
entry.add_to_hass(hass)
device: MockESPHomeDevice = await mock_esphome_device(
mock_client=mock_client,
entry=entry,
entity_info=[],
user_service=[],
device_info={},
states=[],
)
await hass.async_block_till_done()
caplog.set_level(logging.DEBUG)
device.mock_on_log_message(
Mock(level=LogLevel.LOG_LEVEL_INFO, message=b"test_log_message")
)
await hass.async_block_till_done()
assert "test_log_message" in caplog.text
device.mock_on_log_message(
Mock(level=LogLevel.LOG_LEVEL_ERROR, message=b"test_error_log_message")
)
await hass.async_block_till_done()
assert "test_error_log_message" in caplog.text
caplog.set_level(logging.ERROR)
device.mock_on_log_message(
Mock(level=LogLevel.LOG_LEVEL_DEBUG, message=b"test_debug_log_message")
)
await hass.async_block_till_done()
assert "test_debug_log_message" not in caplog.text
caplog.set_level(logging.DEBUG)
device.mock_on_log_message(
Mock(level=LogLevel.LOG_LEVEL_DEBUG, message=b"test_debug_log_message")
)
await hass.async_block_till_done()
assert "test_debug_log_message" in caplog.text
async def test_esphome_device_service_calls_not_allowed(
hass: HomeAssistant,
mock_client: APIClient,