Show More
@@ -24,6 +24,29 b' import pytest' | |||
|
24 | 24 | from vcsserver import subprocessio |
|
25 | 25 | |
|
26 | 26 | |
|
27 | class KindaFilelike(object): # pragma: no cover | |
|
28 | ||
|
29 | def __init__(self, data, size): | |
|
30 | chunks = size / len(data) | |
|
31 | ||
|
32 | self.stream = self._get_stream(data, chunks) | |
|
33 | ||
|
34 | def _get_stream(self, data, chunks): | |
|
35 | for x in xrange(chunks): | |
|
36 | yield data | |
|
37 | ||
|
38 | def read(self, n): | |
|
39 | ||
|
40 | buffer_stream = '' | |
|
41 | for chunk in self.stream: | |
|
42 | buffer_stream += chunk | |
|
43 | if len(buffer_stream) >= n: | |
|
44 | break | |
|
45 | ||
|
46 | # self.stream = self.bytes[n:] | |
|
47 | return buffer_stream | |
|
48 | ||
|
49 | ||
|
27 | 50 | @pytest.fixture(scope='module') |
|
28 | 51 | def environ(): |
|
29 | 52 | """Delete coverage variables, as they make the tests fail.""" |
@@ -101,8 +124,9 b' def test_output_with_no_input_does_not_f' | |||
|
101 | 124 | |
|
102 | 125 | @pytest.mark.parametrize('size', [1, 10 ** 5]) |
|
103 | 126 | def test_output_with_input(size, environ): |
|
104 |
data = |
|
|
105 | inputstream = io.BytesIO(data) | |
|
127 | data_len = size | |
|
128 | inputstream = KindaFilelike('X', size) | |
|
129 | ||
|
106 | 130 | # This acts like the cat command. |
|
107 | 131 | args = _get_python_args('shutil.copyfileobj(sys.stdin, sys.stdout)') |
|
108 | 132 | output = ''.join( |
@@ -111,14 +135,14 b' def test_output_with_input(size, environ' | |||
|
111 | 135 | ) |
|
112 | 136 | ) |
|
113 | 137 | |
|
114 | print("{} {}".format(len(data * size), len(output))) | |
|
115 | assert output == data | |
|
138 | assert len(output) == data_len | |
|
116 | 139 | |
|
117 | 140 | |
|
118 | 141 | @pytest.mark.parametrize('size', [1, 10 ** 5]) |
|
119 | 142 | def test_output_with_input_skipping_iterator(size, environ): |
|
120 |
data = |
|
|
121 | inputstream = io.BytesIO(data) | |
|
143 | data_len = size | |
|
144 | inputstream = KindaFilelike('X', size) | |
|
145 | ||
|
122 | 146 | # This acts like the cat command. |
|
123 | 147 | args = _get_python_args('shutil.copyfileobj(sys.stdin, sys.stdout)') |
|
124 | 148 | |
@@ -128,5 +152,4 b' def test_output_with_input_skipping_iter' | |||
|
128 | 152 | ) |
|
129 | 153 | output = ''.join(chunker.output) |
|
130 | 154 | |
|
131 | print("{} {}".format(len(data * size), len(output))) | |
|
132 | assert output == data | |
|
155 | assert len(output) == data_len |
General Comments 0
You need to be logged in to leave comments.
Login now