Commit 83fadb1
committed
refactor: extract _entry_ip_version helper to DRY multicast_addresses validation and autodetect
1 parent 5fc8e96 commit 83fadb1
1 file changed
Lines changed: 20 additions & 24 deletions
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -439,6 +439,17 @@ def new_respond_socket( | |||
| 439 | 439 | return respond_socket | |
| 440 | 440 | ||
| 441 | 441 | ||
| 442 | + def _entry_ip_version(entry: str | int | tuple[tuple[str, int, int], int]) -> int: | ||
| 443 | + """Return 4 or 6 for the IP version implied by an interface / multicast entry. | ||
| 444 | + | ||
| 445 | + ``int`` (interface index) and ``tuple`` (ifaddr IPv6 adapter tuple) entries | ||
| 446 | + are IPv6 by construction; ``str`` entries are parsed as IP addresses. | ||
| 447 | + """ | ||
| 448 | + if isinstance(entry, str): | ||
| 449 | + return ipaddress.ip_address(entry).version | ||
| 450 | + return 6 | ||
| 451 | + | ||
| 452 | + | ||
| 442 | 453 | def create_sockets( | |
| 443 | 454 | interfaces: InterfacesType = InterfaceChoice.All, | |
| 444 | 455 | unicast: bool = False, | |
@@ -461,16 +472,10 @@ def create_sockets( | |||
| 461 | 472 | # Reject IP-version-incompatible entries up front so callers get a clear | |
| 462 | 473 | # error instead of a confusing adapter-lookup or socket-syscall failure. | |
| 463 | 474 | if multicast_addresses: | |
| 464 | - if ip_version == IPVersion.V4Only: | ||
| 465 | - for entry in multicast_addresses: | ||
| 466 | - if isinstance(entry, (int, tuple)) or ( | ||
| 467 | - isinstance(entry, str) and ipaddress.ip_address(entry).version == 6 | ||
| 468 | - ): | ||
| 469 | - raise ValueError("multicast_addresses contains IPv6 entries but ip_version is V4Only") | ||
| 470 | - elif ip_version == IPVersion.V6Only: | ||
| 471 | - for entry in multicast_addresses: | ||
| 472 | - if isinstance(entry, str) and ipaddress.ip_address(entry).version == 4: | ||
| 473 | - raise ValueError("multicast_addresses contains IPv4 entries but ip_version is V6Only") | ||
| 475 | + if ip_version == IPVersion.V4Only and any(_entry_ip_version(e) == 6 for e in multicast_addresses): | ||
| 476 | + raise ValueError("multicast_addresses contains IPv6 entries but ip_version is V4Only") | ||
| 477 | + if ip_version == IPVersion.V6Only and any(_entry_ip_version(e) == 4 for e in multicast_addresses): | ||
| 478 | + raise ValueError("multicast_addresses contains IPv4 entries but ip_version is V6Only") | ||
| 474 | 479 | ||
| 475 | 480 | if unicast: | |
| 476 | 481 | listen_socket = None | |
@@ -534,22 +539,13 @@ def autodetect_ip_version( | |||
| 534 | 539 | multicast_addresses: Sequence[str | int | tuple[tuple[str, int, int], int]] | None = None, | |
| 535 | 540 | ) -> IPVersion: | |
| 536 | 541 | """Auto detect the IP version when it is not provided.""" | |
| 537 | - has_v6 = False | ||
| 538 | - has_v4 = False | ||
| 542 | + entries: list[str | int | tuple[tuple[str, int, int], int]] = [] | ||
| 539 | 543 | if isinstance(interfaces, list): | |
| 540 | - has_v6 = any( | ||
| 541 | - isinstance(i, int) or (isinstance(i, str) and ipaddress.ip_address(i).version == 6) | ||
| 542 | - for i in interfaces | ||
| 543 | - ) | ||
| 544 | - has_v4 = any(isinstance(i, str) and ipaddress.ip_address(i).version == 4 for i in interfaces) | ||
| 544 | + entries.extend(interfaces) | ||
| 545 | 545 | if multicast_addresses: | |
| 546 | - has_v6 = has_v6 or any( | ||
| 547 | - isinstance(i, (int, tuple)) or (isinstance(i, str) and ipaddress.ip_address(i).version == 6) | ||
| 548 | - for i in multicast_addresses | ||
| 549 | - ) | ||
| 550 | - has_v4 = has_v4 or any( | ||
| 551 | - isinstance(i, str) and ipaddress.ip_address(i).version == 4 for i in multicast_addresses | ||
| 552 | - ) | ||
| 546 | + entries.extend(multicast_addresses) | ||
| 547 | + has_v6 = any(_entry_ip_version(e) == 6 for e in entries) | ||
| 548 | + has_v4 = any(_entry_ip_version(e) == 4 for e in entries) | ||
| 553 | 549 | if has_v4 and has_v6: | |
| 554 | 550 | return IPVersion.All | |
| 555 | 551 | if has_v6: | |
0 commit comments