diff --git a/esphome/__main__.py b/esphome/__main__.py index e56b504398..3893c678e1 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -1429,7 +1429,7 @@ def show_logs(config: ConfigType, args: ArgsProtocol, devices: list[str]) -> int if has_api() and ( network_devices := _resolve_network_devices(devices, config, args) ): - from esphome.components.api.client import run_logs + from esphome.api_client import run_logs return run_logs( config, diff --git a/esphome/api_client.py b/esphome/api_client.py new file mode 100644 index 0000000000..0ee2a7bed3 --- /dev/null +++ b/esphome/api_client.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import asyncio +from contextlib import suppress +from datetime import datetime +import importlib +import logging +from typing import TYPE_CHECKING, Any +import warnings + +# Suppress protobuf version warnings +with warnings.catch_warnings(): + warnings.filterwarnings( + "ignore", category=UserWarning, message=".*Protobuf gencode version.*" + ) + from aioesphomeapi import APIClient, parse_log_message + from aioesphomeapi.log_runner import async_run + +from esphome.const import CONF_ENCRYPTION, CONF_KEY, CONF_PORT, __version__ +from esphome.core import CORE +from esphome.util import safe_print + +if TYPE_CHECKING: + from aioesphomeapi.api_pb2 import ( + SubscribeLogsResponse, # pylint: disable=no-name-in-module + ) + + +_LOGGER = logging.getLogger(__name__) + + +class _LogLineProcessor: + """Feeds incoming log lines to the stack-trace decoder. + + Two responsibilities beyond just calling the decoder: + 1. Catch everything the decoder can raise. aioesphomeapi isolates + exceptions raised by log handlers, so an escaping one no longer + kills the session, but it does log a full traceback per line. A + crash dump carries a PC line plus one per backtrace frame, so the + tracebacks bury the dump the user is trying to read. Decoding is a + diagnostic nicety; nothing it raises is worth that noise. + 2. Disable decoding after the first failure. _decode_pc shells out to + the toolchain to resolve addr2line, which is expensive; a single + crash dump can contain many PC/BT lines and we don't want to retry + the failing subprocess for each one. This only works if every + failure is caught, which is why 1 is not narrowed to EsphomeError. + """ + + def __init__(self, config: dict[str, Any], platform_handler: Any | None) -> None: + self._config = config + self._platform_handler = platform_handler + self._decode_enabled = platform_handler is not None + self.backtrace_state = False + + def process_line(self, raw_line: str) -> None: + if not self._decode_enabled: + return + try: + self.backtrace_state = self._platform_handler( + self._config, raw_line, self.backtrace_state + ) + except Exception as exc: # noqa: BLE001 # pylint: disable=broad-except + self._decode_enabled = False + self.backtrace_state = False + # _run_idedata raises EsphomeError with no message; fall back + # to a generic explanation when str(exc) is empty. + detail = str(exc) or "build artifacts not found locally" + _LOGGER.debug("Stack-trace decoding failed", exc_info=True) + _LOGGER.warning( + "Crash trace decoding unavailable: %s. " + "Run 'esphome compile' for this device to enable PC decoding.", + detail, + ) + + +async def async_run_logs( + config: dict[str, Any], + addresses: list[str], + subscribe_states: bool = True, +) -> None: + """Run the logs command in the event loop.""" + conf = config["api"] + name = config["esphome"]["name"] + port: int = int(conf[CONF_PORT]) + noise_psk: str | None = None + if (encryption := conf.get(CONF_ENCRYPTION)) and (key := encryption.get(CONF_KEY)): + noise_psk = key + + _LOGGER.info( + "Starting log output from %s using esphome API", " or ".join(addresses) + ) + + cli = APIClient( + addresses[0], # Primary address for compatibility + port, + "", # Password auth removed in 2026.1.0 + client_info=f"ESPHome Logs {__version__}", + noise_psk=noise_psk, + addresses=addresses, # Pass all addresses for automatic retry + provide_time=False, + ) + + # Try platform-specific stacktrace handler first, fall back to generic + platform_process_stacktrace = None + try: + module = importlib.import_module("esphome.components." + CORE.target_platform) + platform_process_stacktrace = module.process_stacktrace + except (AttributeError, ImportError): + # Distinguish "platform has no analyzer" from a genuinely broken + # platform package when debugging. + _LOGGER.debug("Stacktrace analyzer lookup failed", exc_info=True) + _LOGGER.info( + 'Stacktrace analysis is unavailable: no compatible analyzer found for target platform "%s".', + CORE.target_platform, + ) + + processor = _LogLineProcessor(config, platform_process_stacktrace) + + def on_log(msg: SubscribeLogsResponse) -> None: + """Handle a new log message.""" + time_ = datetime.now().astimezone() + message: bytes = msg.message + text = message.decode("utf8", "backslashreplace") + nanoseconds = time_.microsecond // 1000 + timestamp = ( + f"[{time_.hour:02}:{time_.minute:02}:{time_.second:02}.{nanoseconds:03}]" + ) + for parsed_msg in parse_log_message(text, timestamp): + # safe_print handles the dashboard \033 escaping and falls back + # to backslashreplace encoding on stdouts that can't represent + # the wifi signal-bar block characters (Windows redirected + # cp1252 pipe). + safe_print(parsed_msg) + for raw_line in text.splitlines(): + processor.process_line(raw_line) + + # Safe to fall back to plaintext here only for this diagnostics use + # case: the stream is one-way from device to client, and this code + # never accepts commands or acts on any message the device sends. + # An on-path attacker could still both inject fabricated log lines + # and passively read the device's log output (and any state data + # delivered when subscribe_states is enabled), so this does lose + # confidentiality as well as authentication/integrity. That tradeoff + # is acceptable for operator-visible logs, which aioesphomeapi also + # warns may come from an unverified device. Never mirror this opt-in + # for any connection that sends data to the device or uses Home + # Assistant actions. + stop = await async_run( + cli, + on_log, + name=name, + subscribe_states=subscribe_states, + allow_plaintext_fallback=True, + # A top-level ``deep_sleep:`` block means the device is only awake + # briefly; cap the reconnect backoff so a wake window is not missed. + deep_sleep="deep_sleep" in config, + ) + try: + await asyncio.Event().wait() + finally: + await stop() + + +def run_logs( + config: dict[str, Any], + addresses: list[str], + subscribe_states: bool = True, +) -> None: + """Run the logs command.""" + with suppress(KeyboardInterrupt): + asyncio.run( + async_run_logs(config, addresses, subscribe_states=subscribe_states) + ) diff --git a/esphome/components/api/__init__.py b/esphome/components/api/__init__.py index 0719cee352..8ec94df1db 100644 --- a/esphome/components/api/__init__.py +++ b/esphome/components/api/__init__.py @@ -13,6 +13,7 @@ from esphome.const import ( CONF_CAPTURE_RESPONSE, CONF_DATA, CONF_DATA_TEMPLATE, + CONF_ENCRYPTION, CONF_EVENT, CONF_ID, CONF_KEY, @@ -102,7 +103,6 @@ SERVICE_ARG_FALLBACK_TYPES: dict[str, MockObj] = { for name, t in _SERVICE_ARG_SCALAR_TYPES.items() }, } -CONF_ENCRYPTION = "encryption" CONF_BATCH_DELAY = "batch_delay" CONF_CUSTOM_SERVICES = "custom_services" CONF_HOMEASSISTANT_SERVICES = "homeassistant_services" diff --git a/esphome/components/api/client.py b/esphome/components/api/client.py index 7f07146dba..5e1c88b2ca 100644 --- a/esphome/components/api/client.py +++ b/esphome/components/api/client.py @@ -1,177 +1,10 @@ -from __future__ import annotations +"""Backward-compatibility shim; the log client lives in esphome.api_client. -import asyncio -from datetime import datetime -import importlib -import logging -from typing import TYPE_CHECKING, Any -import warnings +Importing this module executes the whole api component package, which pulls +in the validation stack. CLI code paths should import esphome.api_client +directly so the logs fast path stays light. +""" -# Suppress protobuf version warnings -with warnings.catch_warnings(): - warnings.filterwarnings( - "ignore", category=UserWarning, message=".*Protobuf gencode version.*" - ) - from aioesphomeapi import APIClient, parse_log_message - from aioesphomeapi.log_runner import async_run +from esphome.api_client import async_run_logs, run_logs -import contextlib - -from esphome.const import CONF_KEY, CONF_PORT, __version__ -from esphome.core import CORE -from esphome.util import safe_print - -from . import CONF_ENCRYPTION - -if TYPE_CHECKING: - from aioesphomeapi.api_pb2 import ( - SubscribeLogsResponse, # pylint: disable=no-name-in-module - ) - - -_LOGGER = logging.getLogger(__name__) - - -class _LogLineProcessor: - """Feeds incoming log lines to the stack-trace decoder. - - Two responsibilities beyond just calling the decoder: - 1. Catch everything the decoder can raise. aioesphomeapi isolates - exceptions raised by log handlers, so an escaping one no longer - kills the session, but it does log a full traceback per line. A - crash dump carries a PC line plus one per backtrace frame, so the - tracebacks bury the dump the user is trying to read. Decoding is a - diagnostic nicety; nothing it raises is worth that noise. - 2. Disable decoding after the first failure. _decode_pc shells out to - the toolchain to resolve addr2line, which is expensive; a single - crash dump can contain many PC/BT lines and we don't want to retry - the failing subprocess for each one. This only works if every - failure is caught, which is why 1 is not narrowed to EsphomeError. - """ - - def __init__(self, config: dict[str, Any], platform_handler: Any | None) -> None: - self._config = config - self._platform_handler = platform_handler - self._decode_enabled = True - self.backtrace_state = False - - def process_line(self, raw_line: str) -> None: - if not self._decode_enabled: - return - try: - if self._platform_handler is not None: - self.backtrace_state = self._platform_handler( - self._config, raw_line, self.backtrace_state - ) - except Exception as exc: # noqa: BLE001 # pylint: disable=broad-except - self._decode_enabled = False - self.backtrace_state = False - # _run_idedata raises EsphomeError with no message; fall back - # to a generic explanation when str(exc) is empty. - detail = str(exc) or "build artifacts not found locally" - _LOGGER.debug("Stack-trace decoding failed", exc_info=True) - _LOGGER.warning( - "Crash trace decoding unavailable: %s. " - "Run 'esphome compile' for this device to enable PC decoding.", - detail, - ) - - -async def async_run_logs( - config: dict[str, Any], - addresses: list[str], - subscribe_states: bool = True, -) -> None: - """Run the logs command in the event loop.""" - conf = config["api"] - name = config["esphome"]["name"] - port: int = int(conf[CONF_PORT]) - noise_psk: str | None = None - if (encryption := conf.get(CONF_ENCRYPTION)) and (key := encryption.get(CONF_KEY)): - noise_psk = key - - if len(addresses) == 1: - _LOGGER.info("Starting log output from %s using esphome API", addresses[0]) - else: - _LOGGER.info( - "Starting log output from %s using esphome API", " or ".join(addresses) - ) - - cli = APIClient( - addresses[0], # Primary address for compatibility - port, - "", # Password auth removed in 2026.1.0 - client_info=f"ESPHome Logs {__version__}", - noise_psk=noise_psk, - addresses=addresses, # Pass all addresses for automatic retry - provide_time=False, - ) - - # Try platform-specific stacktrace handler first, fall back to generic - platform_process_stacktrace = None - try: - module = importlib.import_module("esphome.components." + CORE.target_platform) - platform_process_stacktrace = module.process_stacktrace - except (AttributeError, ImportError): - _LOGGER.info( - 'Stacktrace analysis is unavailable: no compatible analyzer found for target platform "%s".', - CORE.target_platform, - ) - - processor = _LogLineProcessor(config, platform_process_stacktrace) - - def on_log(msg: SubscribeLogsResponse) -> None: - """Handle a new log message.""" - time_ = datetime.now().astimezone() - message: bytes = msg.message - text = message.decode("utf8", "backslashreplace") - nanoseconds = time_.microsecond // 1000 - timestamp = ( - f"[{time_.hour:02}:{time_.minute:02}:{time_.second:02}.{nanoseconds:03}]" - ) - for parsed_msg in parse_log_message(text, timestamp): - # safe_print handles the dashboard \033 escaping and falls back - # to backslashreplace encoding on stdouts that can't represent - # the wifi signal-bar block characters (Windows redirected - # cp1252 pipe). - safe_print(parsed_msg) - for raw_line in text.splitlines(): - processor.process_line(raw_line) - - # Safe to fall back to plaintext here only for this diagnostics use - # case: the stream is one-way from device to client, and this code - # never accepts commands or acts on any message the device sends. - # An on-path attacker could still both inject fabricated log lines - # and passively read the device's log output (and any state data - # delivered when subscribe_states is enabled), so this does lose - # confidentiality as well as authentication/integrity. That tradeoff - # is acceptable for operator-visible logs, which aioesphomeapi also - # warns may come from an unverified device. Never mirror this opt-in - # for any connection that sends data to the device or uses Home - # Assistant actions. - stop = await async_run( - cli, - on_log, - name=name, - subscribe_states=subscribe_states, - allow_plaintext_fallback=True, - # A top-level ``deep_sleep:`` block means the device is only awake - # briefly; cap the reconnect backoff so a wake window is not missed. - deep_sleep="deep_sleep" in config, - ) - try: - await asyncio.Event().wait() - finally: - await stop() - - -def run_logs( - config: dict[str, Any], - addresses: list[str], - subscribe_states: bool = True, -) -> None: - """Run the logs command.""" - with contextlib.suppress(KeyboardInterrupt): - asyncio.run( - async_run_logs(config, addresses, subscribe_states=subscribe_states) - ) +__all__ = ["async_run_logs", "run_logs"] diff --git a/esphome/components/packet_transport/__init__.py b/esphome/components/packet_transport/__init__.py index 4293dffb15..7beb13ca31 100644 --- a/esphome/components/packet_transport/__init__.py +++ b/esphome/components/packet_transport/__init__.py @@ -4,12 +4,12 @@ import hashlib import logging import esphome.codegen as cg -from esphome.components.api import CONF_ENCRYPTION from esphome.components.binary_sensor import BinarySensor from esphome.components.sensor import Sensor import esphome.config_validation as cv from esphome.const import ( CONF_BINARY_SENSORS, + CONF_ENCRYPTION, CONF_ID, CONF_INTERNAL, CONF_KEY, diff --git a/esphome/components/udp/packet_transport/__init__.py b/esphome/components/udp/packet_transport/__init__.py index b6957a372b..e725276717 100644 --- a/esphome/components/udp/packet_transport/__init__.py +++ b/esphome/components/udp/packet_transport/__init__.py @@ -1,12 +1,11 @@ import esphome.codegen as cg -from esphome.components.api import CONF_ENCRYPTION from esphome.components.packet_transport import ( CONF_PING_PONG_ENABLE, PacketTransport, new_packet_transport, transport_schema, ) -from esphome.const import CONF_BINARY_SENSORS, CONF_SENSORS +from esphome.const import CONF_BINARY_SENSORS, CONF_ENCRYPTION, CONF_SENSORS from esphome.cpp_types import PollingComponent from .. import UDP_SCHEMA, register_udp_client, udp_ns diff --git a/esphome/const.py b/esphome/const.py index a7306edeb0..f2d305dced 100644 --- a/esphome/const.py +++ b/esphome/const.py @@ -384,6 +384,7 @@ CONF_ENABLE_PIN = "enable_pin" CONF_ENABLE_PRIVATE_NETWORK_ACCESS = "enable_private_network_access" CONF_ENABLE_RRM = "enable_rrm" CONF_ENABLE_TIME = "enable_time" +CONF_ENCRYPTION = "encryption" CONF_ENERGY = "energy" CONF_ENTITY_CATEGORY = "entity_category" CONF_ENTITY_ID = "entity_id" diff --git a/script/ci-custom.py b/script/ci-custom.py index 90748a13b9..9f3d836f65 100755 --- a/script/ci-custom.py +++ b/script/ci-custom.py @@ -557,7 +557,7 @@ def lint_constants_usage(): # Maximum allowed CONF_ constants in esphome/const.py. # This file is frozen — new constants go in esphome/components/const/__init__.py. # Decrease this number when constants are moved out of const.py. -CONST_PY_MAX_CONF = 1015 +CONST_PY_MAX_CONF = 1016 @lint_content_check(include=["esphome/const.py"]) diff --git a/tests/unit_tests/components/api/__init__.py b/tests/unit_tests/components/api/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/tests/unit_tests/components/api/test_client.py b/tests/unit_tests/test_api_client.py similarity index 64% rename from tests/unit_tests/components/api/test_client.py rename to tests/unit_tests/test_api_client.py index 4ebcecbfff..55907f282c 100644 --- a/tests/unit_tests/components/api/test_client.py +++ b/tests/unit_tests/test_api_client.py @@ -1,17 +1,34 @@ -"""Tests for esphome.components.api.client.""" +"""Tests for esphome.api_client.""" from __future__ import annotations -from unittest.mock import AsyncMock, patch +import asyncio +from unittest.mock import AsyncMock, Mock, patch import pytest +from esphome import api_client from esphome.components import esp32 -from esphome.components.api import client as api_client -from esphome.const import CONF_PORT, KEY_CORE, KEY_TARGET_PLATFORM +from esphome.const import ( + CONF_ENCRYPTION, + CONF_KEY, + CONF_PORT, + KEY_CORE, + KEY_TARGET_PLATFORM, +) from esphome.core import CORE, EsphomeError +def test_component_shim_reexports_runtime_client() -> None: + """The old import paths must keep working for external code.""" + from esphome.components import api + from esphome.components.api import client as shim + + assert shim.run_logs is api_client.run_logs + assert shim.async_run_logs is api_client.async_run_logs + assert api.CONF_ENCRYPTION is CONF_ENCRYPTION + + def test_decoder_swallows_esphome_error() -> None: """A failing stack-trace decode must not propagate. @@ -166,3 +183,73 @@ async def test_async_run_logs_passes_deep_sleep( await api_client.async_run_logs(config, ["1.2.3.4"]) assert mock_run.call_args.kwargs["deep_sleep"] is expected_deep_sleep + + +@pytest.mark.asyncio +async def test_async_run_logs_full_flow(caplog) -> None: + """Drive async_run_logs end to end with a fake connection. + + Covers the encryption key extraction, the multi-address banner, the + missing-stacktrace-analyzer fallback, the on_log handler, and the + stop() cleanup in the finally block. + """ + caplog.set_level("INFO", logger="esphome.api_client") + CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "host"} + config = { + "esphome": {"name": "test"}, + "api": {CONF_PORT: 6053, CONF_ENCRYPTION: {CONF_KEY: "psk123"}}, + } + + stop = AsyncMock() + run_started = asyncio.Event() + + async def fake_async_run(*args, **kwargs): + run_started.set() + return stop + + mock_run = AsyncMock(side_effect=fake_async_run) + printed: list[str] = [] + + with ( + patch.object(api_client, "async_run", mock_run), + patch.object(api_client, "APIClient") as mock_client, + patch.object(api_client, "safe_print", printed.append), + ): + task = asyncio.get_running_loop().create_task( + api_client.async_run_logs(config, ["1.2.3.4", "5.6.7.8"]) + ) + # Let the task run up to the forever-wait; the timeout fails the + # test instead of hanging it if the task dies early. + async with asyncio.timeout(1): + await run_started.wait() + on_log = mock_run.call_args.args[1] + on_log(Mock(message=b"[I][main:001] hello world\nPC: 0x40104960")) + # Cancellation is the real termination path; stop() must still run. + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + # Both addresses reach APIClient, along with the noise key. + assert mock_client.call_args.kwargs["noise_psk"] == "psk123" + assert mock_client.call_args.kwargs["addresses"] == ["1.2.3.4", "5.6.7.8"] + assert "1.2.3.4 or 5.6.7.8" in caplog.text + # host has no stacktrace analyzer; the fallback message is logged. + assert "Stacktrace analysis is unavailable" in caplog.text + # The log message was printed with a timestamp prefix. + assert any("hello world" in line for line in printed) + # stop() ran in the finally block despite the cancellation. + stop.assert_awaited_once() + + +def test_run_logs_suppresses_keyboard_interrupt() -> None: + """Ctrl-C during log streaming exits cleanly instead of tracebacking.""" + with patch.object( + api_client, + "async_run_logs", + AsyncMock(side_effect=KeyboardInterrupt), + ) as mock_run: + api_client.run_logs( + {"esphome": {"name": "test"}}, ["1.2.3.4"], subscribe_states=False + ) + + assert mock_run.call_args.kwargs["subscribe_states"] is False diff --git a/tests/unit_tests/test_lazy_imports.py b/tests/unit_tests/test_lazy_imports.py index ee570a84f6..72ba73c9dc 100644 --- a/tests/unit_tests/test_lazy_imports.py +++ b/tests/unit_tests/test_lazy_imports.py @@ -31,11 +31,16 @@ HEAVY_MODULES = ( ) -def test_main_module_does_not_import_heavy_modules() -> None: - """A bare ``import esphome.__main__`` must not drag in validation/codegen.""" +def _leaked_heavy_modules(module: str) -> str: + """Import ``module`` in a subprocess and report the heavy modules it pulled. + + Any ``esphome.components.*`` package counts as heavy: executing a + component package drags in codegen/validation machinery by design. + """ check = ( - "import sys; import esphome.__main__; " + f"import sys; import {module}; " f"leaked = [m for m in {HEAVY_MODULES!r} if m in sys.modules]; " + "leaked += [m for m in sys.modules if m.startswith('esphome.components.')]; " "print(','.join(leaked))" ) result = subprocess.run( @@ -44,10 +49,30 @@ def test_main_module_does_not_import_heavy_modules() -> None: text=True, check=True, ) - leaked = result.stdout.strip() + return result.stdout.strip() + + +def test_main_module_does_not_import_heavy_modules() -> None: + """A bare ``import esphome.__main__`` must not drag in validation/codegen.""" + leaked = _leaked_heavy_modules("esphome.__main__") assert not leaked, ( f"esphome.__main__ imports heavy modules at top level: {leaked}. " "Import them lazily inside the command that needs them instead; " "every esphome invocation (including each parallel dashboard " "upload subprocess) pays for top-level imports." ) + + +def test_api_client_does_not_import_heavy_modules() -> None: + """``esphome.api_client`` is on the logs fast path and must stay light. + + Importing it must not execute any component package (the api package + pulls the whole validation stack: logger, esp32, writer, config, + jinja2, voluptuous). + """ + leaked = _leaked_heavy_modules("esphome.api_client") + assert not leaked, ( + f"esphome.api_client imports heavy modules at top level: {leaked}. " + "The logs fast path skips validation; importing the validation " + "stack anyway defeats the validated-config cache." + ) diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index e575934870..289497057f 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -2918,7 +2918,7 @@ def test_show_logs_no_logger() -> None: show_logs(CORE.config, args, devices) -@patch("esphome.components.api.client.run_logs") +@patch("esphome.api_client.run_logs") def test_show_logs_api( mock_run_logs: Mock, ) -> None: @@ -2944,7 +2944,7 @@ def test_show_logs_api( ) -@patch("esphome.components.api.client.run_logs") +@patch("esphome.api_client.run_logs") def test_show_logs_api_no_states( mock_run_logs: Mock, ) -> None: @@ -2971,7 +2971,7 @@ def test_show_logs_api_no_states( ) -@patch("esphome.components.api.client.run_logs") +@patch("esphome.api_client.run_logs") def test_show_logs_api_with_fqdn_mdns_disabled( mock_run_logs: Mock, ) -> None: @@ -2998,7 +2998,7 @@ def test_show_logs_api_with_fqdn_mdns_disabled( ) -@patch("esphome.components.api.client.run_logs") +@patch("esphome.api_client.run_logs") def test_show_logs_api_with_mqtt_fallback( mock_run_logs: Mock, mock_mqtt_get_ip: Mock, @@ -4974,7 +4974,7 @@ def test_upload_program_ota_mqttip_deduplication( assert "192.168.1.100" in call_args[0] -@patch("esphome.components.api.client.run_logs") +@patch("esphome.api_client.run_logs") def test_show_logs_api_static_ip_with_mqttip( mock_run_logs: Mock, mock_mqtt_get_ip: Mock, @@ -5013,7 +5013,7 @@ def test_show_logs_api_static_ip_with_mqttip( ) -@patch("esphome.components.api.client.run_logs") +@patch("esphome.api_client.run_logs") def test_show_logs_api_multiple_mqttip_resolves_once( mock_run_logs: Mock, mock_mqtt_get_ip: Mock, @@ -5096,7 +5096,7 @@ def test_upload_program_ota_mqtt_timeout_fallback( ) -@patch("esphome.components.api.client.run_logs") +@patch("esphome.api_client.run_logs") def test_show_logs_api_mqtt_timeout_fallback( mock_run_logs: Mock, mock_mqtt_get_ip: Mock, @@ -6468,7 +6468,7 @@ def test_should_subscribe_states_no_flag_overrides_env() -> None: assert _should_subscribe_states(args) is False -@patch("esphome.components.api.client.run_logs") +@patch("esphome.api_client.run_logs") def test_command_run_passes_no_states_to_show_logs( mock_run_logs: Mock, ) -> None: @@ -6506,7 +6506,7 @@ def test_command_run_passes_no_states_to_show_logs( ) -@patch("esphome.components.api.client.run_logs") +@patch("esphome.api_client.run_logs") def test_command_run_defaults_subscribe_states_true( mock_run_logs: Mock, ) -> None: