Add tests and use walrus operator for crystal freq check

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 = [

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

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"