Restore transport coverage of .local name aliasing

The KolibriBroadcast test-class deletion dropped ~13 tests covering the
.local hostname aliasing (bare/device-name aliases, multi-homed LAN address
selection, outgoing-interface preference, conflict-skipping, rename and
unregister handling). That behaviour was copied verbatim into
ZeroconfNetworkDiscovery, not changed, so port the tests onto the transport
test case rather than leaving the transport's local-name methods untested.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rtibblesbot

kolibri/core/discovery/test/test_network_broadcast.py

@@ -9,6 +9,7 @@
 from zeroconf import ServiceStateChange
 from zeroconf import Zeroconf
 
+from ..utils.network.broadcast import BARE_LOCAL_LABEL
 from ..utils.network.broadcast import EVENT_ADD_INSTANCE
 from ..utils.network.broadcast import EVENT_REGISTER_INSTANCE
 from ..utils.network.broadcast import EVENT_REMOVE_INSTANCE
@@ -21,6 +22,8 @@
 from ..utils.network.broadcast import KolibriInstance
 from ..utils.network.broadcast import KolibriInstanceListener
 from ..utils.network.broadcast import LOCAL_DOMAIN
+from ..utils.network.broadcast import LOCAL_NAME_BARE
+from ..utils.network.broadcast import LOCAL_NAME_DEVICE
 from ..utils.network.broadcast import NetworkDiscoveryBackend
 from ..utils.network.broadcast import SERVICE_TTL
 from ..utils.network.broadcast import SERVICE_TYPE
@@ -323,6 +326,8 @@ def setUp(self):
         enqueue_patcher.start()
         self.addCleanup(enqueue_patcher.stop)
         self.transport = ZeroconfNetworkDiscovery()
+        # Captures the transport's local-name events (published on its own bus).
+        self.listener = self.transport.add_listener(KolibriTestInstanceListener)
         self.on_add = mock.Mock()
         self.on_update = mock.Mock()
         self.on_remove = mock.Mock()
@@ -331,12 +336,23 @@ def setUp(self):
         )
         get_all_addresses_patcher.start()
         self.addCleanup(get_all_addresses_patcher.stop)
+        # Default to "no default route" so tests don't open a real socket and
+        # address selection falls back to the LAN-filtered addresses. Tests
+        # exercising the outgoing-interface preference override this.
         outgoing_patcher = mock.patch(
             BROADCAST_MODULE + "get_outgoing_interface_address", return_value=None
         )
-        outgoing_patcher.start()
+        self.mock_outgoing_interface_address = outgoing_patcher.start()
         self.addCleanup(outgoing_patcher.stop)
 
+    def _register_with_device_name(self, device_name):
+        self.transport.zeroconf = self.zeroconf
+        self.instance.device_info = {"device_name": device_name}
+        self.instance.to_service_info.return_value = mock.Mock(spec_set=ServiceInfo)(
+            "primary"
+        )
+        self.transport.register(self.instance)
+
     @mock.patch(BROADCAST_MODULE + "logger.info")
     @mock.patch(BROADCAST_MODULE + "Zeroconf")
     def test_register_opens_zeroconf_and_registers(self, mock_zeroconf, mock_logger):
