Skip to content

Commit

Permalink
fix: make parsed_scoped_addresses return addresses in the same order …
Browse files Browse the repository at this point in the history
…as all other methods (#1150)
  • Loading branch information
bdraco committed Apr 2, 2023
1 parent 9a16be5 commit 9b6adcf
Show file tree
Hide file tree
Showing 2 changed files with 42 additions and 24 deletions.
62 changes: 40 additions & 22 deletions src/zeroconf/_services/info.py
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
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

0 comments on commit 9b6adcf

Please sign in to comment.