allansimon/mock-amqp-server

mock-amqp-server

![]([***]

概述

这是一个可监控的Mock AMQP服务器（兼容RabbitMQ），用于在网络层面测试您的AMQP发布者和消费者，无需修改业务代码即可验证消息处理流程。

已测试的项目/库

• pika
• celery
• Kombu
• haigha

核心功能

允许构建测试基础设施，验证消息从发布到消费的全流程：

[ mock amqp server ]<--- 端口5672 --->[ 未修改的工作进程 ]<----网络-->[ 数据库 ]
        ^                                                                               ^   
        |                                                                               |
    端口80上的HTTP监控连接                                                               |
        |                                                                               |
[ 侧边运行的测试用例 ]------------------------------------------------------------------------

您可在CI测试中使用以下逻辑（伪代码）：

python
def test_worker_insert_valid_message_in_database():
    database_connection = create_connection_to_database()
    backdoor_amqp = connection_to_amqp_backdoor()
    
    backdoor_amqp.insert_message(message_id=42, text="hello", in_queue="messages_to_treat")
    backdoor_amqp.wait_message_being_ack(message_id=42, timeout=5)
    
    database_connection.assert_row_exists(id="42", content="hello")

为什么选择此镜像？

单元测试覆盖率高并不意味着集成流程正确。例如：

• 重构时颠倒回调注册与工作进程启动顺序，导致消息未处理；
• 切换配置方式时，第三方库的密码错误处理漏洞导致CPU占用100%，本地Mock库无法复现。此镜像可发现这类网络层面问题。

可用HTTP接口

• GET /: 浏览器访问查看内部状态；
• MOCK_FLUSH /: 重置Mock内部状态；
• DELETE /messages-in-queue/$QUEUE_NAME: 清空指定队列消息；
• DELETE /messages-in-exchange/$EXCHANGE_NAME: 清空指定交换机消息；
• GET /authentification-done-with-success-on/$USERNAME: 检查用户名是否成功连接；
• GET /messages-acknowledged/$DELIVERY_TAG: 阻塞直到消息被确认；
• GET /messages-not-acknowledged/$DELIVERY_TAG: 阻塞直到消息被拒绝；
• GET /messages-requeued/$DELIVERY_TAG: 阻塞直到消息被重新入队；
• GET /messages-in-queue/$QUEUE_NAME: 获取队列待消费消息（JSON格式）；
• GET /messages-in-exchange/$EXCHANGE_NAME: 获取交换机待消费消息（JSON格式）；
• GET /queue-bound-to-exchange/$QUEUE_NAME/$EXCHANGE_NAME: 阻塞直到队列绑定到交换机；
• POST /add-message-on/$EXCHANGE_NAME: 向交换机发送模拟消息；
• POST /add-message-in-queue/$QUEUE_NAME: 向队列发送模拟消息；
• POST /create-exchange/$EXCHANGE_NAME: 创建模拟交换机；
• POST /create-queue/$QUEUE_NAME: 创建模拟队列。

/add-message接口扩展

支持二进制数据发送，使用application/octet-stream类型，消息头以amqp_header_前缀传递：

python
import pickle
import requests

message_body = {
    'action': 'my_action',
    'payload': {'id':1, 'foo':'bar'}
}
message_body = pickle.dumps(message_body)

response = requests.post(
    f"{BASE_URL}/add-message-on/{EXCHANGE}",
    headers={
        'Content-Type': 'application/octet-stream',
        'amqp_header_x-delay': '200'
    },
    data=message_body
)
delivery_tag = int(response.content.decode('utf-8'))

Docker部署示例

单容器运行

bash
docker run -d -p 5672:5672 -p 80:80 allansimon/mock-amqp-server

• 5672: AMQP服务端口；
• 80: HTTP监控接口端口。

Docker Compose示例

yaml
version: '3'
services:
  mock-amqp:
    image: allansimon/mock-amqp-server
    ports:
      - "5672:5672"
      - "80:80"
    restart: always

特别感谢

感谢https://github.com/celery/py-amqp%E9%A1%B9%E7%9B%AE%E6%8F%90%E4%BE%9B%E7%9A%84%E7%B1%BB%E5%9E%8B%E5%BA%8F%E5%88%97%E5%8C%96/%E5%8F%8D%E5%BA%8F%E5%88%97%E5%8C%96%E4%BB%A3%E7%A0%81%E3%80%82