Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 34 additions & 1 deletion tests/test_termui.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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"),
[
Expand Down
11 changes: 5 additions & 6 deletions typer/testing.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,19 +53,18 @@ 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:
super().__init__(buffer, **kwargs)
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
Expand Down
Loading