| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -28,7 +28,8 @@ | |||
| 28 | 28 | import socket | |
| 29 | 29 | import struct | |
| 30 | 30 | import sys | |
| 31 | - from collections.abc import Sequence | ||
| 31 | + import warnings | ||
| 32 | + from collections.abc import Iterable, Sequence | ||
| 32 | 33 | from typing import Any, Union, cast | |
| 33 | 34 | ||
| 34 | 35 | import ifaddr | |
@@ -73,19 +74,39 @@ def _encode_address(address: str) -> bytes: | |||
| 73 | 74 | return socket.inet_pton(address_family, address) | |
| 74 | 75 | ||
| 75 | 76 | ||
| 76 | - def get_all_addresses() -> list[str]: | ||
| 77 | - return list({addr.ip for iface in ifaddr.get_adapters() for addr in iface.ips if addr.is_IPv4}) # type: ignore[misc] | ||
| 77 | + def get_all_addresses_ipv4(adapters: Iterable[ifaddr.Adapter]) -> list[str]: | ||
| 78 | + return list({addr.ip for iface in adapters for addr in iface.ips if addr.is_IPv4}) # type: ignore[misc] | ||
| 78 | 79 | ||
| 79 | 80 | ||
| 80 | - def get_all_addresses_v6() -> list[tuple[tuple[str, int, int], int]]: | ||
| 81 | + def get_all_addresses_ipv6(adapters: Iterable[ifaddr.Adapter]) -> list[tuple[tuple[str, int, int], int]]: | ||
| 81 | 82 | # IPv6 multicast uses positive indexes for interfaces | |
| 82 | 83 | # TODO: What about multi-address interfaces? | |
| 83 | 84 | return list( | |
| 84 | - {(addr.ip, iface.index) for iface in ifaddr.get_adapters() for addr in iface.ips if addr.is_IPv6} # type: ignore[misc] | ||
| 85 | + {(addr.ip, iface.index) for iface in adapters for addr in iface.ips if addr.is_IPv6} # type: ignore[misc] | ||
| 86 | + ) | ||
| 87 | + | ||
| 88 | + | ||
| 89 | + def get_all_addresses() -> list[str]: | ||
| 90 | + warnings.warn( | ||
| 91 | + "get_all_addresses is deprecated, and will be removed in a future version. Use ifaddr" | ||
| 92 | + "directly instead to get a list of adapters.", | ||
| 93 | + DeprecationWarning, | ||
| 94 | + stacklevel=2, | ||
| 95 | + ) | ||
| 96 | + return get_all_addresses_ipv4(ifaddr.get_adapters()) | ||
| 97 | + | ||
| 98 | + | ||
| 99 | + def get_all_addresses_v6() -> list[tuple[tuple[str, int, int], int]]: | ||
| 100 | + warnings.warn( | ||
| 101 | + "get_all_addresses_v6 is deprecated, and will be removed in a future version. Use ifaddr" | ||
| 102 | + "directly instead to get a list of adapters.", | ||
| 103 | + DeprecationWarning, | ||
| 104 | + stacklevel=2, | ||
| 85 | 105 | ) | |
| 106 | + return get_all_addresses_ipv6(ifaddr.get_adapters()) | ||
| 86 | 107 | ||
| 87 | 108 | ||
| 88 | - def ip6_to_address_and_index(adapters: list[ifaddr.Adapter], ip: str) -> tuple[tuple[str, int, int], int]: | ||
| 109 | + def ip6_to_address_and_index(adapters: Iterable[ifaddr.Adapter], ip: str) -> tuple[tuple[str, int, int], int]: | ||
| 89 | 110 | if "%" in ip: | |
| 90 | 111 | ip = ip[: ip.index("%")] # Strip scope_id. | |
| 91 | 112 | ipaddr = ipaddress.ip_address(ip) | |
@@ -102,7 +123,7 @@ def ip6_to_address_and_index(adapters: list[ifaddr.Adapter], ip: str) -> tuple[t | |||
| 102 | 123 | raise RuntimeError(f"No adapter found for IP address {ip}") | |
| 103 | 124 | ||
| 104 | 125 | ||
| 105 | - def interface_index_to_ip6_address(adapters: list[ifaddr.Adapter], index: int) -> tuple[str, int, int]: | ||
| 126 | + def interface_index_to_ip6_address(adapters: Iterable[ifaddr.Adapter], index: int) -> tuple[str, int, int]: | ||
| 106 | 127 | for adapter in adapters: | |
| 107 | 128 | if adapter.index == index: | |
| 108 | 129 | for adapter_ip in adapter.ips: | |
@@ -152,10 +173,11 @@ def normalize_interface_choice( | |||
| 152 | 173 | if ip_version != IPVersion.V6Only: | |
| 153 | 174 | result.append("0.0.0.0") | |
| 154 | 175 | elif choice is InterfaceChoice.All: | |
| 176 | + adapters = ifaddr.get_adapters() | ||
| 155 | 177 | if ip_version != IPVersion.V4Only: | |
| 156 | - result.extend(get_all_addresses_v6()) | ||
| 178 | + result.extend(get_all_addresses_ipv6(adapters)) | ||
| 157 | 179 | if ip_version != IPVersion.V6Only: | |
| 158 | - result.extend(get_all_addresses()) | ||
| 180 | + result.extend(get_all_addresses_ipv4(adapters)) | ||
| 159 | 181 | if not result: | |
| 160 | 182 | raise RuntimeError( | |
| 161 | 183 | f"No interfaces to listen on, check that any interfaces have IP version {ip_version}" | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -6,12 +6,14 @@ | |||
| 6 | 6 | import socket | |
| 7 | 7 | import sys | |
| 8 | 8 | import unittest | |
| 9 | + import warnings | ||
| 9 | 10 | from unittest.mock import MagicMock, Mock, patch | |
| 10 | 11 | ||
| 11 | 12 | import ifaddr | |
| 12 | 13 | import pytest | |
| 13 | 14 | ||
| 14 | 15 | import zeroconf as r | |
| 16 | + from zeroconf import get_all_addresses, get_all_addresses_v6 | ||
| 15 | 17 | from zeroconf._utils import net as netutils | |
| 16 | 18 | ||
| 17 | 19 | ||
@@ -35,6 +37,40 @@ def _generate_mock_adapters(): | |||
| 35 | 37 | return [mock_eth0, mock_lo0, mock_eth1, mock_vtun0] | |
| 36 | 38 | ||
| 37 | 39 | ||
| 40 | + def test_get_all_addresses() -> None: | ||
| 41 | + """Test public get_all_addresses API.""" | ||
| 42 | + with ( | ||
| 43 | + patch( | ||
| 44 | + "zeroconf._utils.net.ifaddr.get_adapters", | ||
| 45 | + return_value=_generate_mock_adapters(), | ||
| 46 | + ), | ||
| 47 | + warnings.catch_warnings(record=True) as warned, | ||
| 48 | + ): | ||
| 49 | + addresses = get_all_addresses() | ||
| 50 | + assert isinstance(addresses, list) | ||
| 51 | + assert len(addresses) == 3 | ||
| 52 | + assert len(warned) == 1 | ||
| 53 | + first_warning = warned[0] | ||
| 54 | + assert "get_all_addresses is deprecated" in str(first_warning.message) | ||
| 55 | + | ||
| 56 | + | ||
| 57 | + def test_get_all_addresses_v6() -> None: | ||
| 58 | + """Test public get_all_addresses_v6 API.""" | ||
| 59 | + with ( | ||
| 60 | + patch( | ||
| 61 | + "zeroconf._utils.net.ifaddr.get_adapters", | ||
| 62 | + return_value=_generate_mock_adapters(), | ||
| 63 | + ), | ||
| 64 | + warnings.catch_warnings(record=True) as warned, | ||
| 65 | + ): | ||
| 66 | + addresses = get_all_addresses_v6() | ||
| 67 | + assert isinstance(addresses, list) | ||
| 68 | + assert len(addresses) == 1 | ||
| 69 | + assert len(warned) == 1 | ||
| 70 | + first_warning = warned[0] | ||
| 71 | + assert "get_all_addresses_v6 is deprecated" in str(first_warning.message) | ||
| 72 | + | ||
| 73 | + | ||
| 38 | 74 | def test_ip6_to_address_and_index(): | |
| 39 | 75 | """Test we can extract from mocked adapters.""" | |
| 40 | 76 | adapters = _generate_mock_adapters() | |
@@ -84,8 +120,8 @@ def test_ip6_addresses_to_indexes(): | |||
| 84 | 120 | def test_normalize_interface_choice_errors(): | |
| 85 | 121 | """Test we generate exception on invalid input.""" | |
| 86 | 122 | with ( | |
| 87 | - patch("zeroconf._utils.net.get_all_addresses", return_value=[]), | ||
| 88 | - patch("zeroconf._utils.net.get_all_addresses_v6", return_value=[]), | ||
| 123 | + patch("zeroconf._utils.net.get_all_addresses_ipv4", return_value=[]), | ||
| 124 | + patch("zeroconf._utils.net.get_all_addresses_ipv6", return_value=[]), | ||
| 89 | 125 | pytest.raises(RuntimeError), | |
| 90 | 126 | ): | |
| 91 | 127 | netutils.normalize_interface_choice(r.InterfaceChoice.All) | |
| Back | FazBrowse Home | New Git URL |
0 commit comments