(f, protocol_version)
| 305 | |
| 306 | @staticmethod |
| 307 | def recv_error_info(f, protocol_version): |
| 308 | consistency = read_consistency_level(f) |
| 309 | received_responses = read_int(f) |
| 310 | required_responses = read_int(f) |
| 311 | |
| 312 | if ProtocolVersion.uses_error_code_map(protocol_version): |
| 313 | error_code_map = read_error_code_map(f) |
| 314 | failures = len(error_code_map) |
| 315 | else: |
| 316 | error_code_map = None |
| 317 | failures = read_int(f) |
| 318 | |
| 319 | write_type = WriteType.name_to_value[read_string(f)] |
| 320 | |
| 321 | return { |
| 322 | 'consistency': consistency, |
| 323 | 'received_responses': received_responses, |
| 324 | 'required_responses': required_responses, |
| 325 | 'failures': failures, |
| 326 | 'error_code_map': error_code_map, |
| 327 | 'write_type': write_type |
| 328 | } |
| 329 | |
| 330 | def to_exception(self): |
| 331 | return WriteFailure(self.summary_msg(), **self.info) |
nothing calls this directly
no test coverage detected