如何测试将数据发送到接收器的浮士德代理？答案

【问题标题】：How to test a Faust agent that sends data to a sink?如何测试将数据发送到接收器的浮士德代理？
【发布时间】：2020-09-29 12:47:47
【问题描述】：

我正在尝试使用 pytest 为我的 Faust 应用程序编写单元测试。我参考了文档here，但它没有提到当我的浮士德代理向接收器发送数据时该怎么做。

没有接收器，我的测试工作正常，但是当我使用接收器时，我收到此错误：

RuntimeError: Task <Task pending name='Task-2' coro=<Agent._execute_actor() running at /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/site-packages/faust/agents/agent.py:647> cb=[<TaskWakeupMethWrapper object at 0x7fc28967c5b0>()]> got Future <Future pending> attached to a different loop
INFO     faust.agents.agent:logging.py:265 [^-AgentTestWrapper: ml_exporter.processDetections]: Stopping...

我尝试了各种方法：例如修补我的 Faust 应用程序中将数据发送到接收器的装饰器，尝试在没有装饰器的情况下测试我的功能（通过尝试绕过它），修补我的接收器参数Faust 应用程序具有 None 值（因此它不会将我的数据发送到接收器）等。我对这些都没有运气。

这是我的浮士德特工：

app = faust.App('ml-exporter', broker=dx_broker, value_serializer='json')

detection_topic = app.topic(dx_topic)
graph_topic = app.topic(gwh_topic)

@app.agent(detection_topic, sink=[graph_topic])
async def processDetections(detections):
    detection_count = 0
    async for detection in detections:
        detection_count += 1
        # r.set("detection_count", detection_count)
        yield detection

这是我当前的测试代码：

import ml_exporter

patch('ml_exporter.graph_topic', None)

def create_app():
    return faust.App('ml-exporter', value_serializer='json')

@pytest.fixture()
def test_app(event_loop):
    app = create_app()
    app.finalize()
    app.flow_control.resume()
    return app

@pytest.mark.asyncio()
async def test_processDetections(test_app):
    async with ml_exporter.processDetections.test_context() as agent:
        event = await agent.put('hey')
        assert agent.results[event.message.offset] == 'hey'

我在运行此测试时遇到与上述相同的错误。有什么方法可以成功测试我的 Faust 应用程序吗？

谢谢！

【问题讨论】：

标签： python python-3.x pytest faust

【解决方案1】：

强制 pytest 使用 Faust 的 asyncio 事件循环作为默认的全局循环。将以下夹具添加到您的测试代码中：

@pytest.mark.asyncio()
@pytest.fixture()
def event_loop():
    yield app.loop

如pytest documentation中所述：

event_loop 夹具可以在任何标准 pytest 位置（例如直接在测试文件中或在 conftest.py 中）轻松覆盖，以使用非默认事件循环。如果应用了pytest.mark.asyncio 标记，pytest 钩子将确保将生成的循环设置为默认的全局循环。

【讨论】：