Skip to content

Commit

Permalink
Handle stderr, and add comments to the test
Browse files Browse the repository at this point in the history
  • Loading branch information
tbttfox authored and MHendricks committed Mar 16, 2024
1 parent c23b802 commit aed5090
Show file tree
Hide file tree
Showing 2 changed files with 36 additions and 12 deletions.
1 change: 1 addition & 0 deletions preditor/stream/director.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ def close(self):
self.old_stream
and not self.std_stream_wrapped
and self.old_stream is not sys.__stdout__
and self.old_stream is not sys.__stderr__
):
self.old_stream.close()

Expand Down
47 changes: 35 additions & 12 deletions tests/test_stream.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
import pytest
import six

from preditor.stream import STDOUT, Director, Manager, install_to_std
from preditor.stream import STDOUT, STDERR, Director, Manager, install_to_std


@pytest.fixture
Expand Down Expand Up @@ -48,7 +48,9 @@ def test_no_old_stream(manager):


def test_nul_stream(manager):
# I can't set the `name` of a regular TextIOWrapper
# I can't set the `name` prop of a regular TextIOWrapper
# So I need this subclass so I can mock the pythonw
# standard out/err streams
class NamedTextIOWrapper(io.TextIOWrapper):
def __init__(self, buffer, name=None, **kwargs):
vars(self)['name'] = name
Expand All @@ -59,22 +61,43 @@ def __getattribute__(self, name):
return vars(self)['name']
return super().__getattribute__(name)

orig_director = Director(manager, STDOUT)
wrap_director = Director(manager, 'test_wrap', old_stream=orig_director)
# Make directors that wrap sys.stdout and sys.stderr
# This way we can check that the default behavior works
stdout_director = Director(manager, STDOUT)
stderr_director = Director(manager, STDERR)

# Wrap the stdout/stderr directors so we can check that
# director.std_stream_wrapped is not set for these cases
wrapout_director = Director(manager, 'test_wrapout', old_stream=stdout_director)
wraperr_director = Director(manager, 'test_wraperr', old_stream=stdout_director)

# Back up sys.stdout/stderr and replace them with my own streams that replicate
# the name/encoding of the windows pythonw.exe default streams
orig_stdout = sys.stdout
orig_stderr = sys.stderr
sys.stdout = NamedTextIOWrapper(six.BytesIO(), name='nul', encoding='cp1252')
sys.stderr = NamedTextIOWrapper(six.BytesIO(), name='nul', encoding='cp1252')
try:
null_director = Director(manager, STDOUT)
# Build a director here that will grab the nul streams
# so we can check that they don't store them in .old_stream
nullout_director = Director(manager, STDOUT)
nullerr_director = Director(manager, STDERR)
finally:
# And make sure to restore the backed up stdout
sys.stdout = orig_stdout

assert null_director.old_stream is None
assert orig_director.old_stream is sys.stdout

assert not null_director.std_stream_wrapped
assert orig_director.std_stream_wrapped
assert not wrap_director.std_stream_wrapped
sys.stderr = orig_stderr

assert nullout_director.old_stream is None
assert nullerr_director.old_stream is None
assert stdout_director.old_stream is sys.stdout
assert stderr_director.old_stream is sys.stderr

assert stdout_director.std_stream_wrapped
assert stderr_director.std_stream_wrapped
assert not nullout_director.std_stream_wrapped
assert not nullerr_director.std_stream_wrapped
assert not wrapout_director.std_stream_wrapped
assert not wraperr_director.std_stream_wrapped


def test_callback(manager, stdout, stderr):
Expand Down

0 comments on commit aed5090

Please sign in to comment.