Method hold

tests/server/test_sse_security.py:425–431

(self)

        self.write_stream: WriteStream[SessionMessage]

    async def hold(self) -> None:
        scope, _, _, _ = _sse_scope("GET", "/sse", None)
        async with self.transport.connect_sse(scope, self._receive, self._send) as (read, write):
            self.write_stream = write
            async with read, write, self._body_send, self._body_recv, self._read_send, self._read_recv:
                async for item in read:
                    await self._read_send.send(item)

    def disconnect(self) -> None:
        self._disconnected.set()