Detect available non-loopback address for tests

Some environments disallow '0.0.0.0'

Author: nocarryr

doc/source/conftest.py

@@ -1,6 +1,14 @@
 # from loguru import logger
 import pytest
 import asyncio
+import socket
+
+@pytest.fixture(scope='session')
+def non_loopback_hostaddr():
+    hostname, aliases, addrs = socket.gethostbyname_ex(socket.gethostname())
+    addrs = [addr for addr in addrs if addr != '127.0.0.1']
+    assert len(addrs)
+    return addrs[0]
 
 
 @pytest.fixture(scope='function')
@@ -13,13 +21,11 @@ def new_loop(request, doctest_namespace):
     loop.close()
     policy.set_event_loop(None)
 
-def receiver_setup(request, loop):
-
+def receiver_setup(request, loop, hostaddr):
     cleanup_coro = None
-
     from tslumd import UmdSender, TallyType, TallyColor
 
-    sender = UmdSender(clients=[('0.0.0.0', 65000)], all_off_on_close=False)
+    sender = UmdSender(clients=[(hostaddr, 65000)], all_off_on_close=False)
     screen_index = 1
 
     async def open_sender():
@@ -50,11 +56,11 @@ async def cleanup():
     asyncio.set_event_loop_policy(None)
 
 @pytest.fixture(scope="function", autouse=True)
-def doctest_stuff(request, new_loop, doctest_namespace):
+def doctest_stuff(request, new_loop, non_loopback_hostaddr):
     node_name = request.node.name
     loop = asyncio.get_event_loop()
     assert loop is new_loop
     if node_name == 'receiver.rst':
-        yield from receiver_setup(request, new_loop)
+        yield from receiver_setup(request, new_loop, non_loopback_hostaddr)
     else:
         yield

tests/conftest.py

@@ -1,5 +1,6 @@
 from pathlib import Path
 import json
+import socket
 import pytest
 
 from tslumd import TallyColor, Message, Display
@@ -9,6 +10,13 @@
 MESSAGE_FILE = DATA_DIR / 'uhs500-message.umd'
 MESSAGE_JSON = DATA_DIR / 'uhs500-tally.json'
 
+@pytest.fixture(scope='session')
+def non_loopback_hostaddr():
+    hostname, aliases, addrs = socket.gethostbyname_ex(socket.gethostname())
+    addrs = [addr for addr in addrs if addr != '127.0.0.1']
+    assert len(addrs)
+    return addrs[0]
+
 @pytest.fixture
 def uhs500_msg_bytes() -> bytes:
     """Real message data received from an AV-UHS500 switcher

tests/test_receiver.py

@@ -158,21 +158,30 @@ async def test_broadcast_display(uhs500_msg_bytes, uhs500_msg_parsed, udp_endpoi
 
 
 @pytest.mark.asyncio
-async def test_rebind(uhs500_msg_bytes, uhs500_msg_parsed, udp_endpoint, udp_port, unused_tcp_port_factory):
+async def test_rebind(
+    uhs500_msg_bytes,
+    uhs500_msg_parsed,
+    udp_endpoint,
+    udp_port,
+    unused_tcp_port_factory,
+    non_loopback_hostaddr
+):
     transport, protocol, endpoint_port = udp_endpoint
     assert udp_port != endpoint_port
 
+    host_addrs = ['127.0.0.1', non_loopback_hostaddr]
+
     loop = asyncio.get_event_loop()
 
-    receiver = UmdReceiver(hostaddr='127.0.0.1', hostport=udp_port)
+    receiver = UmdReceiver(hostaddr=host_addrs[0], hostport=udp_port)
 
     evt_listener = EventListener()
     receiver.bind_async(loop, on_tally_added=evt_listener.callback)
 
     async with receiver:
 
         # Send message bytes to receiver and wait for ``on_tally_added`` events
-        transport.sendto(uhs500_msg_bytes, ('127.0.0.1', udp_port))
+        transport.sendto(uhs500_msg_bytes, (host_addrs[0], udp_port))
         _ = await evt_listener.get()
         while not evt_listener.empty():
             _ = await evt_listener.get()
@@ -185,13 +194,13 @@ async def test_rebind(uhs500_msg_bytes, uhs500_msg_parsed, udp_endpoint, udp_por
 
 
         # Change bind address and trigger a change
-        await receiver.set_hostaddr('0.0.0.0')
-        assert receiver.hostaddr == '0.0.0.0'
+        await receiver.set_hostaddr(host_addrs[1])
+        assert receiver.hostaddr == host_addrs[1]
 
         disp = uhs500_msg_parsed.displays[0]
         disp.brightness = 1
 
-        transport.sendto(uhs500_msg_parsed.build_message(), ('0.0.0.0', udp_port))
+        transport.sendto(uhs500_msg_parsed.build_message(), (host_addrs[1], udp_port))
 
         evt_args, evt_kwargs = await evt_listener.get()
         evt_tally = evt_args[0]
@@ -205,7 +214,7 @@ async def test_rebind(uhs500_msg_bytes, uhs500_msg_parsed, udp_endpoint, udp_por
         assert receiver.hostport == new_port
 
         disp.brightness = 2
-        transport.sendto(uhs500_msg_parsed.build_message(), ('0.0.0.0', new_port))
+        transport.sendto(uhs500_msg_parsed.build_message(), (host_addrs[1], new_port))
 
         evt_args, evt_kwargs = await evt_listener.get()
         evt_tally = evt_args[0]