(f, protocol_version)
| 257 | |
| 258 | @staticmethod |
| 259 | def recv_error_info(f, protocol_version): |
| 260 | consistency = read_consistency_level(f) |
| 261 | received_responses = read_int(f) |
| 262 | required_responses = read_int(f) |
| 263 | |
| 264 | if ProtocolVersion.uses_error_code_map(protocol_version): |
| 265 | error_code_map = read_error_code_map(f) |
| 266 | failures = len(error_code_map) |
| 267 | else: |
| 268 | error_code_map = None |
| 269 | failures = read_int(f) |
| 270 | |
| 271 | data_retrieved = bool(read_byte(f)) |
| 272 | |
| 273 | return { |
| 274 | 'consistency': consistency, |
| 275 | 'received_responses': received_responses, |
| 276 | 'required_responses': required_responses, |
| 277 | 'failures': failures, |
| 278 | 'error_code_map': error_code_map, |
| 279 | 'data_retrieved': data_retrieved |
| 280 | } |
| 281 | |
| 282 | def to_exception(self): |
| 283 | return ReadFailure(self.summary_msg(), **self.info) |
nothing calls this directly
no test coverage detected