diff --git a/esphome/espidf/runner.py b/esphome/espidf/runner.py index 7ed11d75543..bf563a4f438 100644 --- a/esphome/espidf/runner.py +++ b/esphome/espidf/runner.py @@ -74,6 +74,64 @@ FILTER_IDF_LINES: list[str] = [ r"Stopping at filesystem boundary", ] +# Windows code page identifier for UTF-8, as used by ``chcp 65001``. +UTF8_CODEPAGE = 65001 + + +def _get_kernel32(): + """Return the Windows kernel32 module, or None on any other platform.""" + if sys.platform != "win32": + return None + import ctypes + + return ctypes.windll.kernel32 + + +class _Utf8Console: + """Keep an attached Windows console on UTF-8 for the length of the build. + + The build tree runs in UTF-8 mode, so esp_idf_size draws its table with + Unicode box characters. ``idf.py size`` reaches it through ``cmake -P``, + and CMake re-decodes the child's output with the console code page before + printing it, which turns the table into mojibake on any code page but + UTF-8. Every process in the build shares this console, so switching it + here covers CMake too. The old code pages go back on exit so the user's + terminal is left as it was. + + A console that is already on UTF-8 is left alone. The code page belongs to + the console, not to this process, so a build that overlaps another one + must not save UTF-8 as the page to go back to. + """ + + def __init__(self, kernel32) -> None: + self._kernel32 = kernel32 + self._codepages: tuple[int, int] | None = None + + def __enter__(self) -> None: + kernel32 = self._kernel32 + if kernel32 is None: + return + old_in = kernel32.GetConsoleCP() + old_out = kernel32.GetConsoleOutputCP() + # Both calls return 0 when no console is attached. + if not old_in or not old_out: + return + if old_in == UTF8_CODEPAGE and old_out == UTF8_CODEPAGE: + return + # Record the old pages first so a switch that fails part way through + # still gets put back on exit. + self._codepages = (old_in, old_out) + kernel32.SetConsoleCP(UTF8_CODEPAGE) + kernel32.SetConsoleOutputCP(UTF8_CODEPAGE) + + def __exit__(self, *exc_info: object) -> None: + if self._codepages is None: + return + old_in, old_out = self._codepages + self._codepages = None + self._kernel32.SetConsoleCP(old_in) + self._kernel32.SetConsoleOutputCP(old_out) + def main() -> int: # ---- sys.path fix-up --------------------------------------------------- @@ -269,8 +327,29 @@ def main() -> int: is_verbose = any(arg in ("-v", "--verbose") for arg in sys.argv[2:]) filter_lines = None if is_verbose else FILTER_IDF_LINES or None - stdout_shim = sys.stdout = _FilteringTTYStream(sys.stdout, filter_lines) # type: ignore[assignment] - stderr_shim = sys.stderr = _FilteringTTYStream(sys.stderr, filter_lines) # type: ignore[assignment] + class _FilteredStreams: + """Route ``sys.stdout`` and ``sys.stderr`` through the filtering shims. + + On exit each shim releases a last line that never got its + terminator. The shims made here are drained rather than whatever + ``sys.stdout`` holds by then, which the script is free to replace. + """ + + def __init__(self, filter_lines: list[str] | None) -> None: + self._stdout = _FilteringTTYStream(sys.stdout, filter_lines) + self._stderr = _FilteringTTYStream(sys.stderr, filter_lines) + + def __enter__(self) -> None: + sys.stdout = self._stdout # type: ignore[assignment] + sys.stderr = self._stderr # type: ignore[assignment] + + def __exit__(self, *exc_info: object) -> None: + # Drain stderr from a finally so a surprise from the first one + # cannot strand the second. + try: + self._stdout.drain() + finally: + self._stderr.drain() # Shift argv so the target script sees its own path as argv[0] and # its own arguments starting at argv[1]. runpy.run_path does not @@ -288,19 +367,11 @@ def main() -> int: # If idf.py calls sys.exit(), SystemExit propagates out of run_path # and carries the exit code back to our caller. For normal returns, - # fall through and exit with 0. Either way the streams get a chance to - # release a last line that never got its terminator. Drain the shims we - # made rather than sys.stdout, which the script is free to replace, and - # report instead of raising so cleanup cannot bury the real exit code. - try: + # fall through and exit with 0. Either way the context managers drain + # the streams and put the console back on the way out, and they report + # instead of raising so cleanup cannot bury the real exit code. + with _FilteredStreams(filter_lines), _Utf8Console(_get_kernel32()): runpy.run_path(script_path, run_name="__main__") - finally: - # Drain stderr from a finally so a surprise from the first one cannot - # strand the second. - try: - stdout_shim.drain() - finally: - stderr_shim.drain() return 0 diff --git a/tests/unit_tests/test_espidf_runner.py b/tests/unit_tests/test_espidf_runner.py index e4cc6e137e7..71ccfee1f2c 100644 --- a/tests/unit_tests/test_espidf_runner.py +++ b/tests/unit_tests/test_espidf_runner.py @@ -209,3 +209,107 @@ def test_runner_streams_output_before_the_build_finishes( # Join before leaving the block, so the reader is done rather than # racing ``Popen`` closing the pipe under it. reader.join(1.0) + + +class _FakeKernel32: + """Stand-in for the Windows kernel32 console code page calls.""" + + def __init__(self, input_cp: int, output_cp: int) -> None: + self.input_cp = input_cp + self.output_cp = output_cp + self.calls: list[tuple[str, int]] = [] + + def GetConsoleCP(self) -> int: # noqa: N802 + return self.input_cp + + def GetConsoleOutputCP(self) -> int: # noqa: N802 + return self.output_cp + + def SetConsoleCP(self, codepage: int) -> int: # noqa: N802 + self.calls.append(("SetConsoleCP", codepage)) + self.input_cp = codepage + return 1 + + def SetConsoleOutputCP(self, codepage: int) -> int: # noqa: N802 + self.calls.append(("SetConsoleOutputCP", codepage)) + self.output_cp = codepage + return 1 + + +def test_main_runs_the_build_with_a_utf8_console( + monkeypatch: pytest.MonkeyPatch, fixture_path: Path +) -> None: + """An attached console is switched to UTF-8 and then put back.""" + kernel32 = _FakeKernel32(850, 850) + monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32) + + _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py") + + assert kernel32.calls == [ + ("SetConsoleCP", runner.UTF8_CODEPAGE), + ("SetConsoleOutputCP", runner.UTF8_CODEPAGE), + ("SetConsoleCP", 850), + ("SetConsoleOutputCP", 850), + ] + + +def test_main_restores_the_console_when_the_build_dies( + monkeypatch: pytest.MonkeyPatch, fixture_path: Path +) -> None: + """A failing build must not leave the user's console on UTF-8.""" + kernel32 = _FakeKernel32(437, 437) + monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32) + _prepare_main(monkeypatch, fixture_path / "espidf" / "crashing_probe.py") + + with pytest.raises(SystemExit): + runner.main() + + assert (kernel32.input_cp, kernel32.output_cp) == (437, 437) + + +def test_main_restores_the_console_when_the_switch_fails_part_way( + monkeypatch: pytest.MonkeyPatch, fixture_path: Path +) -> None: + """A failed output page switch must not strand the changed input page.""" + kernel32 = _FakeKernel32(850, 850) + + def _refuse(codepage: int) -> int: + kernel32.calls.append(("SetConsoleOutputCP", codepage)) + return 0 + + kernel32.SetConsoleOutputCP = _refuse # type: ignore[method-assign] + monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32) + + _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py") + + assert kernel32.input_cp == 850 + assert kernel32.calls[-2:] == [("SetConsoleCP", 850), ("SetConsoleOutputCP", 850)] + + +def test_main_leaves_the_console_alone_when_there_is_none( + monkeypatch: pytest.MonkeyPatch, fixture_path: Path +) -> None: + """Without a console the code page calls return 0 and nothing is set.""" + kernel32 = _FakeKernel32(0, 0) + monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32) + + _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py") + + assert kernel32.calls == [] + + +def test_main_leaves_a_console_already_on_utf8_alone( + monkeypatch: pytest.MonkeyPatch, fixture_path: Path +) -> None: + """An overlapping build must not save UTF-8 as the page to restore.""" + kernel32 = _FakeKernel32(runner.UTF8_CODEPAGE, runner.UTF8_CODEPAGE) + monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32) + + _run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py") + + assert kernel32.calls == [] + + +@pytest.mark.skipif(sys.platform == "win32", reason="kernel32 exists on Windows") +def test_get_kernel32_is_none_off_windows() -> None: + assert runner._get_kernel32() is None