英文:
Running Pytest test cases in transaction isolation in a FastAPI setup
问题
我有一个使用FastAPI、MySQL和asyncio
的应用程序。
我一直在尝试将一些测试用例与我的应用程序集成,具有在每个测试用例后回滚更改的能力,以便所有测试用例可以独立运行。
这是我的控制器设置方式,注入了一个DB依赖项。
from sqlalchemy.ext.asyncio import create_async_engine
async def get_db_connection_dependency():
engine = create_async_engine("mysql+aiomysql://root:root@mysql8:3306/user_db")
connection = engine.connect()
return connection
class UserController:
async def create_user(
self,
request: Request,
connection: AsyncConnection = Depends(get_db_connection_dependency)
) -> JSONResponse:
# START TRANSACTION
await connection.__aenter__()
transaction = connection.begin()
await transaction.__aenter__()
try:
do_stuff()
except:
await transaction.rollback()
else:
await transaction.commit()
finally:
await connection.close()
# END TRANSACTION
return JSONResponse(status_code=201)
我使用Pytest编写了一个测试用例:
import pytest
app = FastAPI()
@pytest.fixture()
def client():
with TestClient(app=app) as c:
yield c
class TestUserCreation:
CREATE_USER_URL = "/users/create"
def test_create_user(self, client):
response = client.post(self.CREATE_USER_URL, json={"name": "John"})
assert response.status_code == 201
这个测试用例可以工作并将新创建的用户持久化到数据库中,但正如我之前所说,我希望在测试用例完成后自动回滚更改。
我已经查看了一些在线资源,但它们都没有帮助。
-
此链接讨论使用工厂对象,但我不能在这里使用工厂对象,因为我的控制器需要DB连接作为依赖项。此外,控制器本身正在更新DB,而不是一个“模拟”的工厂对象。
-
然后,我尝试手动注入依赖项。我希望在调用测试用例中的API之前手动创建一个连接,然后将其注入为所需的依赖项,然后在API完成后强制回滚事务。我找到了这个链接,其中讨论了如何在控制器之外获取依赖项的方法,但没有说明如何手动将其注入到控制器中。
-
FastAPI官方文档对如何回滚数据库中的持久化数据并不详尽。
我唯一想到的方法是不将DB连接作为依赖项注入到控制器中,而是将其附加到Starlette请求对象的请求中间件中。然后,在响应中间件中,根据环境变量(test
vs prod
)进行回滚,如果变量为test
,则始终回滚。
但对于一个健壮的测试套件来说,这似乎对我来说过于工程化。
在FastAPI中是否有任何现成可用的、内置的方法可以做到这一点?或者是否有其他库或包可用于为我完成这项工作?
如果Pytest不是最适合这个目的的框架,我很乐意更改为更合适的框架。
感谢您能提供的任何帮助。
谢谢!
英文:
I have a FastAPI application, with MySQL and asyncio
.
I have been trying to integrate some test cases with my application, with the ability to rollback the changes after every test case, so that all test cases can run in isolation.
This is how my controller is set up, with a DB dependency getting injected.
from sqlalchemy.ext.asyncio import create_async_engine
async def get_db_connection_dependency():
engine = create_async_engine("mysql+aiomysql://root:root@mysql8:3306/user_db")
connection = engine.connect()
return connection
class UserController:
async def create_user(
self,
request: Request,
connection: AsyncConnection = Depends(get_db_connection_dependency)
) -> JSONResponse:
# START TRANSACTION
await connection.__aenter__()
transaction = connection.begin()
await transaction.__aenter__()
try:
do_stuff()
except:
await transaction.rollback()
else:
await transaction.commit()
finally:
await connection.close()
# END TRANSACTION
return JSONResponse(status_code=201)
I have a test case written using Pytest like so
import pytest
app = FastAPI()
@pytest.fixture()
def client():
with TestClient(app=app) as c:
yield c
class TestUserCreation:
CREATE_USER_URL = "/users/create"
def test_create_user(self, client):
response = client.post(self.CREATE_USER_URL, json={"name": "John"})
assert response.status_code == 201
This test case works and persists the newly created user in the DB, but like I said earlier, I want to rollback the changes automatically once the test case finishes.
I have checked a few resources online, but none of them were helpful.
-
This link talks about using factory objects, but I can't use factory objects here because my controller requires the DB connection as a dependency.
Plus, the controller itself is updating the DB, and not a "mocked" factory object. -
I then searched for ways to inject the dependency manually. This was in the hopes that if I can create a connection manually BEFORE calling the API in my test case and inject it as the required dependency, then I can also forcefully rollback the transaction AFTER the API finishes.
- So, I came across this, which talks about a way to get a dependency to use outside of a controller, but not how to inject it into the controller manually.
-
The official FastAPI docs aren't very exhaustive on how to rollback persisted data in a DB-related test case.
The only way I can think of is to not inject the DB connection as a dependency into the controller, but attach it to the Starlette request object in the request middleware. And then in the response middleware, depending on an env var (test
vs prod
), I can ALWAYS rollback if the var is test
.
But this seems over-engineering to me for a very fundamental requirement of a robust testing suite.
Is there any readily-available, built-in way to do this in FastAPI? Or is there any other library or package available that can do it for me?
If Pytest isn't the best suited framework for this, I'm more than happy to change it to something more suitable.
Appreciate any help I can get.
Thank you!
答案1
得分: 0
我已解决了这个问题。
我将所有事务和连接处理都移到了依赖项中,控制器不再处理连接对象。这意味着我的实际API依赖可以根据其适当的情况调用commit
和rollback
函数,而我的测试用例中的覆盖依赖将始终调用rollback
,无论如何。
我的主要问题是我让控制器处理了一些数据库连接的事情,这让事情变得复杂化。
英文:
I have solved this.
I moved all transaction and connection handling into the dependency, with the controller doing nothing with the connection object. This meant that my actual API dependency can have the usual commit
and rollback
functions called for their appropriate scenarios, and my override dependency in my test case will always call rollback
no matter what.
My main problem was that I was letting my controller do some of the DB connection stuff which complicated things.
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论