Method test

Lib/test/test_asyncio/test_server_context.py:28–54

()

    def test_start_server1(self):
        # Test that asyncio.start_server captures the context at the time of server creation
        async def test():
            var = contextvars.ContextVar("var", default="default")

            async def handle_client(reader, writer):
                value = var.get()
                writer.write(value.encode())
                await writer.drain()
                writer.close()

            server = await asyncio.start_server(handle_client, '127.0.0.1', 0,
                                               ssl=self.server_ssl_context)
            # change the value
            var.set("after_server")

            async def client(addr):
                reader, writer = await asyncio.open_connection(*addr,
                                                               ssl=self.client_ssl_context)
                data = await reader.read(100)
                writer.close()
                await writer.wait_closed()
                return data.decode()

            async with server:
                addr = server.sockets[0].getsockname()
                self.assertEqual(await client(addr), "default")

            self.assertEqual(var.get(), "after_server")

        self.run_coro(test())