Skip to content

Commit

Permalink
drop python<=3.7 support
Browse files Browse the repository at this point in the history
According to https://endoflife.date/python python 3.7 has been EOSed
almost year ago.
Filter all code over `pyupgrade --py3.8-plus`.

Signed-off-by: Tomasz Kłoczko <[email protected]>
  • Loading branch information
kloczek committed Jun 12, 2024
1 parent 41fb17f commit aae97d4
Show file tree
Hide file tree
Showing 2 changed files with 29 additions and 33 deletions.
55 changes: 26 additions & 29 deletions test.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,3 @@
# coding: utf-8
from __future__ import print_function

import io
import logging
import os
Expand Down Expand Up @@ -40,55 +37,55 @@ def printf_err(msg):

def test_pipes():
with pipes(stdout=PIPE, stderr=PIPE) as (stdout, stderr):
printf(u"Hellø")
printf_err(u"Hi, stdérr")
printf("Hellø")
printf_err("Hi, stdérr")

assert stdout.read() == u"Hellø\n"
assert stderr.read() == u"Hi, stdérr\n"
assert stdout.read() == "Hellø\n"
assert stderr.read() == "Hi, stdérr\n"


def test_pipe_bytes():
with pipes(encoding=None) as (stdout, stderr):
printf(u"Hellø")
printf_err(u"Hi, stdérr")
printf("Hellø")
printf_err("Hi, stdérr")

assert stdout.read() == u"Hellø\n".encode('utf8')
assert stderr.read() == u"Hi, stdérr\n".encode('utf8')
assert stdout.read() == "Hellø\n".encode()
assert stderr.read() == "Hi, stdérr\n".encode()


def test_forward():
stdout = io.StringIO()
stderr = io.StringIO()
with pipes(stdout=stdout, stderr=stderr) as (_stdout, _stderr):
printf(u"Hellø")
printf_err(u"Hi, stdérr")
printf("Hellø")
printf_err("Hi, stdérr")
assert _stdout is stdout
assert _stderr is stderr

assert stdout.getvalue() == u"Hellø\n"
assert stderr.getvalue() == u"Hi, stdérr\n"
assert stdout.getvalue() == "Hellø\n"
assert stderr.getvalue() == "Hi, stdérr\n"


def test_pipes_stderr():
stdout = io.StringIO()
with pipes(stdout=stdout, stderr=STDOUT) as (_stdout, _stderr):
printf(u"Hellø")
printf("Hellø")
libc.fflush(c_stdout_p)
time.sleep(0.1)
printf_err(u"Hi, stdérr")
printf_err("Hi, stdérr")
assert _stdout is stdout
assert _stderr is None

assert stdout.getvalue() == u"Hellø\nHi, stdérr\n"
assert stdout.getvalue() == "Hellø\nHi, stdérr\n"


def test_flush():
stdout = io.StringIO()
w = Wurlitzer(stdout=stdout, stderr=STDOUT)
with w:
printf_err(u"Hellø")
printf_err("Hellø")
time.sleep(0.5)
assert stdout.getvalue().strip() == u"Hellø"
assert stdout.getvalue().strip() == "Hellø"


def test_sys_pipes():
Expand All @@ -97,11 +94,11 @@ def test_sys_pipes():
with mock.patch('sys.stdout', stdout), mock.patch(
'sys.stderr', stderr
), sys_pipes():
printf(u"Hellø")
printf_err(u"Hi, stdérr")
printf("Hellø")
printf_err("Hi, stdérr")

assert stdout.getvalue() == u"Hellø\n"
assert stderr.getvalue() == u"Hi, stdérr\n"
assert stdout.getvalue() == "Hellø\n"
assert stderr.getvalue() == "Hi, stdérr\n"


def test_sys_pipes_check():
Expand All @@ -119,16 +116,16 @@ def test_redirect_everything():
stderr = io.StringIO()
with mock.patch('sys.stdout', stdout), mock.patch('sys.stderr', stderr):
sys_pipes_forever()
printf(u"Hellø")
printf_err(u"Hi, stdérr")
printf("Hellø")
printf_err("Hi, stdérr")
stop_sys_pipes()
assert stdout.getvalue() == u"Hellø\n"
assert stderr.getvalue() == u"Hi, stdérr\n"
assert stdout.getvalue() == "Hellø\n"
assert stderr.getvalue() == "Hi, stdérr\n"


def count_fds():
"""utility for counting file descriptors"""
proc_fds = '/proc/{}/fd'.format(os.getpid())
proc_fds = f'/proc/{os.getpid()}/fd'
if os.path.isdir(proc_fds):
return len(proc_fds)
else:
Expand Down
7 changes: 3 additions & 4 deletions wurlitzer.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
Use `wurlitzer.pipes` or `wurlitzer.sys_pipes` as context managers.
"""

from __future__ import print_function

__version__ = '3.2.0.dev'

Expand Down Expand Up @@ -120,7 +119,7 @@ def dup2(a, b, timeout=3):
raise dup_err


@lru_cache()
@lru_cache
def _get_max_pipe_size():
"""Get max pipe size
Expand All @@ -139,7 +138,7 @@ def _get_max_pipe_size():
# If Linux procfs (or something that looks like it) exposes its
# maximum F_SETPIPE_SZ, adjust the default buffer sizes.
try:
with open('/proc/sys/fs/pipe-max-size', 'r') as f:
with open('/proc/sys/fs/pipe-max-size') as f:
# Figure out OS max pipe size
pipe_max_size = int(f.read())
except Exception:
Expand Down Expand Up @@ -540,7 +539,7 @@ def sys_pipes(encoding=_default_encoding, bufsize=None):
# check that we aren't forwarding stdout to itself
for name in ("stdout", "stderr"):
stream = getattr(sys, name)
capture_stream = getattr(sys, "__{}__".format(name))
capture_stream = getattr(sys, f"__{name}__")
try:
fd = stream.fileno()
capture_fd = capture_stream.fileno()
Expand Down

0 comments on commit aae97d4

Please sign in to comment.