From c2747a6d35647799dd6fe0ee905743d5b811d45c Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 6 Mar 2026 22:09:52 -1000 Subject: [PATCH] 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"