forked from wolph/python-progressbar
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_stream.py
More file actions
150 lines (107 loc) · 3.57 KB
/
Copy pathtest_stream.py
File metadata and controls
150 lines (107 loc) · 3.57 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
import io
import os
import sys
import pytest
import progressbar
from progressbar import terminal
def test_nowrap() -> None:
# Make sure we definitely unwrap
for _i in range(5):
progressbar.streams.unwrap(stderr=True, stdout=True)
stdout = sys.stdout
stderr = sys.stderr
progressbar.streams.wrap()
assert stdout == sys.stdout
assert stderr == sys.stderr
progressbar.streams.unwrap()
assert stdout == sys.stdout
assert stderr == sys.stderr
# Make sure we definitely unwrap
for _i in range(5):
progressbar.streams.unwrap(stderr=True, stdout=True)
def test_wrap() -> None:
# Make sure we definitely unwrap
for _i in range(5):
progressbar.streams.unwrap(stderr=True, stdout=True)
stdout = sys.stdout
stderr = sys.stderr
progressbar.streams.wrap(stderr=True, stdout=True)
assert stdout != sys.stdout
assert stderr != sys.stderr
# Wrap again
stdout = sys.stdout
stderr = sys.stderr
progressbar.streams.wrap(stderr=True, stdout=True)
assert stdout == sys.stdout
assert stderr == sys.stderr
# Make sure we definitely unwrap
for _i in range(5):
progressbar.streams.unwrap(stderr=True, stdout=True)
def test_excepthook() -> None:
progressbar.streams.wrap(stderr=True, stdout=True)
try:
raise RuntimeError() # noqa: TRY301
except RuntimeError:
progressbar.streams.excepthook(*sys.exc_info())
progressbar.streams.unwrap_excepthook()
progressbar.streams.unwrap_excepthook()
def test_fd_as_io_stream() -> None:
stream = io.StringIO()
with progressbar.ProgressBar(fd=stream) as pb:
for i in range(101):
pb.update(i)
stream.close()
def test_no_newlines() -> None:
kwargs = dict(
redirect_stderr=True,
redirect_stdout=True,
line_breaks=False,
is_terminal=True,
)
with progressbar.ProgressBar(**kwargs) as bar:
for i in range(5):
bar.update(i)
for i in range(5, 10):
try:
print('\n\n', file=progressbar.streams.stdout)
print('\n\n', file=progressbar.streams.stderr)
except ValueError:
pass
bar.update(i)
@pytest.mark.parametrize('stream', [sys.__stdout__, sys.__stderr__])
@pytest.mark.skipif(os.name == 'nt', reason='Windows does not support this')
def test_fd_as_standard_streams(stream) -> None:
with progressbar.ProgressBar(fd=stream) as pb:
for i in range(101):
pb.update(i)
def test_line_offset_stream_wrapper() -> None:
stream = terminal.LineOffsetStreamWrapper(5, io.StringIO())
stream.write('Hello World!')
def test_last_line_stream_methods() -> None:
stream = terminal.LastLineStream(io.StringIO())
# Test write method
stream.write('Hello World!')
assert stream.read() == 'Hello World!'
assert stream.read(5) == 'Hello'
# Test flush method
stream.flush()
assert stream.line == 'Hello World!'
assert stream.readline() == 'Hello World!'
assert stream.readline(5) == 'Hello'
# Test truncate method
stream.truncate(5)
assert stream.line == 'Hello'
stream.truncate()
assert stream.line == ''
# Test seekable/readable
assert not stream.seekable()
assert stream.readable()
stream.writelines(['a', 'b', 'c'])
assert stream.read() == 'c'
assert list(stream) == ['c']
with stream:
stream.write('Hello World!')
assert stream.read() == 'Hello World!'
assert stream.read(5) == 'Hello'
# Test close method
stream.close()