如何在python中使用RabbitMQ进行简单的单元测试？

pip install pika

import pika

# 建立与RabbitMQ服务器的连接
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建消息队列（如果不存在）
channel.queue_declare(queue='test_queue')

# 发布消息到队列
channel.basic_publish(exchange='', routing_key='test_queue', body='Hello RabbitMQ!')

# 关闭与RabbitMQ服务器的连接
connection.close()

import pika

# 定义消息处理函数
def process_message(ch, method, properties, body):
    print("Received message: %s" % body)

# 建立与RabbitMQ服务器的连接
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建消息队列（如果不存在）
channel.queue_declare(queue='test_queue')

# 注册消息处理函数
channel.basic_consume(queue='test_queue', on_message_callback=process_message, auto_ack=True)

# 开始消费消息
channel.start_consuming()

import unittest
import subprocess

class RabbitMQTestCase(unittest.TestCase):
    def test_rabbitmq_integration(self):
        # 运行消费者代码（启动消息消费）
        consumer_process = subprocess.Popen(['python', 'consumer.py'])
        
        # 运行生产者代码（发布消息）
        producer_process = subprocess.Popen(['python', 'producer.py'])
        
        # 等待消费者和生产者进程结束
        producer_process.wait()
        consumer_process.terminate()
        
        # 断言消费者是否成功接收到消息
        # 你可以自定义断言条件，根据具体情况进行判断
        
        self.assertTrue(True)

if __name__ == '__main__':
    unittest.main()

python test_rabbitmq.py