""" transport/ice_filter — the interface filter must name the same adapter on every platform, and must never leave the gather with no address at all. """ import aioice.ice import ifaddr import pytest from meshbay_node.transport import ice_filter class _IP: def __init__(self, ip): self.ip = ip class _Adapter: def __init__(self, name, nice_name, ips): self.name, self.nice_name = name, nice_name self.ips = [_IP(ip) for ip in ips] def _linux(name, *ips): """On Linux ifaddr repeats the kernel name in nice_name.""" return _Adapter(name, name, list(ips)) def _windows(guid, description, *ips): """On Windows `name` is a GUID and `nice_name` the device description.""" return _Adapter(guid, description, list(ips)) @pytest.fixture(autouse=True) def _restore(): saved = aioice.ice.get_host_addresses yield aioice.ice.get_host_addresses = saved @pytest.fixture def adapters(monkeypatch): """Install a fake adapter set; returns a setter the test calls.""" def _set(*adapters): monkeypatch.setattr(ifaddr, "get_adapters", lambda: list(adapters)) return _set def _gather(include=None): ice_filter.install(include) return aioice.ice.get_host_addresses() def test_auto_excludes_virtual_adapters(adapters): adapters( _linux("lo", "127.0.0.1"), _linux("wlp3s0f0", "192.168.1.22"), _linux("virbr0", "192.168.200.254"), _linux("docker0", "172.17.0.1"), ) assert _gather() == ["192.168.1.22"] def test_auto_drops_cgnat_range(adapters): adapters( _linux("eth0", "192.168.1.22"), _linux("tun0", "100.101.102.103"), ) assert _gather() == ["192.168.1.22"] def test_include_list_by_kernel_name(adapters): adapters( _linux("eth0", "192.168.1.22"), _linux("wlp3s0f0", "192.168.1.23"), ) assert _gather(["wlp3s0f0"]) == ["192.168.1.23"] def test_include_list_keeps_explicitly_named_cgnat_adapter(adapters): """An explicit name beats the auto-exclusion rules.""" adapters(_linux("eth0", "192.168.1.22"), _linux("tun0", "100.101.102.103")) assert _gather(["tun0"]) == ["100.101.102.103"] def test_include_list_matches_windows_description(adapters): """ The Windows adapter GUID is not something an operator ever types, so the device description has to match too. """ adapters( _windows("{846EE342-7039-11DE-9D20-806E6F6E6963}", "Gigabit Network Connection", "192.168.200.173"), _windows("{0C6A2C4B-1111-2222-3333-444455556666}", "Loopback Pseudo-Interface", "127.0.0.1"), ) assert _gather(["Gigabit Network Connection"]) == ["192.168.200.173"] def test_include_list_matching_is_case_insensitive(adapters): adapters(_windows("{GUID}", "Gigabit Network Connection", "192.168.200.173")) assert _gather(["gigabit network connection"]) == ["192.168.200.173"] def test_include_list_matches_an_ip_literal(adapters): """The address is the one identifier visible on every platform.""" adapters( _windows("{GUID-A}", "Gigabit Network Connection", "192.168.200.173"), _windows("{GUID-B}", "Virtual Adapter", "10.0.0.5"), ) assert _gather(["192.168.200.173"]) == ["192.168.200.173"] def test_unmatched_include_list_falls_back_to_every_interface(adapters): """ The regression: a Linux-style ice_interfaces carried to a Windows guest matched no GUID and no description, leaving the node with zero host candidates — silently unconnectable rather than merely slow. """ adapters( _windows("{846EE342-7039-11DE-9D20-806E6F6E6963}", "Gigabit Network Connection", "192.168.200.173"), ) assert _gather(["wlp0s20f3"]) == ["192.168.200.173"] def test_fallback_when_every_adapter_is_virtual(adapters): """Auto-exclude must not empty the list either.""" adapters(_linux("virbr0", "192.168.200.254"), _linux("docker0", "172.17.0.1")) assert _gather() == ["192.168.200.254", "172.17.0.1"] def test_fallback_warns(adapters, caplog): adapters(_windows("{GUID}", "Gigabit Network Connection", "192.168.200.173")) with caplog.at_level("WARNING", logger=ice_filter.log.name): _gather(["wlp0s20f3"]) assert "matched no address" in caplog.text def test_loopback_only_host_yields_nothing_and_does_not_warn(adapters, caplog): """No address to offer is not a filter misconfiguration.""" adapters(_linux("lo", "127.0.0.1")) with caplog.at_level("WARNING", logger=ice_filter.log.name): assert _gather() == [] assert caplog.text == "" def test_ipv6_global_kept_link_local_dropped(adapters): adapters(_Adapter("eth0", "eth0", [ ("2a01:cb15:80e8:de00::1", 0, 0), ("fe80::af4c:4912:c1ec:b6dc", 0, 3), ("::1", 0, 0), ])) assert _gather() == ["2a01:cb15:80e8:de00::1"] def test_ipv4_is_not_index_probed_as_ipv6(adapters): """ An IPv4 str must be discriminated by type: the old code fell through to the IPv6 branch and indexed characters out of it. """ ice_filter.install(None) adapters(_linux("eth0", "192.168.1.22")) assert aioice.ice.get_host_addresses(use_ipv4=False, use_ipv6=True) == []