test_retry.py 文件源码

python
阅读 26 收藏 0 点赞 0 评论 0

项目:nameko-amqp-retry 作者: nameko 项目源码 文件源码
def test_message(self, container_factory, rabbit_config):

        from examples.retry import Service

        container = container_factory(Service, rabbit_config)
        container.start()

        timestamp = arrow.utcnow().replace(seconds=+1)

        publish = publisher(rabbit_config)
        with entrypoint_waiter(
            container, 'handle_message', callback=wait_for_result
        ) as result:
            payload = {'timestamp': timestamp.isoformat()}
            publish(payload, routing_key="messages")

        res = result.get()
        assert arrow.get(re.match("Time is (.+)", res).group(1)) >= timestamp
评论列表
文章目录


问题


面经


文章

微信
公众号

扫码关注公众号