| 157 | ) |
| 158 | |
| 159 | def test_readline(self): |
| 160 | def getreader(input): |
| 161 | stream = io.BytesIO(input.encode(self.encoding)) |
| 162 | return codecs.getreader(self.encoding)(stream) |
| 163 | |
| 164 | def readalllines(input, keepends=True, size=None): |
| 165 | reader = getreader(input) |
| 166 | lines = [] |
| 167 | while True: |
| 168 | line = reader.readline(size=size, keepends=keepends) |
| 169 | if not line: |
| 170 | break |
| 171 | lines.append(line) |
| 172 | return "|".join(lines) |
| 173 | |
| 174 | s = "foo\nbar\r\nbaz\rspam\u2028eggs" |
| 175 | sexpected = "foo\n|bar\r\n|baz\r|spam\u2028|eggs" |
| 176 | sexpectednoends = "foo|bar|baz|spam|eggs" |
| 177 | self.assertEqual(readalllines(s, True), sexpected) |
| 178 | self.assertEqual(readalllines(s, False), sexpectednoends) |
| 179 | self.assertEqual(readalllines(s, True, 10), sexpected) |
| 180 | self.assertEqual(readalllines(s, False, 10), sexpectednoends) |
| 181 | |
| 182 | lineends = ("\n", "\r\n", "\r", "\u2028") |
| 183 | # Test long lines (multiple calls to read() in readline()) |
| 184 | vw = [] |
| 185 | vwo = [] |
| 186 | for (i, lineend) in enumerate(lineends): |
| 187 | vw.append((i*200+200)*"\u3042" + lineend) |
| 188 | vwo.append((i*200+200)*"\u3042") |
| 189 | self.assertEqual(readalllines("".join(vw), True), "|".join(vw)) |
| 190 | self.assertEqual(readalllines("".join(vw), False), "|".join(vwo)) |
| 191 | |
| 192 | # Test lines where the first read might end with \r, so the |
| 193 | # reader has to look ahead whether this is a lone \r or a \r\n |
| 194 | for size in range(80): |
| 195 | for lineend in lineends: |
| 196 | s = 10*(size*"a" + lineend + "xxx\n") |
| 197 | reader = getreader(s) |
| 198 | for i in range(10): |
| 199 | self.assertEqual( |
| 200 | reader.readline(keepends=True), |
| 201 | size*"a" + lineend, |
| 202 | ) |
| 203 | self.assertEqual( |
| 204 | reader.readline(keepends=True), |
| 205 | "xxx\n", |
| 206 | ) |
| 207 | reader = getreader(s) |
| 208 | for i in range(10): |
| 209 | self.assertEqual( |
| 210 | reader.readline(keepends=False), |
| 211 | size*"a", |
| 212 | ) |
| 213 | self.assertEqual( |
| 214 | reader.readline(keepends=False), |
| 215 | "xxx", |
| 216 | ) |