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