(self)
| 1130 | conn.request('POST', 'test', conn) |
| 1131 | |
| 1132 | def test_chunked(self): |
| 1133 | expected = chunked_expected |
| 1134 | sock = FakeSocket(chunked_start + last_chunk + chunked_end) |
| 1135 | resp = client.HTTPResponse(sock, method="GET") |
| 1136 | resp.begin() |
| 1137 | self.assertEqual(resp.read(), expected) |
| 1138 | resp.close() |
| 1139 | |
| 1140 | # Explicit full read |
| 1141 | for n in (-123, -1, None): |
| 1142 | with self.subTest('full read', n=n): |
| 1143 | sock = FakeSocket(chunked_start + last_chunk + chunked_end) |
| 1144 | resp = client.HTTPResponse(sock, method="GET") |
| 1145 | resp.begin() |
| 1146 | self.assertTrue(resp.chunked) |
| 1147 | self.assertEqual(resp.read(n), expected) |
| 1148 | resp.close() |
| 1149 | |
| 1150 | # Read first chunk |
| 1151 | with self.subTest('read1(-1)'): |
| 1152 | sock = FakeSocket(chunked_start + last_chunk + chunked_end) |
| 1153 | resp = client.HTTPResponse(sock, method="GET") |
| 1154 | resp.begin() |
| 1155 | self.assertTrue(resp.chunked) |
| 1156 | self.assertEqual(resp.read1(-1), b"hello worl") |
| 1157 | resp.close() |
| 1158 | |
| 1159 | # Various read sizes |
| 1160 | for n in range(1, 12): |
| 1161 | sock = FakeSocket(chunked_start + last_chunk + chunked_end) |
| 1162 | resp = client.HTTPResponse(sock, method="GET") |
| 1163 | resp.begin() |
| 1164 | self.assertEqual(resp.read(n) + resp.read(n) + resp.read(), expected) |
| 1165 | resp.close() |
| 1166 | |
| 1167 | for x in ('', 'foo\r\n'): |
| 1168 | sock = FakeSocket(chunked_start + x) |
| 1169 | resp = client.HTTPResponse(sock, method="GET") |
| 1170 | resp.begin() |
| 1171 | try: |
| 1172 | resp.read() |
| 1173 | except client.IncompleteRead as i: |
| 1174 | self.assertEqual(i.partial, expected) |
| 1175 | expected_message = 'IncompleteRead(%d bytes read)' % len(expected) |
| 1176 | self.assertEqual(repr(i), expected_message) |
| 1177 | self.assertEqual(str(i), expected_message) |
| 1178 | else: |
| 1179 | self.fail('IncompleteRead expected') |
| 1180 | finally: |
| 1181 | resp.close() |
| 1182 | |
| 1183 | def test_readinto_chunked(self): |
| 1184 |
nothing calls this directly
no test coverage detected