KafkaHandlerCallWrapper

async def assert_called_once_with(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
    key: Any = EMPTY,
    partition: Any = EMPTY,
) -> None:
    """Assert the endpoint was called once, with the message described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
        key: The exact record key, as the bytes the client delivered.
        partition: The exact partition the record was read from.
    """
    await self._assert_called_once_with(
        self._expected_call(
            body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
            key=key,
            partition=partition,
        )
    )

async def assert_called_with(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
    key: Any = EMPTY,
    partition: Any = EMPTY,
) -> None:
    """Assert the last message the endpoint saw is the one described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
        key: The exact record key, as the bytes the client delivered.
        partition: The exact partition the record was read from.
    """
    await self._assert_called_with(
        self._expected_call(
            body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
            key=key,
            partition=partition,
        )
    )

async def assert_any_call(
    self,
    body: Any = EMPTY,
    /,
    *,
    headers: Any = EMPTY,
    correlation_id: Any = EMPTY,
    reply_to: Any = EMPTY,
    content_type: Any = EMPTY,
    path: Any = EMPTY,
    context: Mapping[str, Any] = EMPTY,
    key: Any = EMPTY,
    partition: Any = EMPTY,
) -> None:
    """Assert one of the messages the endpoint saw is the one described here.

    Args:
        body: The body as a dict, a model or a matcher; it goes through the codec.
        headers: Headers the message must carry; the rest may carry more.
        correlation_id: The exact correlation id.
        reply_to: The exact reply-to destination.
        content_type: The exact content type.
        path: The exact path parameters the subject template matched.
        context: Context paths, as given to `Context()`, mapped to their values.
        key: The exact record key, as the bytes the client delivered.
        partition: The exact partition the record was read from.
    """
    await self._assert_any_call(
        self._expected_call(
            body,
            headers=headers,
            correlation_id=correlation_id,
            reply_to=reply_to,
            content_type=content_type,
            path=path,
            context=context,
            key=key,
            partition=partition,
        )
    )

async def call_wrapped(self, message: "StreamMessage[Any]") -> Any:
    """Calls the wrapped function with the given message."""
    assert self._wrapped_call, "You should use `set_wrapped` first"
    if self.is_test:
        await self._recorder.record(message)
    return await self._wrapped_call(message)

def set_wrapped(
    self,
    *,
    dependencies: Sequence["Dependant"],
    _call_decorators: Reversible["Decorator"],
    config: "FastDependsConfig",
) -> "CallModel":
    # Composed from the declaration rather than from the last composition:
    # `build_call` answers with the call it decorated, so reading that back
    # as the input would decorate it again, one more layer per build.
    dependent = config.build_call(
        self._declared_call,
        dependencies=dependencies,
        call_decorators=_call_decorators,
    )
    self._composed_call = dependent.original_call
    self._wrapped_call = dependent.wrapped_call
    return dependent.dependent

async def wait_call(self, timeout: float | None = None) -> None:
    """Waits for a call with an optional timeout."""
    assert self.future is not None, "You can use this method only with TestClient"
    with anyio.fail_after(timeout):
        await self.future

def set_test(self) -> None:
    self.is_test = True
    self.refresh(with_mock=True)

def reset_test(self) -> None:
    self.is_test = False
    self._recorder.reset()
    self._recorder.stop_mirroring()
    self.future = None

def trigger(
    self,
    result: Any = None,
    error: BaseException | None = None,
) -> None:
    if not self.is_test:
        return

    if self.future is None:
        msg = "You can use this method only with TestClient"
        raise SetupError(msg)

    if self.future.done():
        self.future = asyncio.Future()

    if error:
        self.future.set_exception(error)
        # Mark the mirrored error as retrieved to avoid unhandled-future reports.
        self.future.exception()

    else:
        self.future.set_result(result)

def refresh(self, with_mock: bool = False) -> None:
    if asyncio.events._get_running_loop() is not None:
        self.future = asyncio.Future()

    if with_mock:
        self._recorder.reset()