Method tcp_server

Lib/test/test_asyncio/test_ssl.py:102–132

(self, server_prog, *,
                   family=socket.AF_INET,
                   addr=None,
                   timeout=support.SHORT_TIMEOUT,
                   backlog=1,
                   max_clients=10)

        super().tearDown()

    def tcp_server(self, server_prog, *,
                   family=socket.AF_INET,
                   addr=None,
                   timeout=support.SHORT_TIMEOUT,
                   backlog=1,
                   max_clients=10):

        if addr is None:
            if family == getattr(socket, "AF_UNIX", None):
                with tempfile.NamedTemporaryFile() as tmp:
                    addr = tmp.name
            else:
                addr = ('127.0.0.1', 0)

        sock = socket.socket(family, socket.SOCK_STREAM)

        if timeout is None:
            raise RuntimeError('timeout is required')
        if timeout <= 0:
            raise RuntimeError('only blocking sockets are supported')
        sock.settimeout(timeout)

        try:
            sock.bind(addr)
            sock.listen(backlog)
        except OSError as ex:
            sock.close()
            raise ex

        return TestThreadedServer(
            self, sock, server_prog, timeout, max_clients)

    def tcp_client(self, client_prog,
                   family=socket.AF_INET,