Assert expected exception when reading response with invalid headers.
(self)
| 95 | self.assertTrue(test_stream.closed) |
| 96 | |
| 97 | def test_incorrect_header_formats(self): |
| 98 | """ |
| 99 | Assert expected exception when reading response with invalid headers. |
| 100 | """ |
| 101 | # Verify end of stream thrown with invalid header. |
| 102 | test_stream = io.BytesIO(b'Content-Length: 15\r\n{"key":"value"}') |
| 103 | json_rpc_reader = jsonrpc.JsonRpcReader(test_stream) |
| 104 | with self.assertRaises(EOFError): |
| 105 | json_rpc_reader.read_response() |
| 106 | |
| 107 | # Test with no content-length header. |
| 108 | test_stream = io.BytesIO(b'Missing-Header: True\r\n\r\n') |
| 109 | json_rpc_reader = jsonrpc.JsonRpcReader(test_stream) |
| 110 | with self.assertRaises(LookupError): |
| 111 | json_rpc_reader.read_response() |
| 112 | |
| 113 | json_rpc_reader.close() |
| 114 | self.assertTrue(test_stream.closed) |
| 115 | |
| 116 | # Missing colon. |
| 117 | test_stream = io.BytesIO(b'Retry-On-Failure True\r\n\r\n') |
| 118 | json_rpc_reader = jsonrpc.JsonRpcReader(test_stream) |
| 119 | with self.assertRaises(KeyError): |
| 120 | json_rpc_reader.read_response() |
| 121 | json_rpc_reader.close() |
| 122 | self.assertTrue(test_stream.closed) |
| 123 | |
| 124 | def test_invalid_json_response(self): |
| 125 | """ |
nothing calls this directly
no test coverage detected