Add fix for Linux and fix unittests

Linux cannot receive broadcasts if IP is bound to
the host IP

Author: Jaxc

PyStageLinQ/PyStageLinQ.py

@@ -11,6 +11,7 @@
 import platform
 import psutil
 import ipaddress
+import os
 from typing import Callable
 
 from . import Device
@@ -39,9 +40,6 @@ def __init__(self, ip=None, discovery_port=51337):
         self.target_interfaces = []
         self.discovery_port = discovery_port
         self.get_interface_from_ip(ip)
-        logger.info(f"Found {len(self.target_interfaces)} network interfaces:")
-        for iface in self.target_interfaces:
-            logger.info(f"  - {iface.name}: {iface.addr_str}")
 
     def get_interface_from_ip(self, ip):
         if ip is None:
@@ -55,30 +53,43 @@ def get_interface_from_ip(self, ip):
             ip_list = ip
         else:
             raise TypeError
+        logger.info(
+            f"Found {len(psutil.net_if_stats().items())} total network interfaces, listing IPv4 interfaces:"
+        )
 
         for interface in psutil.net_if_stats().items():
             for interface_info in psutil.net_if_addrs()[interface[0]]:
                 # Only look for IPV4 binds
-                if socket.AF_INET == interface_info.family and (
-                    interface_info.address in ip_list or ip_list[0] == "any"
-                ):
-                    self.target_interfaces.append(
-                        PyStageLinQ_interface_info(
-                            interface[0],
-                            len(self.target_interfaces),
-                            int(ipaddress.IPv4Address(interface_info.address)),
-                            interface_info.address,
-                            int(ipaddress.IPv4Address(interface_info.netmask)),
-                            interface[1],
-                            0,
+                if socket.AF_INET == interface_info.family:
+                    logger.info(f"  - {interface[0]}: {interface_info.address}")
+                    if interface_info.address in ip_list or ip_list[0] == "any":
+                        self.target_interfaces.append(
+                            PyStageLinQ_interface_info(
+                                interface[0],
+                                len(self.target_interfaces),
+                                int(ipaddress.IPv4Address(interface_info.address)),
+                                interface_info.address,
+                                int(ipaddress.IPv4Address(interface_info.netmask)),
+                                interface[1],
+                                0,
+                            )
                         )
-                    )
+
+        logger.info(
+            f"{len(self.target_interfaces)} interfaces matched with requested interfaces and will be used by PyStageLinQ:"
+        )
+        for iface in self.target_interfaces:
+            logger.info(f"  - {iface.name}: {iface.addr_str}")
 
     def send_discovery_frame(self, discovery_frame):
         for interface in self.target_interfaces:
             try:
-                with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as discovery_socket:
-                    discovery_socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+                with socket.socket(
+                    socket.AF_INET, socket.SOCK_DGRAM
+                ) as discovery_socket:
+                    discovery_socket.setsockopt(
+                        socket.SOL_SOCKET, socket.SO_BROADCAST, 1
+                    )
                     discovery_socket.bind((interface.addr_str, 0))
                     discovery_socket.sendto(
                         discovery_frame,
@@ -182,16 +193,13 @@ def start(self):
 
     def _stop(self):
         logger.info(f"Stop requested, trying graceful shutdown")
-        try:
-            discovery = StageLinQDiscovery()
-            discovery_info = self.discovery_info
-            discovery_info.ConnectionType = ConnectionTypes.EXIT
-            discovery_frame = discovery.encode_frame(discovery_info)
+        discovery = StageLinQDiscovery()
+        discovery_info = self.discovery_info
+        discovery_info.ConnectionType = ConnectionTypes.EXIT
+        discovery_frame = discovery.encode_frame(discovery_info)
 
-            self.network_interface.send_discovery_frame(discovery_frame)
-            logger.info(f"Gracefully shutdown complete")
-        except Exception as e:
-            logger.debug('Could not send "EXIT" discovery frame during shutdown')
+        self.network_interface.send_discovery_frame(discovery_frame)
+        logger.info(f"Gracefully shutdown complete")
 
     def _announce_self(self):
         discovery = StageLinQDiscovery()
@@ -214,9 +222,11 @@ async def _discover_stagelinq_device(self, host_ip, timeout=10):
         # Create socket
         discover_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 
+        bind_ip = "" if os.name == "posix" else host_ip
+
         try:
             discover_socket.bind(
-                (host_ip, self.StageLinQ_discovery_port)
+                (bind_ip, self.StageLinQ_discovery_port)
             )  # bind socket to broadcast
         except Exception as e:
             # Cannot bind to socket, check if IP is correct and link is up
@@ -361,12 +371,16 @@ async def _periodic_announcement(self):
 
     async def _py_stagelinq_strapper(self):
         strapper_tasks = set()
-        logger.info(f"Looking for discovery frames on {len(self.network_interface.target_interfaces)} IP addresses:")
+        logger.info(
+            f"Looking for discovery frames on {len(self.network_interface.target_interfaces)} IP addresses:"
+        )
 
         for interface in self.network_interface.target_interfaces:
             logger.info(f"{interface.addr_str}")
             strapper_tasks.add(
-                asyncio.create_task(self._discover_stagelinq_device(interface.addr_str, timeout=2))
+                asyncio.create_task(
+                    self._discover_stagelinq_device(interface.addr_str, timeout=2)
+                )
             )
 
         while self.get_loop_condition():

tests/Main.py

@@ -53,7 +53,7 @@ def state_map_data_print(data):
 def main():
     logging.basicConfig(level=logging.INFO)
     global PrimeGo
-    ip_choice = 1
+    ip_choice = 2
     match ip_choice:
         case 0:
             PrimeGo = PyStageLinQ.PyStageLinQ(
@@ -69,6 +69,12 @@ def main():
                 name="Jaxcie StageLinQ",
                 ip=["169.254.13.37", "127.0.0.1"],
             )
+        case 3:
+            # should fail
+            PrimeGo = PyStageLinQ.PyStageLinQ(
+                new_device_found_callback, name="Jaxcie StageLinQ", ip="127.0.0.1"
+            )
+
     PrimeGo.start_standalone()
 
 

tests/unit/test_unit_PyStageLinQ.py

@@ -142,10 +142,10 @@ async def test_discover_stagelinq_device_bind_error(
 
     dummy_socket.socket.return_value.bind.side_effect = Exception()
 
-    with pytest.raises(Exception) as exception:
+    assert (
         await dummy_pystagelinq._discover_stagelinq_device(dummy_ip)
-
-    assert exception.type is Exception
+        == PyStageLinQError.CANNOTBINDSOCKET
+    )
 
 
 def test_get_loop_condition(dummy_pystagelinq):
@@ -176,7 +176,7 @@ async def test_discover_stagelinq_check_initialization(
         dummy_socket.AF_INET, dummy_socket.SOCK_DGRAM
     )
     dummy_socket.socket.return_value.bind.assert_called_once_with(
-        (dummy_ip, dummy_pystagelinq.StageLinQ_discovery_port)
+        ("", dummy_pystagelinq.StageLinQ_discovery_port)
     )
     dummy_socket.socket.return_value.setblocking.assert_called_once_with(False)
 
@@ -832,6 +832,11 @@ async def test_py_stagelinq_strapper(dummy_pystagelinq, monkeypatch):
     monkeypatch.setattr(
         dummy_pystagelinq, "_discover_stagelinq_device", discover_device_mock
     )
+    monkeypatch.setattr(
+        dummy_pystagelinq.network_interface,
+        "target_interfaces",
+        [PyStageLinQ.PyStageLinQ.PyStageLinQ_interface_info("", 0 , 0,"", 0, "", 0)]
+    )
 
     await dummy_pystagelinq._py_stagelinq_strapper()
 
@@ -850,6 +855,11 @@ async def test_py_stagelinq_strapper_loop_condition_false(
     monkeypatch.setattr(
         dummy_pystagelinq, "_discover_stagelinq_device", discover_device_mock
     )
+    monkeypatch.setattr(
+        dummy_pystagelinq.network_interface,
+        "target_interfaces",
+        [PyStageLinQ.PyStageLinQ.PyStageLinQ_interface_info("", 0 , 0,"", 0, "", 0)]
+    )
 
     await dummy_pystagelinq._py_stagelinq_strapper()
 
@@ -884,6 +894,9 @@ async def test_py_stagelinq_strapper_task_exception(
     monkeypatch.setattr(
         dummy_pystagelinq, "_discover_stagelinq_device", discover_device_mock
     )
+    monkeypatch.setattr(
+        dummy_pystagelinq.network_interface, "target_interfaces", [PyStageLinQ.PyStageLinQ.PyStageLinQ_interface_info("", 0 , 0,"", 0, "", 0)]
+    )
 
     get_loop_condition_mock.side_effect = [True, False]
 

tests/unit/test_unit_PyStageLinQ_interface_info.py

@@ -364,8 +364,7 @@ def test_send_discovery_frame_permission_error(
 
     monkeypatch.setattr(PyStageLinQ.PyStageLinQ, "socket", dummy_socket)
 
-    with pytest.raises(PermissionError) as exception:
-        dummy_PyStageLinQ_network_interface.send_discovery_frame(dummy_discovery_frame)
+    dummy_PyStageLinQ_network_interface.send_discovery_frame(dummy_discovery_frame)
 
     dummy_socket.socket.assert_called_once_with(
         dummy_socket.AF_INET, dummy_socket.SOCK_DGRAM