我在使用 TortoiseOrm 作为数据库层测试 FastApi 应用程序时遇到问题。请在下面找到一个最小的可重现示例。该示例是一个带有 2 个端点的小型 API，它们接受get请求并返回虚拟数据。一个端点与数据库交互，而另一个则不。
我的主要问题是，当我在浏览器或 Postman 上与它们交互时，虽然两个端点都能正常工作，但与数据库交互的端点在测试期间会引发错误。请在下面找到完整的堆栈跟踪。由于我难以理解的原因，似乎 TortoiseOrm 在测试期间无法建立数据库连接。
pytest使用命令运行测试
pytest -v -s -p no:warnings
主文件
from fastapi import FastAPI, status
from db import Tournament
from db import create_start_app_handler
def get_application():
app = FastAPI()
app.add_event_handler("startup", create_start_app_handler(app))
return app
app = get_application()
@app.get("/",
name="point1",
status_code=status.HTTP_200_OK
)
async def home():
return {
"title":"Hello world"
}
@app.get(
"/save/",
name="point2",
status_code=status.HTTP_200_OK
)
async def save_data():
await Tournament.create(
name="test2"
)
return {
"status":"created"
}
数据库.py
from fastapi import FastAPI
from tortoise.models import Model
from tortoise import fields
from tortoise.contrib.fastapi import register_tortoise
from typing import Callable
class Tournament(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=255)
async def init_db(app: FastAPI):
register_tortoise(
app,
db_url='postgres://admin:admin@localhost:5432/postgres_test',
modules={"models": ["db"]},
generate_schemas=True,
add_exception_handlers=True,
)
def create_start_app_handler(app: FastAPI) -> Callable:
async def start_app() -> None:
await init_db(app)
return start_app
test_main.py
import pytest
from fastapi.testclient import TestClient
from main import app
class TestBase:
client = TestClient(app)
def test_home(self) -> None:
response = self.client.get(
app.url_path_for('point1')
)
assert response.status_code == 200
def test_save(self) -> None:
response = self.client.get(
app.url_path_for('point2')
)
assert response.status_code == 200
堆栈跟踪
_____________________________________________________________________ TestBase.test_save ______________________________________________________________________
self = <test_main.TestBase object at 0x7fb7fb5831f0>
def test_save(self) -> None:
> response = self.client.get(
app.url_path_for('point2')
)
test_main.py:16:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../venv/lib/python3.8/site-packages/requests/sessions.py:555: in get
return self.request('GET', url, **kwargs)
../venv/lib/python3.8/site-packages/starlette/testclient.py:415: in request
return super().request(
../venv/lib/python3.8/site-packages/requests/sessions.py:542: in request
resp = self.send(prep, **send_kwargs)
../venv/lib/python3.8/site-packages/requests/sessions.py:655: in send
r = adapter.send(request, **kwargs)
../venv/lib/python3.8/site-packages/starlette/testclient.py:243: in send
raise exc from None
../venv/lib/python3.8/site-packages/starlette/testclient.py:240: in send
loop.run_until_complete(self.app(scope, receive, send))
/home/unyime/anaconda3/lib/python3.8/asyncio/base_events.py:616: in run_until_complete
return future.result()
../venv/lib/python3.8/site-packages/fastapi/applications.py:208: in __call__
await super().__call__(scope, receive, send)
../venv/lib/python3.8/site-packages/starlette/applications.py:112: in __call__
await self.middleware_stack(scope, receive, send)
../venv/lib/python3.8/site-packages/starlette/middleware/errors.py:181: in __call__
raise exc from None
../venv/lib/python3.8/site-packages/starlette/middleware/errors.py:159: in __call__
await self.app(scope, receive, _send)
../venv/lib/python3.8/site-packages/starlette/exceptions.py:82: in __call__
raise exc from None
../venv/lib/python3.8/site-packages/starlette/exceptions.py:71: in __call__
await self.app(scope, receive, sender)
../venv/lib/python3.8/site-packages/starlette/routing.py:580: in __call__
await route.handle(scope, receive, send)
../venv/lib/python3.8/site-packages/starlette/routing.py:241: in handle
await self.app(scope, receive, send)
../venv/lib/python3.8/site-packages/starlette/routing.py:52: in app
response = await func(request)
../venv/lib/python3.8/site-packages/fastapi/routing.py:219: in app
raw_response = await run_endpoint_function(
../venv/lib/python3.8/site-packages/fastapi/routing.py:152: in run_endpoint_function
return await dependant.call(**values)
main.py:31: in save_data
await Tournament.create(
../venv/lib/python3.8/site-packages/tortoise/models.py:1104: in create
db = kwargs.get("using_db") or cls._choose_db(True)
../venv/lib/python3.8/site-packages/tortoise/models.py:1014: in _choose_db
db = router.db_for_write(cls)
../venv/lib/python3.8/site-packages/tortoise/router.py:39: in db_for_write
return self._db_route(model, "db_for_write")
../venv/lib/python3.8/site-packages/tortoise/router.py:31: in _db_route
return get_connection(self._router_func(model, action))
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <tortoise.router.ConnectionRouter object at 0x7fb7fb6272e0>, model = <class 'db.Tournament'>, action = 'db_for_write'
def _router_func(self, model: Type["Model"], action: str):
> for r in self._routers:
E TypeError: 'NoneType' object is not iterable
../venv/lib/python3.8/site-packages/tortoise/router.py:18: TypeError
=================================================================== short test summary info ===================================================================
FAILED test_main.py::TestBase::test_save - TypeError: 'NoneType' object is not iterable
================================================================= 1 failed, 1 passed in 2.53s =================================================================
我该如何解决？你可以在这里找到 Github 上的完整代码。