(
section: list[ResourceRecord], section_name: str, count: int
)
| 424 | raise struct.error(f"question #{i}: {e}") |
| 425 | |
| 426 | def unpack_rrs( |
| 427 | section: list[ResourceRecord], section_name: str, count: int |
| 428 | ) -> None: |
| 429 | nonlocal buffer, offset |
| 430 | for i in range(0, count): |
| 431 | try: |
| 432 | name = unpack_domain_name() |
| 433 | type, class_, ttl, len_data = ResourceRecord.HEADER.unpack_from( |
| 434 | buffer, offset |
| 435 | ) |
| 436 | offset += ResourceRecord.HEADER.size |
| 437 | end_data = offset + len_data |
| 438 | if len(buffer) < end_data: |
| 439 | raise struct.error( |
| 440 | f"unpack requires a data buffer of {len_data} bytes" |
| 441 | ) |
| 442 | data = buffer[offset:end_data] |
| 443 | |
| 444 | if domain_names.record_data_can_have_compression(type): |
| 445 | data = domain_names.decompress_from_record_data( |
| 446 | buffer, offset, end_data, cached_names |
| 447 | ) |
| 448 | |
| 449 | section.append(ResourceRecord(name, type, class_, ttl, data)) |
| 450 | offset += len_data |
| 451 | except struct.error as e: |
| 452 | raise struct.error(f"{section_name} #{i}: {e}") |
| 453 | |
| 454 | unpack_rrs(msg.answers, "answer", len_answers) |
| 455 | unpack_rrs(msg.authorities, "authority", len_authorities) |
nothing calls this directly
no test coverage detected