[feature request] Support running test suite in parallel with pytest-xdist

[Apteryks]

Hello,

The test suite takes more than 2 minutes to run on a fast machine; with pytest-xdist, it's down to 6 s, but there are 3 test failures that are probably caused by resources contention when run concurrently:

=================================== FAILURES ===================================
_________ test_server_interrupt[capture_signal_async-Signals.SIGTERM] __________
[gw23] linux -- Python 3.10.7 /gnu/store/jjzbav1s0x8lqg1aal13irspwwblwdf4-python-wrapper-3.10.7/bin/python

self = <uvicorn.server.Server object at 0x7ffff41f8550>, sockets = None

    async def startup(self, sockets: list[socket.socket] | None = None) -> None:
        await self.lifespan.startup()
        if self.lifespan.should_exit:
            self.should_exit = True
            return
    
        config = self.config
    
        def create_protocol(
            _loop: asyncio.AbstractEventLoop | None = None,
        ) -> asyncio.Protocol:
            return config.http_protocol_class(  # type: ignore[call-arg]
                config=config,
                server_state=self.server_state,
                app_state=self.lifespan.state,
                _loop=_loop,
            )
    
        loop = asyncio.get_running_loop()
    
        listeners: Sequence[socket.SocketType]
        if sockets is not None:  # pragma: full coverage
            # Explicitly passed a list of open sockets.
            # We use this when the server is run from a Gunicorn worker.
    
            def _share_socket(
                sock: socket.SocketType,
            ) -> socket.SocketType:  # pragma py-linux pragma: py-darwin
                # Windows requires the socket be explicitly shared across
                # multiple workers (processes).
                from socket import fromshare  # type: ignore[attr-defined]
    
                sock_data = sock.share(os.getpid())  # type: ignore[attr-defined]
                return fromshare(sock_data)
    
            self.servers: list[asyncio.base_events.Server] = []
            for sock in sockets:
                is_windows = platform.system() == "Windows"
                if config.workers > 1 and is_windows:  # pragma: py-not-win32
                    sock = _share_socket(sock)  # type: ignore[assignment]
                server = await loop.create_server(create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog)
                self.servers.append(server)
            listeners = sockets
    
        elif config.fd is not None:  # pragma: py-win32
            # Use an existing socket, from a file descriptor.
            sock = socket.fromfd(config.fd, socket.AF_UNIX, socket.SOCK_STREAM)
            server = await loop.create_server(create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog)
            assert server.sockets is not None  # mypy
            listeners = server.sockets
            self.servers = [server]
    
        elif config.uds is not None:  # pragma: py-win32
            # Create a socket using UNIX domain socket.
            uds_perms = 0o666
            if os.path.exists(config.uds):
                uds_perms = os.stat(config.uds).st_mode  # pragma: full coverage
            server = await loop.create_unix_server(
                create_protocol, path=config.uds, ssl=config.ssl, backlog=config.backlog
            )
            os.chmod(config.uds, uds_perms)
            assert server.sockets is not None  # mypy
            listeners = server.sockets
            self.servers = [server]
    
        else:
            # Standard case. Create a socket from a host/port pair.
            try:
>               server = await loop.create_server(
                    create_protocol,
                    host=config.host,
                    port=config.port,
                    ssl=config.ssl,
                    backlog=config.backlog,
                )

