Verify buffer resize trigger.
(self)
| 173 | request_id=1) |
| 174 | |
| 175 | def test_trigger_buffer_resize(self): |
| 176 | """ |
| 177 | Verify buffer resize trigger. |
| 178 | """ |
| 179 | test_stream = io.BytesIO(b'Content-Length: 15\r\n\r\n{"key":"value"}') |
| 180 | json_rpc_reader = jsonrpc.JsonRpcReader(test_stream) |
| 181 | # set the message buffer to a small size triggering a resize. |
| 182 | json_rpc_reader.buffer = bytearray(2) |
| 183 | # Initial size set to 2 bytes. |
| 184 | self.assertEqual(len(json_rpc_reader.buffer), 2) |
| 185 | response = json_rpc_reader.read_response() |
| 186 | baseline = {u'key': u'value'} |
| 187 | self.assertEqual(response, baseline) |
| 188 | # Verify message buffer was reset to it's default max size. |
| 189 | self.assertEqual(len(json_rpc_reader.buffer), 8192) |
| 190 | |
| 191 | def test_max_buffer_resize(self): |
| 192 | """ |
nothing calls this directly
no test coverage detected