@@ -383,6 +399,181 @@ def test_register_rename(self, mock_logger):
             service_info_unique, ttl=SERVICE_TTL
         )
 
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_register__local_names(self, mock_logger):
+        self.transport.zeroconf = self.zeroconf
+        service_info = mock.Mock(spec_set=ServiceInfo)("test")
+        self.instance.to_service_info.return_value = service_info
+        self.transport.register(self.instance)
+        bare_label, bare_service = self.transport.local_names[LOCAL_NAME_BARE]
+        self.assertEqual(BARE_LOCAL_LABEL, bare_label)
+        self.assertEqual("kolibri.local.", bare_service.server)
+        self.assertEqual(socket.inet_aton(MOCK_LAN_IP), bare_service.address)
+        self.assertNotIn(LOCAL_NAME_DEVICE, self.transport.local_names)
+
+    @mock.patch(BROADCAST_MODULE + "get_all_addresses")
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_register__local_names_lan_address_selection(
+        self, mock_logger, mock_get_all_addresses
+    ):
+        self.transport.zeroconf = self.zeroconf
+        service_info = mock.Mock(spec_set=ServiceInfo)("test")
+        self.instance.to_service_info.return_value = service_info
+        cases = [
+            (
+                [MOCK_CGNAT_IP, MOCK_LINK_LOCAL_IP, MOCK_LAN_IP],
+                socket.inet_aton(MOCK_LAN_IP),
+            ),
+            ([MOCK_CGNAT_IP], None),
+        ]
+        for addresses, expected_address in cases:
+            mock_get_all_addresses.return_value = addresses
+            self.transport.register(self.instance)
+            _, bare_service = self.transport.local_names[LOCAL_NAME_BARE]
+            self.assertEqual(expected_address, bare_service.address)
+
+    @mock.patch(BROADCAST_MODULE + "get_all_addresses")
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_register__prefers_outgoing_interface_address(
+        self, mock_logger, mock_get_all_addresses
+    ):
+        # On a multi-homed host both RFC1918 addresses survive the LAN filter,
+        # so we must advertise the default-route interface, not whichever one
+        # sorts first. Reproduces the QA-reported case where a Docker-bridge
+        # address was handed to LAN peers that couldn't reach it.
+        self.transport.zeroconf = self.zeroconf
+        self.instance.to_service_info.return_value = mock.Mock(spec_set=ServiceInfo)(
+            "primary"
+        )
+        mock_get_all_addresses.return_value = [MOCK_SECONDARY_LAN_IP, MOCK_LAN_IP]
+        self.mock_outgoing_interface_address.return_value = MOCK_LAN_IP
+        self.transport.register(self.instance)
+        _, bare_service = self.transport.local_names[LOCAL_NAME_BARE]
+        self.assertEqual(socket.inet_aton(MOCK_LAN_IP), bare_service.address)
+
+    @mock.patch(BROADCAST_MODULE + "get_all_addresses")
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_register__falls_back_when_outgoing_not_lan_reachable(
+        self, mock_logger, mock_get_all_addresses
+    ):
+        # If the outgoing interface isn't LAN-reachable (e.g. a VPN default
+        # route filtered out as CGNAT), fall back to a LAN-filtered address
+        # rather than advertising the unreachable one.
+        self.transport.zeroconf = self.zeroconf
+        self.instance.to_service_info.return_value = mock.Mock(spec_set=ServiceInfo)(
+            "primary"
+        )
+        mock_get_all_addresses.return_value = [MOCK_CGNAT_IP, MOCK_LAN_IP]
+        self.mock_outgoing_interface_address.return_value = MOCK_CGNAT_IP
+        self.transport.register(self.instance)
+        _, bare_service = self.transport.local_names[LOCAL_NAME_BARE]
+        self.assertEqual(socket.inet_aton(MOCK_LAN_IP), bare_service.address)
+
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_register__device_name_alias(self, mock_logger):
+        self._register_with_device_name("Tony's Laptop")
+        device_label, device_service = self.transport.local_names[LOCAL_NAME_DEVICE]
+        self.assertEqual("tonyslaptop", device_label)
+        self.assertEqual("tonyslaptop.local.", device_service.server)
+
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_register__device_name_alias_empty_slug(self, mock_logger):
+        self._register_with_device_name("   ")
+        self.assertNotIn(LOCAL_NAME_DEVICE, self.transport.local_names)
+
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_register__local_name_conflict_skipped(self, mock_logger):
+        # These aliases make no attempt to stay unique: if the name is already
+        # claimed on the network, we skip ours rather than renaming or crashing
+        # the whole broadcast.
+        self.transport.zeroconf = self.zeroconf
+        self.instance.to_service_info.return_value = mock.Mock(spec_set=ServiceInfo)(
+            "primary"
+        )
+        # primary registers fine; the bare alias is already claimed
+        self.zeroconf.register_service.side_effect = [None, NonUniqueNameException()]
+        self.transport.register(self.instance)  # must not raise
+        self.assertNotIn(LOCAL_NAME_BARE, self.transport.local_names)
+        self.assertEqual([], self.transport.local_hostnames)
+
+    def test_local_hostnames(self):
+        self._register_with_device_name("My Device")
+        self.assertEqual(
+            {"kolibri.local", "mydevice.local"},
+            set(self.transport.local_hostnames),
+        )
+        self.assertEqual(
+            {"kolibri.local", "mydevice.local"},
+            set(self.listener.mock.update_local_names.call_args[0][0]),
+        )
+
+    @mock.patch(BROADCAST_MODULE + "get_all_addresses")
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_renew__local_names_follow_lan_address_change(
+        self, mock_logger, mock_get_all_addresses
+    ):
+        mock_get_all_addresses.return_value = [MOCK_LAN_IP]
+        self._register_with_device_name("Some Name")
+
+        new_lan_ip = "192.168.1.9"
+        mock_get_all_addresses.return_value = [new_lan_ip]
+        self.transport.renew()
+
+        _, bare_service = self.transport.local_names[LOCAL_NAME_BARE]
+        self.assertEqual(socket.inet_aton(new_lan_ip), bare_service.address)
+
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_renew__device_name_changed(self, mock_logger):
+        self._register_with_device_name("Old Name")
+        old_service = self.transport.local_names[LOCAL_NAME_DEVICE][1]
+
+        self.instance.device_info = {"device_name": "New Name"}
+        self.transport.renew()
+
+        self.zeroconf.unregister_service.assert_any_call(old_service)
+        new_label, new_service = self.transport.local_names[LOCAL_NAME_DEVICE]
+        self.assertEqual("newname", new_label)
+        self.assertEqual("newname.local.", new_service.server)
+        self.zeroconf.register_service.assert_any_call(new_service, ttl=new_service.ttl)
+
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_renew__device_name_changed_to_empty_slug(self, mock_logger):
+        self._register_with_device_name("Old Name")
+        old_service = self.transport.local_names[LOCAL_NAME_DEVICE][1]
+
+        self.instance.device_info = {"device_name": "   "}
+        self.transport.renew()
+
+        self.zeroconf.unregister_service.assert_any_call(old_service)
+        self.assertNotIn(LOCAL_NAME_DEVICE, self.transport.local_names)
+
+    @mock.patch(BROADCAST_MODULE + "logger.info")
+    def test_renew__device_name_unchanged(self, mock_logger):
+        self._register_with_device_name("Same Name")
+        self.zeroconf.register_service.reset_mock()
+        self.zeroconf.unregister_service.reset_mock()
+
+        self.transport.renew()
+
+        self.zeroconf.unregister_service.assert_not_called()
+        self.zeroconf.register_service.assert_not_called()
+        # 1 for the primary instance (pre-existing, unchanged renew() logic)
+        # + 2 for the bare and device aliases, both re-announced with fresh port/ttl
+        self.assertEqual(3, self.zeroconf.update_service.call_count)
+
+    def test_unregister__local_names(self):
+        self.instance.service_info = mock.Mock(spec_set=ServiceInfo)("test")
+        self._register_with_device_name("Some Name")
+        bare_service = self.transport.local_names[LOCAL_NAME_BARE][1]
+        device_service = self.transport.local_names[LOCAL_NAME_DEVICE][1]
+
+        self.transport.unregister()
+
+        self.zeroconf.unregister_service.assert_any_call(bare_service)
+        self.zeroconf.unregister_service.assert_any_call(device_service)
+        self.assertEqual({}, self.transport.local_names)
+        self.listener.mock.update_local_names.assert_called_with([])
+
     @pytest.mark.skipif(ZEROCONF_NEEDS_UPDATE, reason="Needs updated Zeroconf")
     @mock.patch(BROADCAST_MODULE + "ZeroconfNetworkDiscovery.renew")
     def test_update_returns_false_when_addresses_unchanged(self, mock_renew):