Method test_close_clients

Lib/test/test_asyncio/test_server.py:190–213

(self)

        await srv.wait_closed()

    async def test_close_clients(self):
        async def serve(rd, wr):
            try:
                await rd.read()
            finally:
                wr.close()
                await wr.wait_closed()

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

        addr = srv.sockets[0].getsockname()
        (rd, wr) = await asyncio.open_connection(addr[0], addr[1])
        self.addCleanup(wr.close)

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

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

    async def test_abort_clients(self):
        async def serve(rd, wr):