Merge remote-tracking branch 'upstream/dev' into integration

tests/components/dew_point/common.yaml

@@ -0,0 +1,19 @@
+sensor:
+  - platform: dew_point
+    name: Dew Point
+    temperature: template_temperature
+    humidity: template_humidity
+  - platform: template
+    id: template_humidity
+    lambda: |-
+      if (millis() > 10000) {
+        return 0.6;
+      }
+      return 0.0;
+  - platform: template
+    id: template_temperature
+    lambda: |-
+      if (millis() > 10000) {
+        return 42.0;
+      }
+      return 0.0;

tests/components/dew_point/test.esp32-idf.yaml

@@ -0,0 +1 @@
+<<: !include common.yaml

tests/components/dew_point/test.esp8266-ard.yaml

@@ -0,0 +1 @@
+<<: !include common.yaml

tests/components/dew_point/test.rp2040-ard.yaml

@@ -0,0 +1 @@
+<<: !include common.yaml

tests/unit_tests/test_main.py

@@ -3722,38 +3722,38 @@ esp32:
     assert "secrets.yaml" not in summary_section
 
 
-def test_get_configured_xtal_freq_reads_sdkconfig(setup_core: Path) -> None:
+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 = setup_core
-    sdkconfig = setup_core / "sdkconfig.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(setup_core: Path) -> None:
+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 = setup_core
-    sdkconfig = setup_core / "sdkconfig.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(setup_core: Path) -> None:
+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 = setup_core
+    CORE.build_path = tmp_path
     assert _get_configured_xtal_freq() is None
 
 
-def test_get_configured_xtal_freq_no_xtal_line(setup_core: Path) -> 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 = setup_core
-    sdkconfig = setup_core / "sdkconfig.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
 

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