gpt4 book ai didi

python - 如何测试将数据发送到接收器的 Faust 代理?

转载 作者:行者123 更新时间:2023-12-03 23:46:44 31 4
gpt4 key购买 nike

我正在尝试使用 pytest 为我的 Faust 应用程序编写单元测试。我引用了文档 here但它没有提到当我的 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 值(所以它不会将我的数据发送到接收器)等。我对这些都不走运。

这是我的 Faust 代理:
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 应用程序吗?

谢谢!

最佳答案

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

@pytest.mark.asyncio()
@pytest.fixture()
def event_loop():
yield app.loop
pytest documentation 中所述:

The event_loop fixture can be easily overridden in any of the standard pytest locations (e.g. directly in the test file, or in conftest.py) to use a non-default event loop. If the pytest.mark.asyncio marker is applied, a pytest hook will ensure the produced loop is set as the default global loop.

关于python - 如何测试将数据发送到接收器的 Faust 代理?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/62297894/

31 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com