Fre*_*nke 6 python unit-testing rabbitmq
我有一个使用 pika 和rabbitmq 用 python 编写的简单消费者。
消费者连接到rabbitmq并监听队列。当消息到达时，它会转换消息并将其发布到另一个队列上。
此处介绍： https: //bitbucket.org/snippets/fbanke/8e7zbX
我想制作测试用例来测试消费者和队列之间的交互。例如，我想确保当消息被消耗时，调用“basic_ack”函数让rabbitmq知道消息已被处理。
另一个测试用例是，如果连接断开，消费者是否重新连接到rabbitmq。
等等。我想测试消费者和队列之间的交互而不是消费者中的实际业务逻辑。
如果我模拟 pika 对象，它需要我 100% 理解 API 的行为方式，对 API 的任何误解都会导致错误的代码。通过测试的代码，但实际上不起作用。
我宁愿使用实时队列来测试消费者，并从测试中操作它以查看消费者的行为是否符合预期。
例如 1. 设置队列 2. 启动消费者 3. 向队列发布有效消息 4. 断言该消息已被工作线程消费
或者
是否存在关于如何执行此操作的最佳实践？我可以找到许多针对数据库的类似测试的示例，但不适用于队列。看来我需要在单独的线程中启动消费者并使用它，但似乎没有基础设施支持这一点。