LogicPublisher

def get_payloads(self) -> List[Tuple[AnyDict, str]]:
    payloads: List[Tuple[AnyDict, str]] = []

    if self._schema:
        params = {"response__": (self._schema, ...)}

        call_model: CallModel[Any, Any] = CallModel(
            call=lambda: None,
            model=create_model("Fake"),
            response_model=create_model(  # type: ignore[call-overload]
                "",
                __config__=get_config_base(),  # type: ignore[arg-type]
                **params,  # type: ignore[arg-type]
            ),
            params=params,
        )

        body = get_response_schema(
            call_model,
            prefix=f"{self.name}:Message",
        )
        if body:  # pragma: no branch
            payloads.append((body, ""))

    else:
        for call in self.calls:
            call_model = build_call_model(call)
            body = get_response_schema(
                call_model,
                prefix=f"{self.name}:Message",
            )
            if body:
                payloads.append((body, to_camelcase(unwrap(call).__name__)))

    return payloads

@abstractproperty
def name(self) -> str:
    """Returns the name of the API operation."""
    raise NotImplementedError()

@override
async def publish(  # type: ignore[override]
    self,
    message: SendableMessage,
    channel: Union[str, PubSub, None] = None,
    reply_to: str = "",
    headers: Optional[AnyDict] = None,
    correlation_id: Optional[str] = None,
    *,
    list: Union[str, ListSub, None] = None,
    stream: Union[str, StreamSub, None] = None,
    rpc: bool = False,
    rpc_timeout: Optional[float] = 30.0,
    raise_timeout: bool = False,
) -> Optional[DecodedMessage]:
    assert self._producer, NOT_CONNECTED_YET  # nosec B101

    channel = PubSub.validate(channel or self.channel)
    list = ListSub.validate(list or self.list)
    stream = StreamSub.validate(stream or self.stream)

    assert any((channel, list, stream)), "You have to specify outgoing channel"  # nosec B101

    headers_to_send = (self.headers or {}).copy()
    if headers is not None:
        headers_to_send.update(headers)

    if getattr(list, "batch", False):
        await self._producer.publish_batch(
            *cast(Sequence[SendableMessage], message),
            list=list.name,  # type: ignore[union-attr]
        )
        return None

    else:
        return await self._producer.publish(
            message=message,
            channel=getattr(channel, "name", None),
            list=getattr(list, "name", None),
            stream=getattr(stream, "name", None),
            maxlen=getattr(stream, "maxlen", None),
            reply_to=reply_to or self.reply_to,
            correlation_id=correlation_id,
            headers=headers_to_send,
            rpc=rpc,
            rpc_timeout=rpc_timeout,
            raise_timeout=raise_timeout,
        )

def reset_test(self) -> None:
    self._fake_handler = False
    self.mock = None

def schema(self) -> Dict[str, Channel]:  # pragma: no cover
    """Returns the schema of the API operation as a dictionary of channel names and channel objects."""
    return {}

def set_test(
    self,
    mock: MagicMock,
    with_fake: bool,
) -> None:
    self.mock = mock
    self._fake_handler = with_fake