我有一个用于 FastAPI 的
app/main.py
,它可以:
import qdrant_client as QdrantClient
...
qdrant_client = QdrantClient(url=...)
qdrant_client.create_collection(...)
...
app = FastAPI()
...
@app.get("/my_endpoint")
def handle_my_endpoint(query: str):
...
qdrant_client.query_points(...)
我想为
/my_endpoint
写一个单元。
所以我在单元测试文件中有:
from fastapi.testclient import TestClient
from app.main import app
class Test(unittest.TestCase):
@patch('app.main.QdrantClient', qdrant_client=MagicMock())
def test_my_endpoint(self, qdrant_client):
qdrant_client.create_collection.return_value = None
with TestClient(app) as client:
resp = client.get("/my_endpoint", params={...})
...
但它不起作用,它坚持实际使用真正的客户端来尝试真正调用 Qdrant 服务器,但失败了......我缺少什么来正确设置
patch
?
重构您的代码以获取
qdrant_client
作为依赖项:
import qdrant_client as QdrantClient
...
qdrant_client = QdrantClient(url=...)
qdrant_client.create_collection(...)
...
app = FastAPI()
...
async def get_qdrant_client():
return qdrant_client
@app.get("/my_endpoint")
def handle_my_endpoint(query: str, qdrant_client: Annotated[QdrantClient, Depends(get_qdrant_client)]):
...
qdrant_client.query_points(...)
然后,使用标准 FastAPI 机制进行覆盖:
from fastapi.testclient import TestClient
from app.main import app, get_qdrant_client
@pytest.fixture(name="qdrant_client")
def get_qdrant_client_fixture():
return MagicMock()
@pytest.fixture(name="test_client")
def get_test_client_fixture(qdrant_client):
async def get_qdrant_client_mock():
return qdrant_client
app.dependency_overrides[get_qdrant_client] = get_qdrant_client_mock
with TestClient(app) as client:
yield client
app.dependency_overrides.clear()
class Test(unittest.TestCase):
def test_my_endpoint(self, test_client: TestClient, qdrant_client):
qdrant_client.create_collection.return_value = None
resp = client.get("/my_endpoint", params={...})
...