From c622ee6a6e13fa69271b78dc94ddefd786a8756a Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 21:58:29 -1000 Subject: [PATCH 01/13] [core] Warn on crystal frequency mismatch during serial upload When flashing an ESP32 via serial, esptool prints the detected crystal frequency. This change parses that output in real-time and warns the user if it doesn't match the configured CONFIG_XTAL_FREQ in sdkconfig. This is particularly important for ESP32-C2 (ESP8684) boards where some modules use 26MHz crystals but the default sdkconfig assumes 40MHz, causing UART logging and other clock-dependent features to silently fail. Also adds a generic line_callbacks mechanism to RedirectText so future output-based checks can be added without modifying the class directly. Co-Authored-By: Claude Opus 4.6 --- esphome/__main__.py | 54 ++++++++++++++++++++++++++++++++++++++++++++- esphome/util.py | 25 +++++++++++++++++---- 2 files changed, 74 insertions(+), 5 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index 0164e2eeb33..c9e4862e485 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -628,6 +628,50 @@ def _check_and_emit_build_info() -> None: ) +def _get_configured_xtal_freq() -> int | None: + """Read the configured crystal frequency from the sdkconfig file.""" + sdkconfig_path = CORE.relative_build_path(f"sdkconfig.{CORE.name}") + if not sdkconfig_path.is_file(): + return None + try: + content = sdkconfig_path.read_text() + for line in content.splitlines(): + if line.startswith("CONFIG_XTAL_FREQ="): + return int(line.split("=", 1)[1]) + except (OSError, ValueError): + pass + return None + + +def _make_crystal_freq_callback( + configured_freq: int, +) -> Callable[[str], str | None]: + """Create a callback that checks esptool crystal frequency output.""" + crystal_re = re.compile(r"Crystal frequency:\s+(\d+(?:\.\d+)?)\s*MHz") + + def check_crystal_line(line: str) -> str | None: + if match := crystal_re.search(line): + detected = int(float(match.group(1))) + if detected != configured_freq: + return ( + f"\n\033[33mWARNING: Crystal frequency mismatch! " + f"Device reports {detected}MHz but firmware is configured " + f"for {configured_freq}MHz.\n" + f"UART logging and other clock-dependent features will not " + f"work correctly.\n" + f"Set the correct crystal frequency with sdkconfig_options:\n" + f" esp32:\n" + f" framework:\n" + f" sdkconfig_options:\n" + f" CONFIG_XTAL_FREQ_{detected}: 'y'\n" + f" CONFIG_XTAL_FREQ_{configured_freq}: 'n'\n" + f' CONFIG_XTAL_FREQ: "{detected}"\033[0m\n\n' + ) + return None + + return check_crystal_line + + def upload_using_esptool( config: ConfigType, port: str, file: str, speed: int ) -> str | int: @@ -656,6 +700,12 @@ def upload_using_esptool( mcu = get_esp32_variant().lower() + line_callbacks = [] + if CORE.is_esp32: + configured_freq = _get_configured_xtal_freq() + if configured_freq is not None: + line_callbacks.append(_make_crystal_freq_callback(configured_freq)) + def run_esptool(baud_rate): cmd = [ "esptool", @@ -680,7 +730,9 @@ def upload_using_esptool( if os.environ.get("ESPHOME_USE_SUBPROCESS") is None: import esptool - return run_external_command(esptool.main, *cmd) # pylint: disable=no-member + return run_external_command( + esptool.main, *cmd, line_callbacks=line_callbacks + ) # pylint: disable=no-member return run_external_process(*cmd) diff --git a/esphome/util.py b/esphome/util.py index 686aa74306a..f9ef9bbbc30 100644 --- a/esphome/util.py +++ b/esphome/util.py @@ -124,7 +124,12 @@ ANSI_ESCAPE = re.compile(r"\033[@-_][0-?]*[ -/]*[@-~]") class RedirectText: - def __init__(self, out, filter_lines=None): + def __init__( + self, + out, + filter_lines: str | None = None, + line_callbacks: list[Callable[[str], str | None]] | None = None, + ) -> None: self._out = out if filter_lines is None: self._filter_pattern = None @@ -132,6 +137,7 @@ class RedirectText: pattern = r"|".join(r"(?:" + pattern + r")" for pattern in filter_lines) self._filter_pattern = re.compile(pattern) self._line_buffer = "" + self._line_callbacks = line_callbacks or [] def __getattr__(self, item): return getattr(self._out, item) @@ -180,6 +186,9 @@ class RedirectText: and (help_msg := get_esp32_arduino_flash_error_help()) ): self._write_color_replace(help_msg) + for callback in self._line_callbacks: + if msg := callback(line_without_end): + self._write_color_replace(msg) else: self._write_color_replace(s) @@ -193,7 +202,11 @@ class RedirectText: def run_external_command( - func, *cmd, capture_stdout: bool = False, filter_lines: str = None + func, + *cmd, + capture_stdout: bool = False, + filter_lines: str = None, + line_callbacks: list | None = None, ) -> int | str: """ Run a function from an external package that acts like a main method. @@ -217,9 +230,13 @@ def run_external_command( _LOGGER.debug("Running: %s", full_cmd) orig_stdout = sys.stdout - sys.stdout = RedirectText(sys.stdout, filter_lines=filter_lines) + sys.stdout = RedirectText( + sys.stdout, filter_lines=filter_lines, line_callbacks=line_callbacks + ) orig_stderr = sys.stderr - sys.stderr = RedirectText(sys.stderr, filter_lines=filter_lines) + sys.stderr = RedirectText( + sys.stderr, filter_lines=filter_lines, line_callbacks=line_callbacks + ) if capture_stdout: cap_stdout = sys.stdout = io.StringIO() From 06ac17e443ffd4ee07b4ad0e4052461022b73a15 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:00:44 -1000 Subject: [PATCH 02/13] Use early return pattern in crystal frequency check --- esphome/__main__.py | 38 ++++++++++++++++++++------------------ 1 file changed, 20 insertions(+), 18 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index c9e4862e485..e4b832655f1 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -650,24 +650,26 @@ def _make_crystal_freq_callback( crystal_re = re.compile(r"Crystal frequency:\s+(\d+(?:\.\d+)?)\s*MHz") def check_crystal_line(line: str) -> str | None: - if match := crystal_re.search(line): - detected = int(float(match.group(1))) - if detected != configured_freq: - return ( - f"\n\033[33mWARNING: Crystal frequency mismatch! " - f"Device reports {detected}MHz but firmware is configured " - f"for {configured_freq}MHz.\n" - f"UART logging and other clock-dependent features will not " - f"work correctly.\n" - f"Set the correct crystal frequency with sdkconfig_options:\n" - f" esp32:\n" - f" framework:\n" - f" sdkconfig_options:\n" - f" CONFIG_XTAL_FREQ_{detected}: 'y'\n" - f" CONFIG_XTAL_FREQ_{configured_freq}: 'n'\n" - f' CONFIG_XTAL_FREQ: "{detected}"\033[0m\n\n' - ) - return None + match = crystal_re.search(line) + if not match: + return None + detected = int(float(match.group(1))) + if detected == configured_freq: + return None + return ( + f"\n\033[33mWARNING: Crystal frequency mismatch! " + f"Device reports {detected}MHz but firmware is configured " + f"for {configured_freq}MHz.\n" + f"UART logging and other clock-dependent features will not " + f"work correctly.\n" + f"Set the correct crystal frequency with sdkconfig_options:\n" + f" esp32:\n" + f" framework:\n" + f" sdkconfig_options:\n" + f" CONFIG_XTAL_FREQ_{detected}: 'y'\n" + f" CONFIG_XTAL_FREQ_{configured_freq}: 'n'\n" + f' CONFIG_XTAL_FREQ: "{detected}"\033[0m\n\n' + ) return check_crystal_line From 0e7d4d8301e8b8529a6e539fd3d896d9c8ef002f Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:04:43 -1000 Subject: [PATCH 03/13] Fix line callbacks not firing when no filter_lines set The line buffering and callback processing only ran when a filter pattern was configured. Enter the line processing branch whenever line_callbacks are registered too. --- esphome/util.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/esphome/util.py b/esphome/util.py index f9ef9bbbc30..066f6848ad8 100644 --- a/esphome/util.py +++ b/esphome/util.py @@ -162,7 +162,7 @@ class RedirectText: if not isinstance(s, str): s = s.decode() - if self._filter_pattern is not None: + if self._filter_pattern is not None or self._line_callbacks: self._line_buffer += s lines = self._line_buffer.splitlines(True) for line in lines: @@ -174,7 +174,10 @@ class RedirectText: line_without_ansi = ANSI_ESCAPE.sub("", line) line_without_end = line_without_ansi.rstrip() - if self._filter_pattern.match(line_without_end) is not None: + if ( + self._filter_pattern is not None + and self._filter_pattern.match(line_without_end) is not None + ): # Filter pattern matched, ignore the line continue From c2747a6d35647799dd6fe0ee905743d5b811d45c Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:09:52 -1000 Subject: [PATCH 04/13] Add tests and use walrus operator for crystal freq check --- esphome/__main__.py | 8 +-- tests/unit_tests/test_main.py | 63 ++++++++++++++++++ tests/unit_tests/test_util.py | 121 ++++++++++++++++++++++++++++++++++ 3 files changed, 187 insertions(+), 5 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index e4b832655f1..4950c6d96fc 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -702,11 +702,9 @@ def upload_using_esptool( mcu = get_esp32_variant().lower() - line_callbacks = [] - if CORE.is_esp32: - configured_freq = _get_configured_xtal_freq() - if configured_freq is not None: - line_callbacks.append(_make_crystal_freq_callback(configured_freq)) + line_callbacks: list[Callable[[str], str | None]] = [] + if CORE.is_esp32 and (configured_freq := _get_configured_xtal_freq()) is not None: + line_callbacks.append(_make_crystal_freq_callback(configured_freq)) def run_esptool(baud_rate): cmd = [ diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index cef561c54b7..6c32c60ad49 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -18,6 +18,8 @@ from pytest import CaptureFixture from esphome import platformio_api from esphome.__main__ import ( Purpose, + _get_configured_xtal_freq, + _make_crystal_freq_callback, choose_upload_log_host, command_analyze_memory, command_clean_all, @@ -3297,3 +3299,64 @@ esp32: clean_output.split("SUMMARY")[1] if "SUMMARY" in clean_output else "" ) assert "secrets.yaml" not in summary_section + + +def test_get_configured_xtal_freq_reads_sdkconfig(setup_core: Path) -> None: + """Test reading XTAL_FREQ from sdkconfig.""" + CORE.name = "test-device" + CORE.build_path = setup_core + sdkconfig = setup_core / "sdkconfig.test-device" + sdkconfig.write_text( + "CONFIG_SOC_XTAL_SUPPORT_26M=y\nCONFIG_XTAL_FREQ=26\nCONFIG_XTAL_FREQ_26=y\n" + ) + assert _get_configured_xtal_freq() == 26 + + +def test_get_configured_xtal_freq_default_40(setup_core: Path) -> None: + """Test reading default 40MHz XTAL_FREQ from sdkconfig.""" + CORE.name = "test-device" + CORE.build_path = setup_core + sdkconfig = setup_core / "sdkconfig.test-device" + sdkconfig.write_text("CONFIG_XTAL_FREQ=40\nCONFIG_XTAL_FREQ_40=y\n") + assert _get_configured_xtal_freq() == 40 + + +def test_get_configured_xtal_freq_missing_file(setup_core: Path) -> None: + """Test that missing sdkconfig returns None.""" + CORE.name = "test-device" + CORE.build_path = setup_core + assert _get_configured_xtal_freq() is None + + +def test_get_configured_xtal_freq_no_xtal_line(setup_core: Path) -> None: + """Test that sdkconfig without XTAL_FREQ returns None.""" + CORE.name = "test-device" + CORE.build_path = setup_core + sdkconfig = setup_core / "sdkconfig.test-device" + sdkconfig.write_text("CONFIG_OTHER=123\n") + assert _get_configured_xtal_freq() is None + + +def test_crystal_freq_callback_mismatch() -> None: + """Test callback returns warning on crystal frequency mismatch.""" + callback = _make_crystal_freq_callback(40) + result = callback("Crystal frequency: 26MHz") + assert result is not None + assert "26MHz" in result + assert "40MHz" in result + assert "CONFIG_XTAL_FREQ_26" in result + + +def test_crystal_freq_callback_match() -> None: + """Test callback returns None when frequencies match.""" + callback = _make_crystal_freq_callback(40) + result = callback("Crystal frequency: 40MHz") + assert result is None + + +def test_crystal_freq_callback_no_crystal_line() -> None: + """Test callback returns None for unrelated lines.""" + callback = _make_crystal_freq_callback(40) + assert callback("Chip type: ESP8684H") is None + assert callback("MAC: a0:b7:65:8b:16:d4") is None + assert callback("") is None diff --git a/tests/unit_tests/test_util.py b/tests/unit_tests/test_util.py index 85873caea81..0f006c9185a 100644 --- a/tests/unit_tests/test_util.py +++ b/tests/unit_tests/test_util.py @@ -2,7 +2,9 @@ from __future__ import annotations +import io from pathlib import Path +from unittest.mock import patch import pytest @@ -402,3 +404,122 @@ def test_shlex_quote_edge_cases() -> None: assert util.shlex_quote("\t") == "'\t'" assert util.shlex_quote("\n") == "'\n'" assert util.shlex_quote(" ") == "' '" + + +def _make_redirect( + line_callbacks: list | None = None, filter_lines: list[str] | None = None +) -> tuple[util.RedirectText, io.StringIO]: + """Create a RedirectText that writes to a StringIO buffer.""" + buf = io.StringIO() + with patch("esphome.core.CORE") as mock_core: + mock_core.dashboard = False + redirect = util.RedirectText( + buf, filter_lines=filter_lines, line_callbacks=line_callbacks + ) + return redirect, buf + + +def test_redirect_text_callback_called_on_matching_line() -> None: + """Test that a line callback is called and its output is written.""" + results: list[str] = [] + + def callback(line: str) -> str | None: + results.append(line) + if "target" in line: + return "CALLBACK OUTPUT\n" + return None + + redirect, buf = _make_redirect(line_callbacks=[callback]) + redirect.write("some target line\n") + + assert "some target line" in buf.getvalue() + assert "CALLBACK OUTPUT" in buf.getvalue() + assert len(results) == 1 + + +def test_redirect_text_callback_not_triggered_on_non_matching_line() -> None: + """Test that callback returns None for non-matching lines.""" + + def callback(line: str) -> str | None: + if "target" in line: + return "FOUND\n" + return None + + redirect, buf = _make_redirect(line_callbacks=[callback]) + redirect.write("no match here\n") + + assert "no match here" in buf.getvalue() + assert "FOUND" not in buf.getvalue() + + +def test_redirect_text_callback_works_without_filter_pattern() -> None: + """Test that callbacks fire even when no filter_lines is set.""" + + def callback(line: str) -> str | None: + if "Crystal" in line: + return "WARNING: mismatch\n" + return None + + redirect, buf = _make_redirect(line_callbacks=[callback]) + redirect.write("Crystal frequency: 26MHz\n") + + assert "Crystal frequency: 26MHz" in buf.getvalue() + assert "WARNING: mismatch" in buf.getvalue() + + +def test_redirect_text_callback_works_with_filter_pattern() -> None: + """Test that callbacks fire alongside filter patterns.""" + + def callback(line: str) -> str | None: + if "important" in line: + return "NOTED\n" + return None + + redirect, buf = _make_redirect( + line_callbacks=[callback], + filter_lines=[r"^skip this.*"], + ) + redirect.write("skip this line\n") + redirect.write("important line\n") + + assert "skip this" not in buf.getvalue() + assert "important line" in buf.getvalue() + assert "NOTED" in buf.getvalue() + + +def test_redirect_text_multiple_callbacks() -> None: + """Test that multiple callbacks are all invoked.""" + + def callback_a(line: str) -> str | None: + if "test" in line: + return "FROM A\n" + return None + + def callback_b(line: str) -> str | None: + if "test" in line: + return "FROM B\n" + return None + + redirect, buf = _make_redirect(line_callbacks=[callback_a, callback_b]) + redirect.write("test line\n") + + output = buf.getvalue() + assert "FROM A" in output + assert "FROM B" in output + + +def test_redirect_text_incomplete_line_buffered() -> None: + """Test that incomplete lines are buffered until newline.""" + results: list[str] = [] + + def callback(line: str) -> str | None: + results.append(line) + return None + + redirect, buf = _make_redirect(line_callbacks=[callback]) + redirect.write("partial") + assert len(results) == 0 + + redirect.write(" line\n") + assert len(results) == 1 + assert results[0] == "partial line" From 8a50c3884436c72f68b5428525cb856b809fecd6 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:11:07 -1000 Subject: [PATCH 05/13] Use contextlib.suppress instead of try/except/pass --- esphome/__main__.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index 4950c6d96fc..cfbfc2a4b15 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -1,6 +1,7 @@ # PYTHON_ARGCOMPLETE_OK import argparse from collections.abc import Callable +from contextlib import suppress from datetime import datetime import functools import getpass @@ -633,13 +634,11 @@ def _get_configured_xtal_freq() -> int | None: sdkconfig_path = CORE.relative_build_path(f"sdkconfig.{CORE.name}") if not sdkconfig_path.is_file(): return None - try: + with suppress(OSError, ValueError): content = sdkconfig_path.read_text() for line in content.splitlines(): if line.startswith("CONFIG_XTAL_FREQ="): return int(line.split("=", 1)[1]) - except (OSError, ValueError): - pass return None From 880962aa83639fadc82f98085f1d2e8167f3db41 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:24:12 -1000 Subject: [PATCH 06/13] Address review: pass line_callbacks to run_external_process, fix filter_lines type - Pass line_callbacks through to run_external_process so the crystal frequency warning works when ESPHOME_USE_SUBPROCESS is set - Fix filter_lines type annotation from str to list[str] to match actual usage --- esphome/__main__.py | 2 +- esphome/util.py | 11 ++++++++--- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index cfbfc2a4b15..360940b3b96 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -733,7 +733,7 @@ def upload_using_esptool( esptool.main, *cmd, line_callbacks=line_callbacks ) # pylint: disable=no-member - return run_external_process(*cmd) + return run_external_process(*cmd, line_callbacks=line_callbacks) rc = run_esptool(first_baudrate) if rc == 0 or first_baudrate == 115200: diff --git a/esphome/util.py b/esphome/util.py index 066f6848ad8..029ba1793d1 100644 --- a/esphome/util.py +++ b/esphome/util.py @@ -127,7 +127,7 @@ class RedirectText: def __init__( self, out, - filter_lines: str | None = None, + filter_lines: list[str] | None = None, line_callbacks: list[Callable[[str], str | None]] | None = None, ) -> None: self._out = out @@ -273,14 +273,19 @@ def run_external_process(*cmd: str, **kwargs: Any) -> int | str: full_cmd = " ".join(shlex_quote(x) for x in cmd) _LOGGER.debug("Running: %s", full_cmd) filter_lines = kwargs.get("filter_lines") + line_callbacks = kwargs.get("line_callbacks") capture_stdout = kwargs.get("capture_stdout", False) if capture_stdout: sub_stdout = subprocess.PIPE else: - sub_stdout = RedirectText(sys.stdout, filter_lines=filter_lines) + sub_stdout = RedirectText( + sys.stdout, filter_lines=filter_lines, line_callbacks=line_callbacks + ) - sub_stderr = RedirectText(sys.stderr, filter_lines=filter_lines) + sub_stderr = RedirectText( + sys.stderr, filter_lines=filter_lines, line_callbacks=line_callbacks + ) try: proc = subprocess.run( From ae8dabc41adca202cd7f6ac380672ddc05123fe5 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:25:18 -1000 Subject: [PATCH 07/13] Add test for run_external_command with line_callbacks --- tests/unit_tests/test_util.py | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/tests/unit_tests/test_util.py b/tests/unit_tests/test_util.py index 0f006c9185a..74d18eb6777 100644 --- a/tests/unit_tests/test_util.py +++ b/tests/unit_tests/test_util.py @@ -523,3 +523,28 @@ def test_redirect_text_incomplete_line_buffered() -> None: redirect.write(" line\n") assert len(results) == 1 assert results[0] == "partial line" + + +def test_run_external_command_line_callbacks(capsys: pytest.CaptureFixture) -> None: + """Test that run_external_command passes line_callbacks to RedirectText.""" + results: list[str] = [] + + def callback(line: str) -> str | None: + results.append(line) + if "hello" in line: + return "CALLBACK FIRED\n" + return None + + def fake_main() -> int: + print("hello world") + return 0 + + with patch("esphome.core.CORE") as mock_core: + mock_core.dashboard = False + rc = util.run_external_command(fake_main, "fake", line_callbacks=[callback]) + + assert rc == 0 + assert len(results) == 1 + assert "hello world" in results[0] + captured = capsys.readouterr() + assert "CALLBACK FIRED" in captured.out From 511e47b0f52847aa180f8d7964804798716bd958 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:26:12 -1000 Subject: [PATCH 08/13] Remove unnecessary CORE mock from tests CORE.reset() is called after each test via the reset_core fixture, and CORE.dashboard defaults to False, so patching is not needed. --- tests/unit_tests/test_util.py | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) diff --git a/tests/unit_tests/test_util.py b/tests/unit_tests/test_util.py index 74d18eb6777..ff9db647a69 100644 --- a/tests/unit_tests/test_util.py +++ b/tests/unit_tests/test_util.py @@ -4,7 +4,6 @@ from __future__ import annotations import io from pathlib import Path -from unittest.mock import patch import pytest @@ -411,11 +410,9 @@ def _make_redirect( ) -> tuple[util.RedirectText, io.StringIO]: """Create a RedirectText that writes to a StringIO buffer.""" buf = io.StringIO() - with patch("esphome.core.CORE") as mock_core: - mock_core.dashboard = False - redirect = util.RedirectText( - buf, filter_lines=filter_lines, line_callbacks=line_callbacks - ) + redirect = util.RedirectText( + buf, filter_lines=filter_lines, line_callbacks=line_callbacks + ) return redirect, buf @@ -539,9 +536,7 @@ def test_run_external_command_line_callbacks(capsys: pytest.CaptureFixture) -> N print("hello world") return 0 - with patch("esphome.core.CORE") as mock_core: - mock_core.dashboard = False - rc = util.run_external_command(fake_main, "fake", line_callbacks=[callback]) + rc = util.run_external_command(fake_main, "fake", line_callbacks=[callback]) assert rc == 0 assert len(results) == 1 From dc4506453ec00984d76a6ccca5dda7b57cc2e95a Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:29:23 -1000 Subject: [PATCH 09/13] Add tests for crystal callback wiring in upload_using_esptool Tests both the in-process (run_external_command) and subprocess (run_external_process) paths to ensure line_callbacks are passed. --- tests/unit_tests/test_main.py | 61 +++++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index 6c32c60ad49..172e7a6d82c 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -6,6 +6,7 @@ from collections.abc import Generator from dataclasses import dataclass import json import logging +import os from pathlib import Path import re import time @@ -3360,3 +3361,63 @@ def test_crystal_freq_callback_no_crystal_line() -> None: assert callback("Chip type: ESP8684H") is None assert callback("MAC: a0:b7:65:8b:16:d4") is None assert callback("") is None + + +def test_upload_using_esptool_passes_crystal_callback( + tmp_path: Path, + mock_run_external_command_main: Mock, + mock_get_idedata: Mock, +) -> None: + """Test that upload_using_esptool passes crystal freq callback for ESP32.""" + setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path, name="test") + CORE.data[KEY_ESP32] = {KEY_VARIANT: VARIANT_ESP32} + + # Create sdkconfig with XTAL_FREQ + build_dir = Path(CORE.build_path) + build_dir.mkdir(parents=True, exist_ok=True) + sdkconfig = build_dir / "sdkconfig.test" + sdkconfig.write_text("CONFIG_XTAL_FREQ=40\n") + + mock_idedata = MagicMock(spec=platformio_api.IDEData) + mock_idedata.firmware_bin_path = tmp_path / "firmware.bin" + mock_idedata.extra_flash_images = [] + mock_get_idedata.return_value = mock_idedata + (tmp_path / "firmware.bin").touch() + + config = {CONF_ESPHOME: {"platformio_options": {}}} + upload_using_esptool(config, "/dev/ttyUSB0", None, None) + + # Verify line_callbacks was passed with the crystal callback + call_kwargs = mock_run_external_command_main.call_args[1] + assert "line_callbacks" in call_kwargs + assert len(call_kwargs["line_callbacks"]) == 1 + + +def test_upload_using_esptool_subprocess_passes_crystal_callback( + mock_run_external_process: Mock, + mock_get_idedata: Mock, + tmp_path: Path, +) -> None: + """Test that crystal freq callback is passed via run_external_process.""" + setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path, name="test") + CORE.data[KEY_ESP32] = {KEY_VARIANT: VARIANT_ESP32} + + # Create sdkconfig with XTAL_FREQ + build_dir = Path(CORE.build_path) + build_dir.mkdir(parents=True, exist_ok=True) + sdkconfig = build_dir / "sdkconfig.test" + sdkconfig.write_text("CONFIG_XTAL_FREQ=40\n") + + mock_idedata = MagicMock(spec=platformio_api.IDEData) + mock_idedata.firmware_bin_path = tmp_path / "firmware.bin" + mock_idedata.extra_flash_images = [] + mock_get_idedata.return_value = mock_idedata + (tmp_path / "firmware.bin").touch() + + config = {CONF_ESPHOME: {"platformio_options": {}}} + with patch.dict(os.environ, {"ESPHOME_USE_SUBPROCESS": "1"}): + upload_using_esptool(config, "/dev/ttyUSB0", None, None) + + call_kwargs = mock_run_external_process.call_args[1] + assert "line_callbacks" in call_kwargs + assert len(call_kwargs["line_callbacks"]) == 1 From 84383755d7028fde7f121235a4a57ebdba3bd128 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:30:40 -1000 Subject: [PATCH 10/13] Fix pylint disable comment placement for esptool.main --- esphome/__main__.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index 360940b3b96..0ae6aed0145 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -730,8 +730,10 @@ def upload_using_esptool( import esptool return run_external_command( - esptool.main, *cmd, line_callbacks=line_callbacks - ) # pylint: disable=no-member + esptool.main, # pylint: disable=no-member + *cmd, + line_callbacks=line_callbacks, + ) return run_external_process(*cmd, line_callbacks=line_callbacks) From d1b2010ed216a2bde2e388d06cb2342d43276c94 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:31:38 -1000 Subject: [PATCH 11/13] Use walrus operator in crystal regex match --- esphome/__main__.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index 0ae6aed0145..b216593edb7 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -649,8 +649,7 @@ def _make_crystal_freq_callback( crystal_re = re.compile(r"Crystal frequency:\s+(\d+(?:\.\d+)?)\s*MHz") def check_crystal_line(line: str) -> str | None: - match = crystal_re.search(line) - if not match: + if not (match := crystal_re.search(line)): return None detected = int(float(match.group(1))) if detected == configured_freq: From 9f72d5e428859a6f33c899deb35887ea2a59c123 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:35:33 -1000 Subject: [PATCH 12/13] Add test for run_external_process line_callbacks coverage Co-Authored-By: Claude Opus 4.6 --- tests/unit_tests/test_util.py | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/tests/unit_tests/test_util.py b/tests/unit_tests/test_util.py index ff9db647a69..47bc4b68710 100644 --- a/tests/unit_tests/test_util.py +++ b/tests/unit_tests/test_util.py @@ -4,6 +4,8 @@ from __future__ import annotations import io from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch import pytest @@ -543,3 +545,36 @@ def test_run_external_command_line_callbacks(capsys: pytest.CaptureFixture) -> N assert "hello world" in results[0] captured = capsys.readouterr() assert "CALLBACK FIRED" in captured.out + + +def test_run_external_process_line_callbacks() -> None: + """Test that run_external_process passes line_callbacks to RedirectText.""" + results: list[str] = [] + + def callback(line: str) -> str | None: + results.append(line) + if "from subprocess" in line: + return "PROCESS CALLBACK\n" + return None + + with patch("subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0) + + # Capture the RedirectText objects passed to subprocess.run + def run_side_effect(*args: Any, **kwargs: Any) -> MagicMock: + # Simulate subprocess writing to the stdout RedirectText + stdout = kwargs.get("stdout") + if stdout is not None and isinstance(stdout, util.RedirectText): + stdout.write("from subprocess\n") + return MagicMock(returncode=0) + + mock_run.side_effect = run_side_effect + + rc = util.run_external_process( + "echo", + "test", + line_callbacks=[callback], + ) + + assert rc == 0 + assert any("from subprocess" in r for r in results) From 696c0f021c0acf7a623b549e6a39b8042093f0dc Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:38:44 -1000 Subject: [PATCH 13/13] Patch at import point instead of subprocess.run Co-Authored-By: Claude Opus 4.6 --- tests/unit_tests/test_util.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/unit_tests/test_util.py b/tests/unit_tests/test_util.py index 47bc4b68710..5a159ff7bbf 100644 --- a/tests/unit_tests/test_util.py +++ b/tests/unit_tests/test_util.py @@ -557,10 +557,8 @@ def test_run_external_process_line_callbacks() -> None: return "PROCESS CALLBACK\n" return None - with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock(returncode=0) + with patch("esphome.util.subprocess.run") as mock_run: - # Capture the RedirectText objects passed to subprocess.run def run_side_effect(*args: Any, **kwargs: Any) -> MagicMock: # Simulate subprocess writing to the stdout RedirectText stdout = kwargs.get("stdout")