Verify max buffer resize.
(self)
| 189 | self.assertEqual(len(json_rpc_reader.buffer), 8192) |
| 190 | |
| 191 | def test_max_buffer_resize(self): |
| 192 | """ |
| 193 | Verify max buffer resize. |
| 194 | """ |
| 195 | test_stream = io.BytesIO(b'Content-Length: 15\r\n\r\n{"key":"value"}') |
| 196 | json_rpc_reader = jsonrpc.JsonRpcReader(test_stream) |
| 197 | # Double buffer size to max to verify resize takes leftover size. |
| 198 | json_rpc_reader.buffer = bytearray(16384) |
| 199 | # Verify initial buffer size was set. |
| 200 | self.assertEqual(len(json_rpc_reader.buffer), 16384) |
| 201 | response = json_rpc_reader.read_response() |
| 202 | baseline = {u'key': u'value'} |
| 203 | self.assertEqual(response, baseline) |
| 204 | # Verify buffer size decreased by bytes_read. |
| 205 | self.assertEqual(len(json_rpc_reader.buffer), 16347) |
| 206 | |
| 207 | def test_read_state(self): |
| 208 | """ |
nothing calls this directly
no test coverage detected