Merge pull request #5 from relativityspace-vtang/Fix_IPv6_tests

Fix IPv6 tests failing when no default gateway is found

.github/workflows/test.yml

@@ -1,6 +1,6 @@
 name: test
 
-on: [push]
+on: [push, pull_request]
 
 jobs:
   test:
@@ -22,5 +22,29 @@ jobs:
         uses: snok/install-poetry@v1
       - name: Install project
         run: poetry install --no-interaction
+      - name: Configure IP routes # See "Development Setup.md"
+        if: ${{ matrix.os == 'ubuntu-latest' }}
+        run: sudo ip -6 route add table local ff11::/16 dev lo
       - name: Run Pytest
         run: poetry run pytest
+
+  my-py:
+    strategy:
+      matrix:
+        os: ["ubuntu-latest", "windows-latest", "macos-latest"]
+    defaults:
+      run:
+        shell: bash
+    runs-on: ${{ matrix.os }}
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.8"
+      - name: Install Poetry
+        uses: snok/install-poetry@v1
+      - name: Install project
+        run: poetry install --no-interaction
+      - name: Run Mypy
+        run: poetry run mypy multicast_expert

multicast_expert/os_multicast.py

@@ -1,6 +1,6 @@
 # Script containing the nitty-gritty OS level functions of multicast_expert.
 
-import platform
+import sys
 import socket
 import struct
 import ctypes
@@ -11,7 +11,7 @@
 import netifaces
 
 # Import needed Win32 DLL functions
-if is_windows:
+if sys.platform == 'win32':
     iphlpapi = ctypes.WinDLL('iphlpapi')
     win32_GetAdapterIndex = iphlpapi.GetAdapterIndex
     win32_GetAdapterIndex.argtypes = [ctypes.c_wchar_p, ctypes.POINTER(ctypes.c_ulong)]
@@ -44,7 +44,7 @@ def iface_name_to_index(iface_name: str) -> int:
     Convert a network interface's name into its interface index.
     """
 
-    if is_windows:
+    if sys.platform == 'win32':
 
         # To get the if index on Windows we have to use the GetAdapterIndex() function.
         # docs here: https://docs.microsoft.com/en-us/windows/win32/api/iphlpapi/nf-iphlpapi-getadapterindex

multicast_expert/tx_socket.py

@@ -62,7 +62,12 @@ def __enter__(self) -> McastTxSocket:
 
         # Open the socket and set options
         self.socket = socket.socket(family=self.addr_family, type=socket.SOCK_DGRAM)
-        self.socket.bind((self.iface_ip, 0)) # Bind to any available port
+
+        # Note: for Unix IPv6, need to specify the scope ID in the bind address
+        if self.addr_family == socket.AF_INET6 and not is_windows:
+            self.socket.bind((self.iface_ip, 0, 0, self.iface_info.iface_idx))
+        else:
+            self.socket.bind((self.iface_ip, 0)) # Bind to any available port
 
         # Use the IP_MULTICAST_IF option to set the interface to use.
         os_multicast.set_multicast_if(self.socket, self.mcast_ips, self.iface_info, self.addr_family)

tests/test_multicast_expert.py

@@ -2,6 +2,7 @@
 import socket
 import pytest
 import platform
+import warnings
 
 # Test constants
 mcast_address_v4 = '239.2.2.2'
@@ -12,6 +13,22 @@
 test_string_alternate = b'Test of Multicast -- alternate address!'
 port = 12345
 
+@pytest.fixture()
+def nonloopback_iface_ipv6() -> str:
+    """Try to obtain a non-loopback IPv6 interface. If the default interface cannot be found, then use an arbitrary interface."""
+    nonloopback_iface_ipv6 = multicast_expert.get_default_gateway_iface_ip_v6()
+    if nonloopback_iface_ipv6 is None:
+        for iface_ip in multicast_expert.get_interface_ips(include_ipv4=False, include_ipv6=True):
+            if iface_ip != multicast_expert.LOCALHOST_IPV6:
+                nonloopback_iface_ipv6 = iface_ip
+                break
+
+        if nonloopback_iface_ipv6 is None:
+            raise RuntimeError("Couldn't find an ipv6 interface to use for the test!")
+
+        warnings.warn(f"netifaces was not able to determine the default ipv6 gateway on this machine. Using arbitrarily selected interface {nonloopback_iface_ipv6} instead.")
+    return nonloopback_iface_ipv6
+
 
 def test_get_ifaces() -> None:
     """
@@ -44,17 +61,17 @@ def test_tx_v4_can_be_used() -> None:
         mcast_sock.sendto(b'Hello IPv4', (mcast_address_v4, port))
 
 
-def test_tx_v6_can_be_used() -> None:
+def test_tx_v6_can_be_used(nonloopback_iface_ipv6: str) -> None:
     """
     Sanity check that a Tx IPv6 socket can be opened and used using the default gateway
     :return:
     """
 
-    with multicast_expert.McastTxSocket(socket.AF_INET6, mcast_ips=[mcast_address_v6]) as mcast_sock:
+    with multicast_expert.McastTxSocket(socket.AF_INET6, mcast_ips=[mcast_address_v6], iface_ip=nonloopback_iface_ipv6) as mcast_sock:
         mcast_sock.sendto(b'Hello IPv6', (mcast_address_v6, port))
 
 
-def test_non_mcast_raises_error() -> None:
+def test_non_mcast_raises_error(nonloopback_iface_ipv6: str) -> None:
     """
     Check that trying to use a non-multicast address raises an error
     """
@@ -63,7 +80,7 @@ def test_non_mcast_raises_error() -> None:
         multicast_expert.McastTxSocket(socket.AF_INET, mcast_ips=['239.2.2.2', '192.168.5.1'])
 
     with pytest.raises(multicast_expert.MulticastExpertError, match="not a multicast address"):
-        multicast_expert.McastTxSocket(socket.AF_INET6, mcast_ips=['abcd::'])
+        multicast_expert.McastTxSocket(socket.AF_INET6, mcast_ips=['abcd::'], iface_ip=nonloopback_iface_ipv6)
 
 
 def test_rx_v4_can_be_opened() -> None:
@@ -84,12 +101,12 @@ def test_rx_v4_ssm_can_be_opened() -> None:
         pass
 
 
-def test_rx_v6_can_be_opened() -> None:
+def test_rx_v6_can_be_opened(nonloopback_iface_ipv6: str) -> None:
     """
     Sanity check that a Rx IPv6 socket can be opened using the default gateway
     """
 
-    with multicast_expert.McastRxSocket(socket.AF_INET6, mcast_ips=[mcast_address_v6], port=port) as mcast_sock:
+    with multicast_expert.McastRxSocket(socket.AF_INET6, mcast_ips=[mcast_address_v6], port=port, iface_ips=[nonloopback_iface_ipv6]) as mcast_sock:
         pass