Tests trying to find unused ports are flakey when automated
- uvloop version: 0.17.0
- Python version: 3.10
- Platform: nix/darwin
-
Can you reproduce the bug with
PYTHONASYNCIODEBUGin env?: N/A - Does uvloop behave differently from vanilla asyncio? How?: N/A
The tests using find_free_port (like https://github.com/MagicStack/uvloop/blob/d2deffefa18653636eb03ea77a5dab6e4febf6c6/tests/test_tcp.py#L252) may fail in busy CI environments. For example building for nixpkgs involves running tests every time. It ends up with: (https://hydra.nixos.org/build/197525892/nixlog/1)
=================================== FAILURES ===================================
_______________________ Test_UV_TCP.test_create_server_5 _______________________
Traceback (most recent call last):
File "/private/tmp/tmp.xkcj6kx4o4/tests/test_tcp.py", line 245, in test_create_server_5
self.loop.run_until_complete(runner())
File "uvloop/loop.pyx", line 1517, in uvloop.loop.Loop.run_until_complete
File "/private/tmp/tmp.xkcj6kx4o4/tests/test_tcp.py", line 238, in runner
srv = await self.loop.create_server(
File "uvloop/loop.pyx", line 1790, in create_server
OSError: [Errno 48] error while attempting to bind on address ('::', 50000, 0, 0): address already in use
_______________________ Test_UV_TCP.test_create_server_6 _______________________
Traceback (most recent call last):
File "/private/tmp/tmp.xkcj6kx4o4/tests/test_tcp.py", line 271, in test_create_server_6
self.loop.run_until_complete(runner())
File "uvloop/loop.pyx", line 1517, in uvloop.loop.Loop.run_until_complete
File "/private/tmp/tmp.xkcj6kx4o4/tests/test_tcp.py", line 255, in runner
srv1 = await self.loop.create_server(
File "uvloop/loop.pyx", line 1790, in create_server
OSError: [Errno 48] error while attempting to bind on address ('::', 50000, 0, 0): address already in use
=============================== warnings summary ===============================
Do those tests need to use a common ipv4/6 socket? Could they use only one protocol at a time with port 0 and then get the local port number from the result?
Or maybe build the retry loop around the first socket creation rather than 2 independent steps?
import asyncio import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
class Test_UV_TCP: async def create_server(self, host, port): server = await self.loop.create_server(self.factory, host, port) return server
async def runner(self):
self.factory = ... # Create your factory here
for attempt in range(5):
try:
srv = await self.create_server('::', 0)
local_port = srv.sockets[0].getsockname()[1]
srv.close()
await srv.wait_closed()
srv = await self.create_server('::', local_port)
srv.close()
await srv.wait_closed()
break
except OSError as exc:
if attempt < 4 and exc.errno == 48:
continue
raise
def test_create_server_5(self):
self.loop.run_until_complete(self.runner())