[core] Warn on crystal frequency mismatch during serial upload (#14582)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

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 sys
@@ -19,6 +20,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,
@@ -3717,3 +3720,124 @@ 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(tmp_path: Path) -> None:
+    """Test reading XTAL_FREQ from sdkconfig."""
+    CORE.name = "test-device"
+    CORE.build_path = tmp_path
+    sdkconfig = tmp_path / "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(tmp_path: Path) -> None:
+    """Test reading default 40MHz XTAL_FREQ from sdkconfig."""
+    CORE.name = "test-device"
+    CORE.build_path = tmp_path
+    sdkconfig = tmp_path / "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(tmp_path: Path) -> None:
+    """Test that missing sdkconfig returns None."""
+    CORE.name = "test-device"
+    CORE.build_path = tmp_path
+    assert _get_configured_xtal_freq() is None
+
+
+def test_get_configured_xtal_freq_no_xtal_line(tmp_path: Path) -> None:
+    """Test that sdkconfig without XTAL_FREQ returns None."""
+    CORE.name = "test-device"
+    CORE.build_path = tmp_path
+    sdkconfig = tmp_path / "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
+
+
+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

tests/unit_tests/test_util.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
+import io
 from pathlib import Path
 import subprocess
 import sys
+from typing import Any
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -407,6 +410,178 @@ def test_shlex_quote_edge_cases() -> None:
     assert util.shlex_quote("   ") == "'   '"
 
 
+def _make_redirect(
+    line_callbacks: list[Callable[[str], str | None]] | 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()
+    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"
+
+
+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
+
+    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
+
+
+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("esphome.util.subprocess.run") as mock_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)
+
+
 def test_get_picotool_path_found(tmp_path: Path) -> None:
     """Test picotool path derivation from cc_path."""
     # Create the expected directory structure