求助： celery pytest 的单元怎么写

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