From 442486c7108f245fefc880bf6464b89a0f3e0339 Mon Sep 17 00:00:00 2001 From: Christopher Krah <48338417+chrikrah@users.noreply.github.com> Date: Mon, 12 Oct 2026 07:52:10 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Keep=20`CliRunner`=20output=20bu?= =?UTF-8?q?ffers=20open=20when=20its=20stream=20wrappers=20close?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port the `typer/testing.py` part of pallets/click#3139, released in Click 8.3.2. `_NamedTextIOWrapper` gets an empty `close()`, so dropping the wrapper no longer closes the `StreamMixer` buffer that `invoke()` reads afterwards, and `StreamMixer.__del__` goes, so a stream captured inside the command stays writable after `invoke()` returns. --- tests/test_termui.py | 35 ++++++++++++++++++++++++++++++++++- typer/testing.py | 11 +++++------ 2 files changed, 39 insertions(+), 7 deletions(-) diff --git a/tests/test_termui.py b/tests/test_termui.py index 8f908923db..edc4abd37f 100644 --- a/tests/test_termui.py +++ b/tests/test_termui.py @@ -5,8 +5,9 @@ import io import os +import sys from contextlib import contextmanager -from typing import Literal +from typing import Literal, TextIO import pytest import typer @@ -103,6 +104,38 @@ def main() -> None: assert os.environ.get(env_key) == "present" +def test_clirunner_stream_replaced_during_invoke() -> None: + # pytest's live logging suspends capturing mid-command: it swaps sys.stdout + # and sys.stderr, which drops the last reference to the runner's wrappers + runner = CliRunner() + app = typer.Typer() + + @app.command() + def main() -> None: + sys.stdout = io.StringIO() + sys.stderr = io.StringIO() + + result = runner.invoke(app, []) + assert result.exit_code == 0, result.output + assert result.output == "" + + +def test_clirunner_stream_usable_after_invoke() -> None: + # A logging.StreamHandler created inside the command keeps writing here + runner = CliRunner() + app = typer.Typer() + streams: list[TextIO] = [] + + @app.command() + def main() -> None: + streams.append(sys.stderr) + + result = runner.invoke(app, []) + assert result.exit_code == 0, result.output + streams[0].write("late") + streams[0].flush() + + @pytest.mark.parametrize( ("exit_value", "expected_exit_code", "expected_stdout"), [ diff --git a/typer/testing.py b/typer/testing.py index 7ecc0e693c..aebd459caa 100644 --- a/typer/testing.py +++ b/typer/testing.py @@ -53,12 +53,6 @@ def __init__(self) -> None: self.stdout: io.BytesIO = BytesIOCopy(copy_to=self.output) self.stderr: io.BytesIO = BytesIOCopy(copy_to=self.output) - def __del__(self) -> None: - """Guarantee that file-like objects are closed in a predictable order""" - self.stderr.close() - self.stdout.close() - self.output.close() - class _NamedTextIOWrapper(io.TextIOWrapper): def __init__(self, buffer: BinaryIO, name: str, mode: str, **kwargs: Any) -> None: @@ -66,6 +60,11 @@ def __init__(self, buffer: BinaryIO, name: str, mode: str, **kwargs: Any) -> Non self._name = name self._mode = mode + def close(self) -> None: + """The buffer belongs to `StreamMixer`, so keep it open when the + wrapper is closed or garbage collected. + """ + @property def name(self) -> str: return self._name # pragma: no cover