我对您的场景没有一个清晰的概念！根据我的理解，您可以使用相同的方法创建连接和通道，以便在需要时可以发布、使用、Assert和停止使用
希望这对你有帮助！

def test_rabbitmq():
    from pika import BlockingConnection, ConnectionParameters, PlainCredentials

    conn = BlockingConnection(ConnectionParameters(host='host', virtual_host='vhost', credentials=PlainCredentials('username', 'password')))
    channel = conn.channel()

    # define your consumer
    def on_message(channel, method_frame, header_frame, body):
        message = body.decode()
        # assert your message here
        # asset message == 'value'
        channel.basic_cancel('test-consumer')  # stops the consumer

    # define your publisher
    def publish_message(message):
        channel.basic_publish(exchange='', routing_key='', body=message')

    publish('your message')
    tag = channel.basic_consume(queue='queue', on_message_callback=on_message, consumer_tag='test-consumer')

stop_consuming-取消所有使用者，并发出退出start_consuming循环的信号。
basic_cancel-此方法取消消费者。消费者标记将作为输入。
close-关闭连接/通道
Reference

以下是pytest-rabbitmq插件的示例：

import asyncio
import rabbitpy

async def test_queue(fastapi_client, mocker, rabbitmq):
    mock_consumer_method = mocker.MagicMock(name="consumer_method")
    mocker.patch(
        "my.project.consumer_method",
        new=mock_consumer_method,
    )

    connection_details = rabbitmq.args
    os.environ[
        "RABBITMQ_CONNECTION_STRING"
    ] = f"amqp://{connection_details['username']}:{connection_details['password']}@{connection_details['host']}:{connection_details['port']}/"  # this is used by my consumer

    channel = rabbitmq.channel()
    queue = rabbitpy.Queue(
        channel, name="QUEUE_NAME", durable=False, message_ttl=5 * 60 * 1000
    )
    queue.declare()
    message = rabbitpy.Message(channel, "some message")
    message.publish(exchange="", routing_key=PRODUCE_QUEUE_NAME)

    with fastapi_client:  # triggers the consumer startup event
        await asyncio.sleep(20)

    mock_consumer_method.assert_called_once_with("some message")

我的FastAPI服务器中已经有一个消费者，您的用例可能会有所不同。