(self)
| 1181 | resp.close() |
| 1182 | |
| 1183 | def test_readinto_chunked(self): |
| 1184 | |
| 1185 | expected = chunked_expected |
| 1186 | nexpected = len(expected) |
| 1187 | b = bytearray(128) |
| 1188 | |
| 1189 | sock = FakeSocket(chunked_start + last_chunk + chunked_end) |
| 1190 | resp = client.HTTPResponse(sock, method="GET") |
| 1191 | resp.begin() |
| 1192 | n = resp.readinto(b) |
| 1193 | self.assertEqual(b[:nexpected], expected) |
| 1194 | self.assertEqual(n, nexpected) |
| 1195 | resp.close() |
| 1196 | |
| 1197 | # Various read sizes |
| 1198 | for n in range(1, 12): |
| 1199 | sock = FakeSocket(chunked_start + last_chunk + chunked_end) |
| 1200 | resp = client.HTTPResponse(sock, method="GET") |
| 1201 | resp.begin() |
| 1202 | m = memoryview(b) |
| 1203 | i = resp.readinto(m[0:n]) |
| 1204 | i += resp.readinto(m[i:n + i]) |
| 1205 | i += resp.readinto(m[i:]) |
| 1206 | self.assertEqual(b[:nexpected], expected) |
| 1207 | self.assertEqual(i, nexpected) |
| 1208 | resp.close() |
| 1209 | |
| 1210 | for x in ('', 'foo\r\n'): |
| 1211 | sock = FakeSocket(chunked_start + x) |
| 1212 | resp = client.HTTPResponse(sock, method="GET") |
| 1213 | resp.begin() |
| 1214 | try: |
| 1215 | n = resp.readinto(b) |
| 1216 | except client.IncompleteRead as i: |
| 1217 | self.assertEqual(i.partial, expected) |
| 1218 | expected_message = 'IncompleteRead(%d bytes read)' % len(expected) |
| 1219 | self.assertEqual(repr(i), expected_message) |
| 1220 | self.assertEqual(str(i), expected_message) |
| 1221 | else: |
| 1222 | self.fail('IncompleteRead expected') |
| 1223 | finally: |
| 1224 | resp.close() |
| 1225 | |
| 1226 | def test_chunked_head(self): |
| 1227 | chunked_start = ( |
nothing calls this directly
no test coverage detected