找到了这个资料 也尝试了改 conftest.py 文件的
# conftest.py
@pytest.fixture(scope='session')
def celery_config():
return {
'broker_url': 'redis://127.0.0.1:6379/0',
'result_backend': 'redis://127.0.0.1:6379/0',
}
# test_file_a.py
@pytest.mark.celery_app
@pytest.mark.celery_worker
@pytest.mark.asyncio
async def test_invoice_ws(self, async_client):
async with async_client.websocket_connect(f"/server_a/api/v1/assets/long_query") as ws:
await ws.send_json(query)
....
并不起效果
# pip list | grep celery
celery == 5.0.0
pytest-celery == 0.0.0
# python --version
Python 3.9.4
求大佬们给个能跑的 demo 抄一下 Orz
1
z740713651 OP 已解决
celery_app.conf['task_always_eager'] = True 搞定 |