FilesExpand file tree

 develop

/

test_stream.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

147 lines (105 loc) · 3.44 KB

 develop

/

test_stream.py

Copy path

Top

File metadata and controls

• Code
• Blame

147 lines (105 loc) · 3.44 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

import io

import sys

import progressbar

import pytest

from progressbar import terminal

def test_nowrap():

# 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():

# 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():

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():

stream = io.StringIO()

with progressbar.ProgressBar(fd=stream) as pb:

for i in range(101):

pb.update(i)

stream.close()

def test_no_newlines():

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__])

def test_fd_as_standard_streams(stream):

with progressbar.ProgressBar(fd=stream) as pb:

for i in range(101):

pb.update(i)

def test_line_offset_stream_wrapper():

stream = terminal.LineOffsetStreamWrapper(5, io.StringIO())

stream.write('Hello World!')

def test_last_line_stream_methods():

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()