Fix address receiver bug

posttroll/address_receiver.py

@@ -54,15 +54,15 @@
 zero_seconds = dt.timedelta(seconds=0)
 
 
-def get_configured_address_port():
+def get_configured_address_port() -> int:
     return config.get("address_publish_port", DEFAULT_ADDRESS_PUBLISH_PORT)
 
 
 def get_local_ips():
     """Get local IP addresses."""
     inet_addrs = [netifaces.ifaddresses(iface).get(netifaces.AF_INET)
                   for iface in netifaces.interfaces()]
-    ips = []
+    ips:list[str] = []
     for addr in inet_addrs:
         if addr is not None:
             for add in addr:
@@ -132,7 +132,7 @@ def _check_age(self, pub, min_interval=zero_seconds):
         if (now - self._last_age_check) <= min_interval:
             return
 
-        LOGGER.debug("%s - checking addresses", str(dt.datetime.now(dt.timezone.utc)))
+        logger.debug("%s - checking addresses", str(dt.datetime.now(dt.timezone.utc)))
         self._last_age_check = now
         to_del = []
         with self._address_lock:

posttroll/tests/test_nameserver.py

@@ -18,7 +18,7 @@
 from posttroll.subscriber import Subscribe
 
 
-def free_port():
+def free_port() -> int:
     """Get a free port.
 
     From https://gist.github.com/bertjwregeer/0be94ced48383a42e70c3d9fff1f4ad0
@@ -39,7 +39,7 @@ def free_port():
     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     s.bind(("0.0.0.0", 0))
-    portnum = s.getsockname()[1]
+    portnum:int = s.getsockname()[1]
     s.close()
 
     return portnum
@@ -252,6 +252,35 @@ def test_noisypublisher_heartbeat():
         thr.join()
 
 
+def test_noisypublisher_heartbeat_no_multicast():
+    """Test that the heartbeat in the NoisyPublisher works with multicast disabled."""
+    from posttroll.publisher import NoisyPublisher
+    from posttroll.subscriber import Subscribe
+
+    min_interval = 10
+
+    try:
+        with config.set(address_publish_port=free_port(), nameserver_port=free_port()):
+            ns_ = NameServer(multicast_enabled=False)
+            thr = Thread(target=ns_.run)
+            thr.start()
+
+            pub = NoisyPublisher("test", nameservers=["localhost"])
+            pub.start()
+            time.sleep(0.1)
+
+            with Subscribe("test", topics="/heartbeat/test", nameserver="localhost") as sub:
+                time.sleep(0.1)
+                pub.heartbeat(min_interval=min_interval)
+                msg = next(sub.recv(1))
+            assert msg.type == "beat"
+            assert msg.data == {"min_interval": min_interval}
+    finally:
+        pub.stop()
+        ns_.stop()
+        thr.join()
+
+
 def test_switch_backend_for_nameserver():
     """Test switching backend for nameserver."""
     with config.set(backend="spurious_backend"):