Skip to content

fix: make parsed_scoped_addresses return addresses in the same order as all other methods #1150

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bdraco merged 3 commits into from
Apr 2, 2023
Merged

fix: make parsed_scoped_addresses return addresses in the same order as all other methods #1150

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
681073a
feat: improve performance of ip address handling in ServiceInfo
bdraco Apr 2, 2023
efe23fc
fix: update test
bdraco Apr 2, 2023
7543ba1
Merge branch 'master' into ip_address_cleanups
bdraco Apr 2, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 40 additions & 22 deletions src/zeroconf/_services/info.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,8 @@

import ipaddress
import random
import socket
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union, cast
from typing import TYPE_CHECKING, Dict, List, Optional, Set, Union, cast

from .._dns import (
DNSAddress,
Expand All @@ -40,7 +39,7 @@
from .._updates import RecordUpdate, RecordUpdateListener
from .._utils.asyncio import get_running_loop, run_coro_with_timeout
from .._utils.name import service_type_name
from .._utils.net import IPVersion, _encode_address, _is_v6_address
from .._utils.net import IPVersion, _encode_address
from .._utils.time import current_time_millis
from ..const import (
_CLASS_IN,
Expand Down Expand Up @@ -223,7 +222,14 @@ def properties(self) -> Dict:
return self._properties

def addresses_by_version(self, version: IPVersion) -> List[bytes]:
"""List addresses matching IP version."""
"""List addresses matching IP version.

Addresses are guaranteed to be returned in LIFO (last in, first out)
order with IPv4 addresses first and IPv6 addresses second.

This means the first address will always be the most recently added
address of the given IP version.
"""
if version == IPVersion.V4Only:
return [addr.packed for addr in self._ipv4_addresses]
if version == IPVersion.V6Only:
Expand All @@ -236,35 +242,47 @@ def addresses_by_version(self, version: IPVersion) -> List[bytes]:
def ip_addresses_by_version(
self, version: IPVersion
) -> Union[List[ipaddress.IPv4Address], List[ipaddress.IPv6Address], List[ipaddress._BaseAddress]]:
"""List ip_address objects matching IP version."""
"""List ip_address objects matching IP version.

Addresses are guaranteed to be returned in LIFO (last in, first out)
order with IPv4 addresses first and IPv6 addresses second.

This means the first address will always be the most recently added
address of the given IP version.
"""
if version == IPVersion.V4Only:
return self._ipv4_addresses
if version == IPVersion.V6Only:
return self._ipv6_addresses
return [*self._ipv4_addresses, *self._ipv6_addresses]

def parsed_addresses(self, version: IPVersion = IPVersion.All) -> List[str]:
"""List addresses in their parsed string form."""
result = self.addresses_by_version(version)
return [
socket.inet_ntop(socket.AF_INET6 if _is_v6_address(addr) else socket.AF_INET, addr)
for addr in result
]
"""List addresses in their parsed string form.

Addresses are guaranteed to be returned in LIFO (last in, first out)
order with IPv4 addresses first and IPv6 addresses second.

This means the first address will always be the most recently added
address of the given IP version.
"""
return [str(addr) for addr in self.ip_addresses_by_version(version)]

def parsed_scoped_addresses(self, version: IPVersion = IPVersion.All) -> List[str]:
"""Equivalent to parsed_addresses, with the exception that IPv6 Link-Local
addresses are qualified with %<interface_index> when available

Addresses are guaranteed to be returned in LIFO (last in, first out)
order with IPv4 addresses first and IPv6 addresses second.

This means the first address will always be the most recently added
address of the given IP version.
"""
if self.interface_index is None:
return self.parsed_addresses(version)

def is_link_local(addr_str: str) -> Any:
addr = _cached_ip_addresses(addr_str)
return addr.version == 6 and addr.is_link_local

ll_addrs = list(filter(is_link_local, self.parsed_addresses(version)))
other_addrs = list(filter(lambda addr: not is_link_local(addr), self.parsed_addresses(version)))
return [f"{addr}%{self.interface_index}" for addr in ll_addrs] + other_addrs
return [
f"{addr}%{self.interface_index}" if addr.version == 6 and addr.is_link_local else str(addr)
for addr in self.ip_addresses_by_version(version)
]

def _set_properties(self, properties: Dict) -> None:
"""Sets properties and text of this info from a dictionary"""
Expand Down Expand Up @@ -399,13 +417,13 @@ def dns_addresses(
return [
DNSAddress(
self.server,
_TYPE_AAAA if _is_v6_address(address) else _TYPE_A,
_TYPE_AAAA if address.version == 6 else _TYPE_A,
_CLASS_IN | _CLASS_UNIQUE,
override_ttl if override_ttl is not None else self.host_ttl,
address,
address.packed,
created=created,
)
for address in self.addresses_by_version(version)
for address in self.ip_addresses_by_version(version)
]

def dns_pointer(self, override_ttl: Optional[int] = None, created: Optional[float] = None) -> DNSPointer:
Expand Down
4 changes: 2 additions & 2 deletions tests/services/test_info.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -557,14 +557,14 @@ def test_multiple_addresses():
assert info.parsed_addresses(r.IPVersion.V4Only) == [address_parsed]
assert info.parsed_addresses(r.IPVersion.V6Only) == [address_v6_parsed, address_v6_ll_parsed]
assert info.parsed_scoped_addresses() == [
address_v6_ll_scoped_parsed,
address_parsed,
address_v6_parsed,
address_v6_ll_scoped_parsed,
]
assert info.parsed_scoped_addresses(r.IPVersion.V4Only) == [address_parsed]
assert info.parsed_scoped_addresses(r.IPVersion.V6Only) == [
address_v6_ll_scoped_parsed,
address_v6_parsed,
address_v6_ll_scoped_parsed,
]


Expand Down