(self)
| 1084 | conn.request('POST', 'test', conn) |
| 1085 | |
| 1086 | def test_chunked(self): |
| 1087 | expected = chunked_expected |
| 1088 | sock = FakeSocket(chunked_start + last_chunk + chunked_end) |
| 1089 | resp = client.HTTPResponse(sock, method="GET") |
| 1090 | resp.begin() |
| 1091 | self.assertEqual(resp.read(), expected) |
| 1092 | resp.close() |
| 1093 | |
| 1094 | # Explicit full read |
| 1095 | for n in (-123, -1, None): |
| 1096 | with self.subTest('full read', n=n): |
| 1097 | sock = FakeSocket(chunked_start + last_chunk + chunked_end) |
| 1098 | resp = client.HTTPResponse(sock, method="GET") |
| 1099 | resp.begin() |
| 1100 | self.assertTrue(resp.chunked) |
| 1101 | self.assertEqual(resp.read(n), expected) |
| 1102 | resp.close() |
| 1103 | |
| 1104 | # Read first chunk |
| 1105 | with self.subTest('read1(-1)'): |
| 1106 | sock = FakeSocket(chunked_start + last_chunk + chunked_end) |
| 1107 | resp = client.HTTPResponse(sock, method="GET") |
| 1108 | resp.begin() |
| 1109 | self.assertTrue(resp.chunked) |
| 1110 | self.assertEqual(resp.read1(-1), b"hello worl") |
| 1111 | resp.close() |
| 1112 | |
| 1113 | # Various read sizes |
| 1114 | for n in range(1, 12): |
| 1115 | sock = FakeSocket(chunked_start + last_chunk + chunked_end) |
| 1116 | resp = client.HTTPResponse(sock, method="GET") |
| 1117 | resp.begin() |
| 1118 | self.assertEqual(resp.read(n) + resp.read(n) + resp.read(), expected) |
| 1119 | resp.close() |
| 1120 | |
| 1121 | for x in ('', 'foo\r\n'): |
| 1122 | sock = FakeSocket(chunked_start + x) |
| 1123 | resp = client.HTTPResponse(sock, method="GET") |
| 1124 | resp.begin() |
| 1125 | try: |
| 1126 | resp.read() |
| 1127 | except client.IncompleteRead as i: |
| 1128 | self.assertEqual(i.partial, expected) |
| 1129 | expected_message = 'IncompleteRead(%d bytes read)' % len(expected) |
| 1130 | self.assertEqual(repr(i), expected_message) |
| 1131 | self.assertEqual(str(i), expected_message) |
| 1132 | else: |
| 1133 | self.fail('IncompleteRead expected') |
| 1134 | finally: |
| 1135 | resp.close() |
| 1136 | |
| 1137 | def test_readinto_chunked(self): |
| 1138 |
nothing calls this directly
no test coverage detected