小编Nih*_*shi的帖子

我正在尝试使用 pytest 为我的 Faust 应用程序编写单元测试。我已经参考了此处的文档，但它没有提到当我的 Faust 代理将数据发送到接收器时要做什么。

如果没有水槽，我的测试工作正常，但是当我使用水槽时，我收到此错误：

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 …