Method test_abort_clients

Lib/test/test_asyncio/test_server.py:215–267

(self)

        self.assertTrue(task.done())

    async def test_abort_clients(self):
        async def serve(rd, wr):
            fut.set_result((rd, wr))
            await wr.wait_closed()

        fut = asyncio.Future()
        srv = await asyncio.start_server(serve, socket_helper.HOSTv4, 0)
        self.addCleanup(srv.close)

        addr = srv.sockets[0].getsockname()
        (c_rd, c_wr) = await asyncio.open_connection(addr[0], addr[1], limit=4096)
        self.addCleanup(c_wr.close)

        (s_rd, s_wr) = await fut

        # Limit the socket buffers so we can more reliably overfill them
        s_sock = s_wr.get_extra_info('socket')
        s_sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)
        c_sock = c_wr.get_extra_info('socket')
        c_sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)

        # Get the reader in to a paused state by sending more than twice
        # the configured limit
        s_wr.write(b'a' * 4096)
        s_wr.write(b'a' * 4096)
        s_wr.write(b'a' * 4096)
        while c_wr.transport.is_reading():
            await asyncio.sleep(0)

        # Get the writer in a waiting state by sending data until the
        # kernel stops accepting more data in the send buffer.
        # gh-122136: getsockopt() does not reliably report the buffer size
        # available for message content.
        # We loop until we start filling up the asyncio buffer.
        # To avoid an infinite loop we cap at 10 times the expected value
        c_bufsize = c_sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
        s_bufsize = s_sock.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF)
        for i in range(10):
            s_wr.write(b'a' * c_bufsize)
            s_wr.write(b'a' * s_bufsize)
            if s_wr.transport.get_write_buffer_size() > 0:
                break
        self.assertNotEqual(s_wr.transport.get_write_buffer_size(), 0)

        task = asyncio.create_task(srv.wait_closed())
        await asyncio.sleep(0)
        self.assertFalse(task.done())

        srv.close()
        srv.abort_clients()
        await asyncio.sleep(0)
        await asyncio.sleep(0)
        self.assertTrue(task.done())

    async def test_close_with_hanging_client(self):
        # Synchronize server cancellation only after the socket connection is