[core] Skip colorama init for terminal and dashboard runs (#18224)

esphome/__main__.py

@@ -1941,7 +1941,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
     new_name = args.name
     for c in new_name:
         if c not in ALLOWED_NAME_CHARS:
-            print(
+            safe_print(
                 color(
                     AnsiFore.BOLD_RED,
                     f"'{c}' is an invalid character for names. Valid characters are: "
@@ -1954,7 +1954,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
 
     yaml = yaml_util.load_yaml(CORE.config_path)
     if CONF_ESPHOME not in yaml or CONF_NAME not in yaml[CONF_ESPHOME]:
-        print(
+        safe_print(
             color(
                 AnsiFore.BOLD_RED, "Complex YAML files cannot be automatically renamed."
             )
@@ -2001,7 +2001,9 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
             )
             > 1
         ):
-            print(color(AnsiFore.BOLD_RED, "Too many matches in YAML to safely rename"))
+            safe_print(
+                color(AnsiFore.BOLD_RED, "Too many matches in YAML to safely rename")
+            )
             return 1
 
         new_raw = re.sub(
@@ -2019,7 +2021,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
     # ``kitchen``; running ``esphome rename weird-file.yaml kitchen``
     # would otherwise just re-flash the same hostname).
     if new_name == old_name:
-        print(
+        safe_print(
             color(
                 AnsiFore.BOLD_RED,
                 f"'{new_name}' is already the device's name.",
@@ -2029,7 +2031,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
 
     new_path: Path = CORE.config_dir / (new_name + ".yaml")
     if new_path.resolve() == CORE.config_path.resolve():
-        print(
+        safe_print(
             color(
                 AnsiFore.BOLD_RED,
                 f"'{new_name}' is already the device's name.",
@@ -2037,7 +2039,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
         )
         return 1
     if new_path.exists():
-        print(
+        safe_print(
             color(
                 AnsiFore.BOLD_RED,
                 f"Cannot rename: {new_path} already exists. "
@@ -2045,7 +2047,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
             )
         )
         return 1
-    print(
+    safe_print(
         f"Updating {color(AnsiFore.CYAN, str(CORE.config_path))} to {color(AnsiFore.CYAN, str(new_path))}"
     )
     print()
@@ -2054,7 +2056,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
 
     rc = run_external_process(*ESPHOME_COMMAND, "config", str(new_path))
     if rc != 0:
-        print(color(AnsiFore.BOLD_RED, "Rename failed. Reverting changes."))
+        safe_print(color(AnsiFore.BOLD_RED, "Rename failed. Reverting changes."))
         new_path.unlink()
         return 1
 
@@ -2080,7 +2082,7 @@ def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
     if CORE.config_path != new_path:
         CORE.config_path.unlink()
 
-    print(color(AnsiFore.BOLD_GREEN, "SUCCESS"))
+    safe_print(color(AnsiFore.BOLD_GREEN, "SUCCESS"))
     print()
     return 0
 

esphome/log.py

@@ -1,5 +1,7 @@
 from enum import Enum
 import logging
+import sys
+from typing import TextIO
 
 from esphome.core import CORE
 
@@ -72,10 +74,27 @@ class ESPHomeLogFormatter(logging.Formatter):
         return message
 
 
+def _is_tty(stream: TextIO | None) -> bool:
+    # A stream can be missing, closed, or not a real file object; colorama
+    # tolerates all three, so treat them like a redirect and let its own
+    # handling apply.
+    if stream is None or getattr(stream, "closed", True):
+        return False
+    return hasattr(stream, "isatty") and stream.isatty()
+
+
 def setup_log(
     log_level: int = logging.INFO,
     include_timestamp: bool = False,
 ) -> None:
+    # colorama translates ANSI escapes for old Windows consoles and strips
+    # them from redirected output. POSIX terminals render ANSI natively, and
+    # dashboard runs escape their color codes before printing, so both would
+    # use colorama as a plain passthrough; skip the import there (it pulls
+    # in ctypes, ~3ms on every CLI invocation).
+    if sys.platform == "win32" or not (
+        CORE.dashboard or (_is_tty(sys.stdout) and _is_tty(sys.stderr))
+    ):
         import colorama
 
         colorama.init()

tests/unit_tests/conftest.py

@@ -10,6 +10,7 @@ not be part of a unit test suite.
 """
 
 from collections.abc import Generator
+import os
 from pathlib import Path
 import sys
 from unittest.mock import Mock, patch
@@ -40,6 +41,19 @@ def fixture_path() -> Path:
     return here / "fixtures"
 
 
+@pytest.fixture
+def probe_env() -> dict[str, str]:
+    """Environment for running fixture probe scripts as subprocesses.
+
+    Running a script file drops the cwd from sys.path, so prepend the
+    repo root for the child.
+    """
+    python_path = str(package_root)
+    if ambient := os.environ.get("PYTHONPATH"):
+        python_path = os.pathsep.join((python_path, ambient))
+    return os.environ | {"PYTHONPATH": python_path}
+
+
 @pytest.fixture
 def setup_core(tmp_path: Path) -> Path:
     """Set up CORE with test paths."""

tests/unit_tests/fixtures/log/setup_log_probe.py

@@ -0,0 +1,21 @@
+"""Report whether setup_log() pulled in colorama, then print a colored line.
+
+Executed as a subprocess by test_log.py because module imports are
+process-global: the parent prints ``colorama_loaded=True/False`` plus an
+ANSI colored line so the caller can observe whether the codes survive to
+the stream. Pass ``--dashboard`` to simulate a dashboard-spawned run.
+"""
+
+import sys
+
+from esphome.core import CORE
+from esphome.log import setup_log
+
+if "--dashboard" in sys.argv:
+    CORE.dashboard = True
+
+setup_log()
+
+print(f"colorama_loaded={'colorama' in sys.modules}")
+print("\033[31mred\033[0m end")
+sys.stdout.flush()

tests/unit_tests/test_lazy_imports.py

@@ -15,7 +15,6 @@ test pins down *which* heavy modules must stay out entirely.
 from __future__ import annotations
 
 import importlib.util
-import os
 from pathlib import Path
 import subprocess
 import sys
@@ -120,18 +119,17 @@ def test_watched_heavy_modules_exist() -> None:
 
 
 def _leaked_from_fixture(
-    fixture_path: Path, script_name: str, extra: tuple[str, ...] = ()
+    fixture_path: Path,
+    env: dict[str, str],
+    script_name: str,
+    extra: tuple[str, ...] = (),
 ) -> str:
     """Run a fixture script with the watched modules on argv.
 
-    Running a script file drops the cwd from sys.path, so prepend the
-    repo root for the child; a non-zero exit surfaces the child's stderr.
+    ``env`` comes from the ``probe_env`` fixture so the child can import
+    the repo checkout; a non-zero exit surfaces the child's stderr.
     """
     script = fixture_path / "lazy_imports" / script_name
-    python_path = str(Path(__file__).parents[2])
-    if ambient := os.environ.get("PYTHONPATH"):
-        python_path = os.pathsep.join((python_path, ambient))
-    env = os.environ | {"PYTHONPATH": python_path}
     result = subprocess.run(
         [sys.executable, str(script), *FAST_PATH_HEAVY_MODULES, *extra],
         capture_output=True,
@@ -145,12 +143,13 @@ def _leaked_from_fixture(
 
 def test_storage_json_fast_path_does_not_import_heavy_modules(
     fixture_path: Path,
+    probe_env: dict[str, str],
 ) -> None:
     """``apply_to_core`` runs on the upload/logs fast path for every
     platform; parsing the stored framework version must not drag in the
     validation stack or the esp32 component package.
     """
-    leaked = _leaked_from_fixture(fixture_path, "storage_json_fast_path.py")
+    leaked = _leaked_from_fixture(fixture_path, probe_env, "storage_json_fast_path.py")
     assert not leaked, (
         f"storage_json.apply_to_core pulls in heavy modules: {leaked}. "
         "The upload/logs fast path skips validation; importing the "
@@ -160,12 +159,15 @@ def test_storage_json_fast_path_does_not_import_heavy_modules(
 
 def test_esptool_upload_fast_path_does_not_import_heavy_modules(
     fixture_path: Path,
+    probe_env: dict[str, str],
 ) -> None:
     """The esptool serial upload reads the esp32 variant from CORE.data;
     resolving it must not drag in the esp32 component package or the
     validation stack.
     """
-    leaked = _leaked_from_fixture(fixture_path, "esptool_upload_fast_path.py")
+    leaked = _leaked_from_fixture(
+        fixture_path, probe_env, "esptool_upload_fast_path.py"
+    )
     assert not leaked, (
         f"upload_using_esptool pulls in heavy modules: {leaked}. "
         "The upload fast path skips validation; importing the validation "
@@ -266,6 +268,7 @@ def test_yaml_util_does_not_import_heavy_modules() -> None:
 
 def test_upload_command_path_does_not_import_heavy_modules(
     fixture_path: Path,
+    probe_env: dict[str, str],
 ) -> None:
     """The single-config dispatch path checks the bundle suffix on every
     run; reading it from esphome.const must not drag in esphome.bundle
@@ -273,6 +276,7 @@ def test_upload_command_path_does_not_import_heavy_modules(
     """
     leaked = _leaked_from_fixture(
         fixture_path,
+        probe_env,
         "upload_command_fast_path.py",
         extra=BUNDLE_HEAVY_MODULES + CACHE_HIT_HEAVY_MODULES + STDLIB_FAST_PATH_MODULES,
     )

tests/unit_tests/test_log.py

@@ -1,6 +1,44 @@
+from collections.abc import Generator
+import errno
+import io
+import logging
+import os
+from pathlib import Path
+import select
+import subprocess
+import sys
+import time
+
 import pytest
 
-from esphome.log import AnsiFore, AnsiStyle, color
+from esphome.core import CORE
+from esphome.log import AnsiFore, AnsiStyle, color, setup_log
+
+
+class _FakeTty(io.StringIO):
+    def isatty(self) -> bool:
+        return True
+
+
+@pytest.fixture
+def restore_logging_state() -> Generator[None, None, None]:
+    """Undo the global logging changes setup_log() makes."""
+    root = logging.getLogger()
+    handlers = root.handlers[:]
+    formatters = [handler.formatter for handler in handlers]
+    level = root.level
+    urllib3_level = logging.getLogger("urllib3").level
+    yield
+    root.handlers[:] = handlers
+    for handler, formatter in zip(handlers, formatters, strict=True):
+        handler.setFormatter(formatter)
+    root.setLevel(level)
+    logging.getLogger("urllib3").setLevel(urllib3_level)
+
+
+def _probe_command(fixture_path: Path, *args: str) -> list[str]:
+    """Build the command line for the setup_log probe fixture script."""
+    return [sys.executable, str(fixture_path / "log" / "setup_log_probe.py"), *args]
 
 
 def test_color_keep_returns_unchanged_message() -> None:
@@ -78,3 +116,230 @@ def test_ansi_fore_keep_is_enum_member() -> None:
     assert bool(AnsiFore.KEEP) is True
     # But the value itself is still an empty string
     assert AnsiFore.KEEP.value == ""
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="colorama always initializes on Windows"
+)
+def test_setup_log_redirected_output_strips_ansi(
+    fixture_path: Path, probe_env: dict[str, str]
+) -> None:
+    """A redirected run must keep colorama so ANSI codes are stripped."""
+    result = subprocess.run(
+        _probe_command(fixture_path),
+        capture_output=True,
+        text=True,
+        timeout=60,
+        check=False,
+        env=probe_env,
+    )
+    assert result.returncode == 0, result.stderr
+    assert "colorama_loaded=True" in result.stdout
+    assert "red end" in result.stdout
+    assert "\033" not in result.stdout
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="colorama always initializes on Windows"
+)
+def test_setup_log_dashboard_skips_colorama(
+    fixture_path: Path, probe_env: dict[str, str]
+) -> None:
+    """Dashboard runs escape their color codes, so colorama must not load."""
+    result = subprocess.run(
+        _probe_command(fixture_path, "--dashboard"),
+        capture_output=True,
+        text=True,
+        timeout=60,
+        check=False,
+        env=probe_env,
+    )
+    assert result.returncode == 0, result.stderr
+    assert "colorama_loaded=False" in result.stdout
+    # Codes pass through untouched for the dashboard to handle.
+    assert "\033[31mred\033[0m end" in result.stdout
+
+
+def _run_probe_on_pty(
+    fixture_path: Path, probe_env: dict[str, str], *, stderr_to_pty: bool
+) -> str:
+    """Run the probe with stdout on a pty and return the decoded pty output.
+
+    With ``stderr_to_pty=False`` stderr goes to a pipe instead, giving the
+    mixed tty/redirect stream combination while keeping any traceback
+    available for the exit assertion.
+    """
+    # Unix-only; a module-level import would break test collection on
+    # Windows, where all the callers are skipped anyway.
+    import pty
+
+    controller, follower = pty.openpty()
+    proc = None
+    output = b""
+    deadline = time.monotonic() + 60
+    try:
+        try:
+            proc = subprocess.Popen(
+                _probe_command(fixture_path),
+                stdout=follower,
+                stderr=follower if stderr_to_pty else subprocess.PIPE,
+                stdin=follower,
+                env=probe_env,
+            )
+        finally:
+            os.close(follower)
+        while True:
+            timeout = deadline - time.monotonic()
+            if timeout <= 0 or not select.select([controller], [], [], timeout)[0]:
+                pytest.fail(f"pty probe produced no EOF in time; got {output!r}")
+            try:
+                chunk = os.read(controller, 1024)
+            except OSError as err:
+                # macOS raises EIO once the child closes its end of the pty;
+                # anything else is a real failure, not end-of-stream.
+                if err.errno != errno.EIO:
+                    raise
+                break
+            if not chunk:
+                break
+            output += chunk
+        stderr_text = ""
+        if proc.stderr is not None:
+            stderr_text = proc.stderr.read().decode(errors="replace")
+            proc.stderr.close()
+        assert proc.wait(60) == 0, stderr_text
+    finally:
+        os.close(controller)
+        if proc is not None and proc.poll() is None:
+            proc.kill()
+            proc.wait()
+    return output.decode()
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="pty is POSIX-only; colorama loads on Windows"
+)
+def test_setup_log_tty_skips_colorama(
+    fixture_path: Path, probe_env: dict[str, str]
+) -> None:
+    """A terminal run must skip colorama and keep ANSI codes intact."""
+    text = _run_probe_on_pty(fixture_path, probe_env, stderr_to_pty=True)
+    assert "colorama_loaded=False" in text
+    assert "\033[31mred\033[0m end" in text
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="pty is POSIX-only; colorama loads on Windows"
+)
+def test_setup_log_mixed_streams_init_colorama(
+    fixture_path: Path, probe_env: dict[str, str]
+) -> None:
+    """A tty stdout with a redirected stderr must still initialize colorama.
+
+    The guard requires both streams to be a tty; collapsing it to a
+    single-stream check would stop stripping ANSI from a redirected
+    stderr while stdout is a terminal.
+    """
+    text = _run_probe_on_pty(fixture_path, probe_env, stderr_to_pty=False)
+    assert "colorama_loaded=True" in text
+    # stdout is a tty, so colorama leaves its codes alone.
+    assert "\033[31mred\033[0m end" in text
+
+
+@pytest.fixture
+def colorama_probe(
+    monkeypatch: pytest.MonkeyPatch, restore_logging_state: None
+) -> Generator[None, None, None]:
+    """Shared preamble for the in-process guard-branch tests.
+
+    Clears colorama from sys.modules so the assertions prove what
+    setup_log() itself did, and snapshots CORE.verbose/quiet, which is
+    not a no-op: CORE.reset() does not restore them, so without the
+    snapshot setup_log()'s log-level side effects would leak into later
+    tests.
+    """
+    monkeypatch.delitem(sys.modules, "colorama", raising=False)
+    monkeypatch.setattr(CORE, "verbose", CORE.verbose)
+    monkeypatch.setattr(CORE, "quiet", CORE.quiet)
+    yield
+    # init() rebinds sys.stdout/stderr; restore them before monkeypatch
+    # puts the originals back.
+    if (colorama := sys.modules.get("colorama")) is not None:
+        colorama.deinit()
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="colorama always initializes on Windows"
+)
+def test_setup_log_dashboard_branch_skips_colorama_import(
+    monkeypatch: pytest.MonkeyPatch, colorama_probe: None
+) -> None:
+    """The dashboard side of the guard must not import colorama."""
+    monkeypatch.setattr(CORE, "dashboard", True)
+    setup_log()
+    assert "colorama" not in sys.modules
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="colorama always initializes on Windows"
+)
+def test_setup_log_tty_branch_skips_colorama_import(
+    monkeypatch: pytest.MonkeyPatch, colorama_probe: None
+) -> None:
+    """The tty side of the guard must not import colorama."""
+    monkeypatch.setattr(sys, "stdout", _FakeTty())
+    monkeypatch.setattr(sys, "stderr", _FakeTty())
+    setup_log()
+    assert "colorama" not in sys.modules
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="colorama always initializes on Windows"
+)
+def test_setup_log_redirected_branch_imports_colorama(
+    monkeypatch: pytest.MonkeyPatch, colorama_probe: None
+) -> None:
+    """Redirected streams must keep importing and initializing colorama."""
+    monkeypatch.setattr(sys, "stdout", io.StringIO())
+    monkeypatch.setattr(sys, "stderr", io.StringIO())
+    setup_log()
+    assert "colorama" in sys.modules
+
+
+@pytest.mark.parametrize("broken", ["missing", "closed"])
+def test_setup_log_broken_streams_import_colorama(
+    broken: str, monkeypatch: pytest.MonkeyPatch, colorama_probe: None
+) -> None:
+    """A missing or closed stream counts as a redirect and must not crash.
+
+    colorama tolerates both, so setup_log() has to reach its init rather
+    than raise inside the tty probe.
+    """
+    if broken == "missing":
+        stream = None
+    else:
+        stream = io.StringIO()
+        stream.close()
+    monkeypatch.setattr(sys, "stdout", stream)
+    monkeypatch.setattr(sys, "stderr", stream)
+    setup_log()
+    assert "colorama" in sys.modules
+
+
+def test_setup_log_win32_always_imports_colorama(
+    monkeypatch: pytest.MonkeyPatch, colorama_probe: None
+) -> None:
+    """The Windows clause must init colorama even when both streams are ttys.
+
+    Old Windows consoles need colorama to translate ANSI escapes, so the
+    platform check has to win over the tty check. colorama itself keys
+    off os.name, so on a POSIX host its init/deinit pair is a
+    passthrough.
+    """
+    monkeypatch.setattr(sys, "platform", "win32")
+    # Both streams are ttys: without the platform clause this combination
+    # would skip colorama.
+    monkeypatch.setattr(sys, "stdout", _FakeTty())
+    monkeypatch.setattr(sys, "stderr", _FakeTty())
+    setup_log()
+    assert "colorama" in sys.modules