uvicorn/server.py:162: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <_UnixSelectorEventLoop running=False closed=False debug=False>
protocol_factory = <function Server.startup.<locals>.create_protocol at 0x7fffefd57760>
host = '127.0.0.1', port = 8000

    async def create_server(
            self, protocol_factory, host=None, port=None,
            *,
            family=socket.AF_UNSPEC,
            flags=socket.AI_PASSIVE,
            sock=None,
            backlog=100,
            ssl=None,
            reuse_address=None,
            reuse_port=None,
            ssl_handshake_timeout=None,
            start_serving=True):
        """Create a TCP server.
    
        The host parameter can be a string, in that case the TCP server is
        bound to host and port.
    
        The host parameter can also be a sequence of strings and in that case
        the TCP server is bound to all hosts of the sequence. If a host
        appears multiple times (possibly indirectly e.g. when hostnames
        resolve to the same IP address), the server is only bound once to that
        host.
    
        Return a Server object which can be used to stop the service.
    
        This method is a coroutine.
        """
        if isinstance(ssl, bool):
            raise TypeError('ssl argument must be an SSLContext or None')
    
        if ssl_handshake_timeout is not None and ssl is None:
            raise ValueError(
                'ssl_handshake_timeout is only meaningful with ssl')
    
        if sock is not None:
            _check_ssl_socket(sock)
    
        if host is not None or port is not None:
            if sock is not None:
                raise ValueError(
                    'host/port and sock can not be specified at the same time')
    
            if reuse_address is None:
                reuse_address = os.name == 'posix' and sys.platform != 'cygwin'
            sockets = []
            if host == '':
                hosts = [None]
            elif (isinstance(host, str) or
                  not isinstance(host, collections.abc.Iterable)):
                hosts = [host]
            else:
                hosts = host
    
            fs = [self._create_server_getaddrinfo(host, port, family=family,
                                                  flags=flags)
                  for host in hosts]
            infos = await tasks.gather(*fs)
            infos = set(itertools.chain.from_iterable(infos))
    
            completed = False
            try:
                for res in infos:
                    af, socktype, proto, canonname, sa = res
                    try:
                        sock = socket.socket(af, socktype, proto)
                    except socket.error:
                        # Assume it's a bad family/type/protocol combination.
                        if self._debug:
                            logger.warning('create_server() failed to create '
                                           'socket.socket(%r, %r, %r)',
                                           af, socktype, proto, exc_info=True)
                        continue
                    sockets.append(sock)
                    if reuse_address:
                        sock.setsockopt(
                            socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
                    if reuse_port:
                        _set_reuseport(sock)
                    # Disable IPv4/IPv6 dual stack support (enabled by
                    # default on Linux) which makes a single socket
                    # listen on both address families.
                    if (_HAS_IPv6 and
                            af == socket.AF_INET6 and
                            hasattr(socket, 'IPPROTO_IPV6')):
                        sock.setsockopt(socket.IPPROTO_IPV6,
                                        socket.IPV6_V6ONLY,
                                        True)
                    try:
                        sock.bind(sa)
                    except OSError as err:
>                       raise OSError(err.errno, 'error while attempting '
                                      'to bind on address %r: %s'
                                      % (sa, err.strerror.lower())) from None
E                                     OSError: [Errno 98] error while attempting to bind on address ('127.0.0.1', 8000): address already in use

/gnu/store/jjcka1g6sk2cvwx8nm4fdwpdq3vll0v0-python-3.10.7/lib/python3.10/asyncio/base_events.py:1505: OSError

During handling of the above exception, another exception occurred:

exception_signal = <Signals.SIGTERM: 15>
capture_signal = <function capture_signal_async at 0x7ffff41e7e20>

    @pytest.mark.parametrize("exception_signal", signals)
    @pytest.mark.parametrize("capture_signal", signal_captures)
    async def test_server_interrupt(
        exception_signal: signal.Signals, capture_signal: Callable[[signal.Signals], ContextManager[None]]
    ):  # pragma: py-win32
        """Test interrupting a Server that is run explicitly inside asyncio"""
    
        async def interrupt_running(srv: Server):
            while not srv.started:
                await asyncio.sleep(0.01)
            signal.raise_signal(exception_signal)
    
        server = Server(Config(app=dummy_app, loop="asyncio"))
        asyncio.create_task(interrupt_running(server))
        with capture_signal(exception_signal) as witness:
>           await server.serve()

tests/test_server.py:77: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
uvicorn/server.py:69: in serve
    await self._serve(sockets)
uvicorn/server.py:84: in _serve
    await self.startup(sockets=sockets)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <uvicorn.server.Server object at 0x7ffff41f8550>, sockets = None

    async def startup(self, sockets: list[socket.socket] | None = None) -> None:
        await self.lifespan.startup()
        if self.lifespan.should_exit:
            self.should_exit = True
            return
    
        config = self.config
    
        def create_protocol(
            _loop: asyncio.AbstractEventLoop | None = None,
        ) -> asyncio.Protocol:
            return config.http_protocol_class(  # type: ignore[call-arg]
                config=config,
                server_state=self.server_state,
                app_state=self.lifespan.state,
                _loop=_loop,
            )
    
        loop = asyncio.get_running_loop()
    
        listeners: Sequence[socket.SocketType]
        if sockets is not None:  # pragma: full coverage
            # Explicitly passed a list of open sockets.
            # We use this when the server is run from a Gunicorn worker.
    
            def _share_socket(
                sock: socket.SocketType,
            ) -> socket.SocketType:  # pragma py-linux pragma: py-darwin
                # Windows requires the socket be explicitly shared across
                # multiple workers (processes).
                from socket import fromshare  # type: ignore[attr-defined]
    
                sock_data = sock.share(os.getpid())  # type: ignore[attr-defined]
                return fromshare(sock_data)
    
            self.servers: list[asyncio.base_events.Server] = []
            for sock in sockets:
                is_windows = platform.system() == "Windows"
                if config.workers > 1 and is_windows:  # pragma: py-not-win32
                    sock = _share_socket(sock)  # type: ignore[assignment]
                server = await loop.create_server(create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog)
                self.servers.append(server)
            listeners = sockets
    
        elif config.fd is not None:  # pragma: py-win32
            # Use an existing socket, from a file descriptor.
            sock = socket.fromfd(config.fd, socket.AF_UNIX, socket.SOCK_STREAM)
            server = await loop.create_server(create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog)
            assert server.sockets is not None  # mypy
            listeners = server.sockets
            self.servers = [server]
    
        elif config.uds is not None:  # pragma: py-win32
            # Create a socket using UNIX domain socket.
            uds_perms = 0o666
            if os.path.exists(config.uds):
                uds_perms = os.stat(config.uds).st_mode  # pragma: full coverage
            server = await loop.create_unix_server(
                create_protocol, path=config.uds, ssl=config.ssl, backlog=config.backlog
            )
            os.chmod(config.uds, uds_perms)
            assert server.sockets is not None  # mypy
            listeners = server.sockets
            self.servers = [server]
    
        else:
            # Standard case. Create a socket from a host/port pair.
            try:
                server = await loop.create_server(
                    create_protocol,
                    host=config.host,
                    port=config.port,
                    ssl=config.ssl,
                    backlog=config.backlog,
                )
            except OSError as exc:
                logger.error(exc)
                await self.lifespan.shutdown()
>               sys.exit(1)
E               SystemExit: 1

uvicorn/server.py:172: SystemExit
----------------------------- Captured stderr call -----------------------------
INFO:     Started server process [690]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
ERROR:    [Errno 98] error while attempting to bind on address ('127.0.0.1', 8000): address already in use
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
------------------------------ Captured log call -------------------------------
INFO     uvicorn.error:server.py:82 Started server process [690]
INFO     uvicorn.error:on.py:48 Waiting for application startup.
INFO     uvicorn.error:on.py:62 Application startup complete.
ERROR    uvicorn.error:server.py:170 [Errno 98] error while attempting to bind on address ('127.0.0.1', 8000): address already in use
INFO     uvicorn.error:on.py:67 Waiting for application shutdown.
INFO     uvicorn.error:on.py:76 Application shutdown complete.
__________ test_server_interrupt[capture_signal_sync-Signals.SIGTERM] __________
[gw29] linux -- Python 3.10.7 /gnu/store/jjzbav1s0x8lqg1aal13irspwwblwdf4-python-wrapper-3.10.7/bin/python

self = <uvicorn.server.Server object at 0x7ffff41fbd60>, sockets = None

    async def startup(self, sockets: list[socket.socket] | None = None) -> None:
        await self.lifespan.startup()
        if self.lifespan.should_exit:
            self.should_exit = True
            return
    
        config = self.config
    
        def create_protocol(
            _loop: asyncio.AbstractEventLoop | None = None,
        ) -> asyncio.Protocol:
            return config.http_protocol_class(  # type: ignore[call-arg]
                config=config,
                server_state=self.server_state,
                app_state=self.lifespan.state,
                _loop=_loop,
            )
    
        loop = asyncio.get_running_loop()
    
        listeners: Sequence[socket.SocketType]
        if sockets is not None:  # pragma: full coverage
            # Explicitly passed a list of open sockets.
            # We use this when the server is run from a Gunicorn worker.
    
            def _share_socket(
                sock: socket.SocketType,
            ) -> socket.SocketType:  # pragma py-linux pragma: py-darwin
                # Windows requires the socket be explicitly shared across
                # multiple workers (processes).
                from socket import fromshare  # type: ignore[attr-defined]
    
                sock_data = sock.share(os.getpid())  # type: ignore[attr-defined]
                return fromshare(sock_data)
    
            self.servers: list[asyncio.base_events.Server] = []
            for sock in sockets:
                is_windows = platform.system() == "Windows"
                if config.workers > 1 and is_windows:  # pragma: py-not-win32
                    sock = _share_socket(sock)  # type: ignore[assignment]
                server = await loop.create_server(create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog)
                self.servers.append(server)
            listeners = sockets
    
        elif config.fd is not None:  # pragma: py-win32
            # Use an existing socket, from a file descriptor.
            sock = socket.fromfd(config.fd, socket.AF_UNIX, socket.SOCK_STREAM)
            server = await loop.create_server(create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog)
            assert server.sockets is not None  # mypy
            listeners = server.sockets
            self.servers = [server]
    
        elif config.uds is not None:  # pragma: py-win32
            # Create a socket using UNIX domain socket.
            uds_perms = 0o666
            if os.path.exists(config.uds):
                uds_perms = os.stat(config.uds).st_mode  # pragma: full coverage
            server = await loop.create_unix_server(
                create_protocol, path=config.uds, ssl=config.ssl, backlog=config.backlog
            )
            os.chmod(config.uds, uds_perms)
            assert server.sockets is not None  # mypy
            listeners = server.sockets
            self.servers = [server]
    
        else:
            # Standard case. Create a socket from a host/port pair.
            try:
>               server = await loop.create_server(
                    create_protocol,
                    host=config.host,
                    port=config.port,
                    ssl=config.ssl,
                    backlog=config.backlog,
                )

uvicorn/server.py:162: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <_UnixSelectorEventLoop running=False closed=False debug=False>
protocol_factory = <function Server.startup.<locals>.create_protocol at 0x7fffefa303a0>
host = '127.0.0.1', port = 8000

    async def create_server(
            self, protocol_factory, host=None, port=None,
            *,
            family=socket.AF_UNSPEC,
            flags=socket.AI_PASSIVE,
            sock=None,
            backlog=100,
            ssl=None,
            reuse_address=None,
            reuse_port=None,
            ssl_handshake_timeout=None,
            start_serving=True):
        """Create a TCP server.
    
        The host parameter can be a string, in that case the TCP server is
        bound to host and port.
    
        The host parameter can also be a sequence of strings and in that case
        the TCP server is bound to all hosts of the sequence. If a host
        appears multiple times (possibly indirectly e.g. when hostnames
        resolve to the same IP address), the server is only bound once to that
        host.
    
        Return a Server object which can be used to stop the service.
    
        This method is a coroutine.
        """
        if isinstance(ssl, bool):
            raise TypeError('ssl argument must be an SSLContext or None')
    
        if ssl_handshake_timeout is not None and ssl is None:
            raise ValueError(
                'ssl_handshake_timeout is only meaningful with ssl')
    
        if sock is not None:
            _check_ssl_socket(sock)
    
        if host is not None or port is not None:
            if sock is not None:
                raise ValueError(
                    'host/port and sock can not be specified at the same time')
    
            if reuse_address is None:
                reuse_address = os.name == 'posix' and sys.platform != 'cygwin'
            sockets = []
            if host == '':
                hosts = [None]
            elif (isinstance(host, str) or
                  not isinstance(host, collections.abc.Iterable)):
                hosts = [host]
            else:
                hosts = host
    
            fs = [self._create_server_getaddrinfo(host, port, family=family,
                                                  flags=flags)
                  for host in hosts]
            infos = await tasks.gather(*fs)
            infos = set(itertools.chain.from_iterable(infos))
    
            completed = False
            try:
                for res in infos:
                    af, socktype, proto, canonname, sa = res
                    try:
                        sock = socket.socket(af, socktype, proto)
                    except socket.error:
                        # Assume it's a bad family/type/protocol combination.
                        if self._debug:
                            logger.warning('create_server() failed to create '
                                           'socket.socket(%r, %r, %r)',
                                           af, socktype, proto, exc_info=True)
                        continue
                    sockets.append(sock)
                    if reuse_address:
                        sock.setsockopt(
                            socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
                    if reuse_port:
                        _set_reuseport(sock)
                    # Disable IPv4/IPv6 dual stack support (enabled by
                    # default on Linux) which makes a single socket
                    # listen on both address families.
                    if (_HAS_IPv6 and
                            af == socket.AF_INET6 and
                            hasattr(socket, 'IPPROTO_IPV6')):
                        sock.setsockopt(socket.IPPROTO_IPV6,
                                        socket.IPV6_V6ONLY,
                                        True)
                    try:
                        sock.bind(sa)
                    except OSError as err:
>                       raise OSError(err.errno, 'error while attempting '
                                      'to bind on address %r: %s'
                                      % (sa, err.strerror.lower())) from None
E                                     OSError: [Errno 98] error while attempting to bind on address ('127.0.0.1', 8000): address already in use

/gnu/store/jjcka1g6sk2cvwx8nm4fdwpdq3vll0v0-python-3.10.7/lib/python3.10/asyncio/base_events.py:1505: OSError

During handling of the above exception, another exception occurred:

exception_signal = <Signals.SIGTERM: 15>
capture_signal = <function capture_signal_sync at 0x7ffff41e7d00>

    @pytest.mark.parametrize("exception_signal", signals)
    @pytest.mark.parametrize("capture_signal", signal_captures)
    async def test_server_interrupt(
        exception_signal: signal.Signals, capture_signal: Callable[[signal.Signals], ContextManager[None]]
    ):  # pragma: py-win32
        """Test interrupting a Server that is run explicitly inside asyncio"""
    
        async def interrupt_running(srv: Server):
            while not srv.started:
                await asyncio.sleep(0.01)
            signal.raise_signal(exception_signal)
    
        server = Server(Config(app=dummy_app, loop="asyncio"))
        asyncio.create_task(interrupt_running(server))
        with capture_signal(exception_signal) as witness:
>           await server.serve()

tests/test_server.py:77: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
uvicorn/server.py:69: in serve
    await self._serve(sockets)
uvicorn/server.py:84: in _serve
    await self.startup(sockets=sockets)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <uvicorn.server.Server object at 0x7ffff41fbd60>, sockets = None

    async def startup(self, sockets: list[socket.socket] | None = None) -> None:
        await self.lifespan.startup()
        if self.lifespan.should_exit:
            self.should_exit = True
            return
    
        config = self.config
    
        def create_protocol(
            _loop: asyncio.AbstractEventLoop | None = None,
        ) -> asyncio.Protocol:
            return config.http_protocol_class(  # type: ignore[call-arg]
                config=config,
                server_state=self.server_state,
                app_state=self.lifespan.state,
                _loop=_loop,
            )
    
        loop = asyncio.get_running_loop()
    
        listeners: Sequence[socket.SocketType]
        if sockets is not None:  # pragma: full coverage
            # Explicitly passed a list of open sockets.
            # We use this when the server is run from a Gunicorn worker.
    
            def _share_socket(
                sock: socket.SocketType,
            ) -> socket.SocketType:  # pragma py-linux pragma: py-darwin
                # Windows requires the socket be explicitly shared across
                # multiple workers (processes).
                from socket import fromshare  # type: ignore[attr-defined]
    
                sock_data = sock.share(os.getpid())  # type: ignore[attr-defined]
                return fromshare(sock_data)
    
            self.servers: list[asyncio.base_events.Server] = []
            for sock in sockets:
                is_windows = platform.system() == "Windows"
                if config.workers > 1 and is_windows:  # pragma: py-not-win32
                    sock = _share_socket(sock)  # type: ignore[assignment]
                server = await loop.create_server(create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog)
                self.servers.append(server)
            listeners = sockets
    
        elif config.fd is not None:  # pragma: py-win32
            # Use an existing socket, from a file descriptor.
            sock = socket.fromfd(config.fd, socket.AF_UNIX, socket.SOCK_STREAM)
            server = await loop.create_server(create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog)
            assert server.sockets is not None  # mypy
            listeners = server.sockets
            self.servers = [server]
    
        elif config.uds is not None:  # pragma: py-win32
            # Create a socket using UNIX domain socket.
            uds_perms = 0o666
            if os.path.exists(config.uds):
                uds_perms = os.stat(config.uds).st_mode  # pragma: full coverage
            server = await loop.create_unix_server(
                create_protocol, path=config.uds, ssl=config.ssl, backlog=config.backlog
            )
            os.chmod(config.uds, uds_perms)
            assert server.sockets is not None  # mypy
            listeners = server.sockets
            self.servers = [server]
    
        else:
            # Standard case. Create a socket from a host/port pair.
            try:
                server = await loop.create_server(
                    create_protocol,
                    host=config.host,
                    port=config.port,
                    ssl=config.ssl,
                    backlog=config.backlog,
                )
            except OSError as exc:
                logger.error(exc)
                await self.lifespan.shutdown()
>               sys.exit(1)
E               SystemExit: 1

uvicorn/server.py:172: SystemExit
----------------------------- Captured stderr call -----------------------------
INFO:     Started server process [923]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
ERROR:    [Errno 98] error while attempting to bind on address ('127.0.0.1', 8000): address already in use
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
------------------------------ Captured log call -------------------------------
INFO     uvicorn.error:server.py:82 Started server process [923]
INFO     uvicorn.error:on.py:48 Waiting for application startup.
INFO     uvicorn.error:on.py:62 Application startup complete.
ERROR    uvicorn.error:server.py:170 [Errno 98] error while attempting to bind on address ('127.0.0.1', 8000): address already in use
INFO     uvicorn.error:on.py:67 Waiting for application shutdown.
INFO     uvicorn.error:on.py:76 Application shutdown complete.
_______________________________ test_socket_bind _______________________________
[gw11] linux -- Python 3.10.7 /gnu/store/jjzbav1s0x8lqg1aal13irspwwblwdf4-python-wrapper-3.10.7/bin/python

self = <uvicorn.config.Config object at 0x7ffff40298d0>

    def bind_socket(self) -> socket.socket:
        logger_args: list[str | int]
        if self.uds:  # pragma: py-win32
            path = self.uds
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            try:
                sock.bind(path)
                uds_perms = 0o666
                os.chmod(self.uds, uds_perms)
            except OSError as exc:  # pragma: full coverage
                logger.error(exc)
                sys.exit(1)
    
            message = "Uvicorn running on unix socket %s (Press CTRL+C to quit)"
            sock_name_format = "%s"
            color_message = "Uvicorn running on " + click.style(sock_name_format, bold=True) + " (Press CTRL+C to quit)"
            logger_args = [self.uds]
        elif self.fd:  # pragma: py-win32
            sock = socket.fromfd(self.fd, socket.AF_UNIX, socket.SOCK_STREAM)
            message = "Uvicorn running on socket %s (Press CTRL+C to quit)"
            fd_name_format = "%s"
            color_message = "Uvicorn running on " + click.style(fd_name_format, bold=True) + " (Press CTRL+C to quit)"
            logger_args = [sock.getsockname()]
        else:
            family = socket.AF_INET
            addr_format = "%s://%s:%d"
    
            if self.host and ":" in self.host:  # pragma: full coverage
                # It's an IPv6 address.
                family = socket.AF_INET6
                addr_format = "%s://[%s]:%d"
    
            sock = socket.socket(family=family)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            try:
>               sock.bind((self.host, self.port))
E               OSError: [Errno 98] Address already in use

uvicorn/config.py:514: OSError

During handling of the above exception, another exception occurred:

    def test_socket_bind() -> None:
        config = Config(app=asgi_app)
        config.load()
>       sock = config.bind_socket()

tests/test_config.py:265: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <uvicorn.config.Config object at 0x7ffff40298d0>

    def bind_socket(self) -> socket.socket:
        logger_args: list[str | int]
        if self.uds:  # pragma: py-win32
            path = self.uds
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            try:
                sock.bind(path)
                uds_perms = 0o666
                os.chmod(self.uds, uds_perms)
            except OSError as exc:  # pragma: full coverage
                logger.error(exc)
                sys.exit(1)
    
            message = "Uvicorn running on unix socket %s (Press CTRL+C to quit)"
            sock_name_format = "%s"
            color_message = "Uvicorn running on " + click.style(sock_name_format, bold=True) + " (Press CTRL+C to quit)"
            logger_args = [self.uds]
        elif self.fd:  # pragma: py-win32
            sock = socket.fromfd(self.fd, socket.AF_UNIX, socket.SOCK_STREAM)
            message = "Uvicorn running on socket %s (Press CTRL+C to quit)"
            fd_name_format = "%s"
            color_message = "Uvicorn running on " + click.style(fd_name_format, bold=True) + " (Press CTRL+C to quit)"
            logger_args = [sock.getsockname()]
        else:
            family = socket.AF_INET
            addr_format = "%s://%s:%d"
    
            if self.host and ":" in self.host:  # pragma: full coverage
                # It's an IPv6 address.
                family = socket.AF_INET6
                addr_format = "%s://[%s]:%d"
    
            sock = socket.socket(family=family)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            try:
                sock.bind((self.host, self.port))
            except OSError as exc:  # pragma: full coverage
                logger.error(exc)
>               sys.exit(1)
E               SystemExit: 1

uvicorn/config.py:517: SystemExit
----------------------------- Captured stderr call -----------------------------
ERROR:    [Errno 98] Address already in use
------------------------------ Captured log call -------------------------------
ERROR    uvicorn.error:config.py:516 [Errno 98] Address already in use
=========================== short test summary info ============================
SKIPPED [1] tests/supervisors/test_multiprocess.py:110: platform unsupports SIGBREAK
SKIPPED [1] tests/supervisors/test_reload.py:89: Needed dependency not installed
SKIPPED [1] tests/supervisors/test_reload.py:76: Needed dependency not installed
SKIPPED [1] tests/supervisors/test_reload.py:335: watchfiles not available
SKIPPED [1] tests/supervisors/test_reload.py:352: watchfiles not available
SKIPPED [1] tests/supervisors/test_reload.py:114: Needed dependency not installed
SKIPPED [1] tests/supervisors/test_reload.py:143: Needed dependency not installed
SKIPPED [1] tests/supervisors/test_reload.py:170: Needed dependency not installed
SKIPPED [1] tests/supervisors/test_reload.py:102: Needed dependency not installed
SKIPPED [1] tests/supervisors/test_reload.py:131: Needed dependency not installed
SKIPPED [1] tests/supervisors/test_reload.py:182: Needed dependency not installed
SKIPPED [1] tests/supervisors/test_reload.py:202: Needed dependency not installed
SKIPPED [1] tests/supervisors/test_reload.py:255: Needed dependency not installed
SKIPPED [1] tests/supervisors/test_reload.py:281: watchfiles not available
================== 3 failed, 805 passed, 14 skipped in 6.03s ===================

[Kludex]

PR welcome