求助： celery pytest 的单元怎么写

[z740713651]

找到了这个资料 也尝试了改 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

[z740713651]

已解决
celery_app.conf['task_always_eager'] = True
搞定