How to use the grpclib.exceptions.ProtocolError function in grpclib

async def test_no_response(stream, stub):
    with pytest.raises(ProtocolError, match='requires a single message'):
        async with stream:
            pass
    assert stub.__events__ == [
        SendHeaders(
            [(':status', '200'),
             ('content-type', 'application/grpc+proto'),
             ('grpc-status', str(Status.UNKNOWN.value)),
             ('grpc-message', 'Internal Server Error')],
            end_stream=True,
        ),
        Reset(ErrorCodes.NO_ERROR),
    ]

"""Coroutine to send trailers with trailing metadata to the client.

        This coroutine allows sending trailers-only responses, in case of some
        failure conditions during handling current request, i.e. when
        ``status is not OK``.

        .. note:: This coroutine will be called implicitly at exit from
            request handler, with appropriate status code, if not called
            explicitly during handler execution.

        :param status: resulting status of this coroutine call
        :param status_message: description for a status
        :param metadata: custom trailing metadata, dict or list of pairs
        """
        if self._send_trailing_metadata_done:
            raise ProtocolError('Trailing metadata was already sent')

        if (
            not self._cardinality.server_streaming
            and not self._send_message_done
            and status is Status.OK
        ):
            raise ProtocolError('Unary response with OK status requires '
                                'a single message to be sent')

        if self._send_initial_metadata_done:
            headers: _Headers = []
        else:
            # trailers-only response
            headers = [
                (':status', '200'),
                ('content-type', self._content_type),

.. note:: This coroutine will be called implicitly during first
            :py:meth:`recv_message` coroutine call, if not called before
            explicitly.

        May raise :py:class:`~grpclib.exceptions.GRPCError` if server returned
        non-:py:attr:`Status.OK ` in trailers-only
        response.

        When this coroutine finishes, you can access received initial metadata
        by using :py:attr:`initial_metadata` attribute.
        """
        if not self._send_request_done:
            raise ProtocolError('Request was not sent yet')

        if self._recv_initial_metadata_done:
            raise ProtocolError('Initial metadata was already received')

        with self._wrapper:
            headers = await self._stream.recv_headers()
            self._recv_initial_metadata_done = True
            headers_map = dict(headers)
            self._raise_for_status(headers_map)
            self._raise_for_content_type(headers_map)
            if 'grpc-status' in headers_map:  # trailers-only response
                self._trailers_only = True

                im = cast(_Metadata, MultiDict())
                im, = await self._dispatch.recv_initial_metadata(im)
                self.initial_metadata = im

                tm = decode_metadata(headers)
                tm, = await self._dispatch.recv_trailing_metadata(tm)

.. note:: This coroutine will be called implicitly at exit from
            this call (context manager's exit), if not called before explicitly.

        May raise :py:class:`~grpclib.exceptions.GRPCError` if server returned
        non-:py:attr:`Status.OK ` in trailers.

        When this coroutine finishes, you can access received trailing metadata
        by using :py:attr:`trailing_metadata` attribute.
        """
        if (not self._end_done  # explicit end
            and not (not self._cardinality.client_streaming  # implicit end
                     and self._send_message_done)):
            raise ProtocolError('Outgoing stream was not ended')

        if not self._recv_initial_metadata_done:
            raise ProtocolError('Initial metadata was not received before '
                                'waiting for trailing metadata')

        if self._recv_trailing_metadata_done:
            raise ProtocolError('Trailing metadata was already received')

        if self._trailers_only:
            self._recv_trailing_metadata_done = True
        else:
            with self._wrapper:
                trailers = await self._stream.recv_trailers()
                self._recv_trailing_metadata_done = True

                tm = decode_metadata(trailers)
                tm, = await self._dispatch.recv_trailing_metadata(tm)
                self.trailing_metadata = tm

First approach is preferred, because it doesn't require sending
        additional HTTP/2 frame.
        """
        if not self._send_request_done:
            await self.send_request()

        end_stream = end
        if not self._cardinality.client_streaming:
            if self._send_message_done:
                raise ProtocolError('Message was already sent')
            else:
                end_stream = True

        if self._end_done:
            raise ProtocolError('Stream is ended')

        with self._wrapper:
            message, = await self._dispatch.send_message(message)
            await send_message(self._stream, self._codec, message,
                               self._send_type, end=end_stream)
            self._send_message_done = True
            self._messages_sent += 1
            self._stream.connection.messages_sent += 1
            self._stream.connection.last_message_sent = time.monotonic()
            if end:
                self._end_done = True

async def cancel(self) -> None:
        """Coroutine to cancel this request/stream.

        Client will send RST_STREAM frame to the server, so it will be
        explicitly informed that there is nothing to expect from the client
        regarding this request/stream.
        """
        if not self._send_request_done:
            raise ProtocolError('Request was not sent yet')

        if self._cancel_done:
            raise ProtocolError('Stream was already cancelled')

        with self._wrapper:
            await self._stream.reset()  # TODO: specify error code
            self._cancel_done = True

async def func_wrapper_gen(*args: Any, **kwargs: Any) -> Any:  # pyre-ignore
        try:
            async for item in func(*args, **kwargs):
                yield item
        except GRPCError as e:
            raise IdbException(e.message) from e  # noqa B306
        except (ProtocolError, StreamTerminatedError) as e:
            raise IdbException(e.args) from e

It should be used to finally end stream from the client-side when we're
        finished sending messages to the server and stream wasn't closed with
        last DATA frame. See :py:meth:`send_message` for more details.

        HTTP/2 stream will have half-closed (local) state after this coroutine
        call.
        """
        if not self._send_request_done:
            raise ProtocolError('Request was not sent')

        if self._end_done:
            raise ProtocolError('Stream was already ended')

        if not self._cardinality.client_streaming:
            if not self._send_message_done:
                raise ProtocolError('Unary request requires a single message '
                                    'to be sent')
            else:
                # `send_message` must already ended stream
                self._end_done = True
                return
        else:
            await self._stream.end()
            self._end_done = True

async def cancel(self) -> None:
        """Coroutine to cancel this request/stream.

        Client will send RST_STREAM frame to the server, so it will be
        explicitly informed that there is nothing to expect from the client
        regarding this request/stream.
        """
        if not self._send_request_done:
            raise ProtocolError('Request was not sent yet')

        if self._cancel_done:
            raise ProtocolError('Stream was already cancelled')

        with self._wrapper:
            await self._stream.reset()  # TODO: specify error code
            self._cancel_done = True

'a single message to be sent: {!r}'
                              .format(self._method_name))
        else:
            status = Status.OK
            status_message = None
            status_details = None

        try:
            await self.send_trailing_metadata(status=status,
                                              status_message=status_message,
                                              status_details=status_details)
        except h2.exceptions.StreamClosedError:
            pass

        if protocol_error is not None:
            raise ProtocolError(protocol_error)

        # to suppress exception propagation
        return True