如何解决由于 Tortoise ORM 中的闭合事件循环,Pytest @parametrize 在第一次测试后失败
在 FastAPI 中,运行使用 @pytest.mark.parametrize
的测试会通过,但仅针对第一组值。第二个和后续的没有。无论运行的测试数据如何,它们都具有相同的错误。
RuntimeError: Event loop is closed
如果 @pytest.mark.parametrize
有 3 种类型的数据要测试,那么上面的错误会出现 2 倍,因为只有第一个测试会起作用。我猜在第一次测试之后它认为一切都已完成并关闭事件循环。
我试过改变灯具的 scope
但只导致
ScopeMismatch: You tried to access the 'function' scoped fixture 'event_loop' with a 'module' scoped request object,involved factories
../../venv/myvenv/lib/python3.8/site-packages/pytest_asyncio/plugin.py:136: def wrapper(*args,**kwargs)
测试
from tortoise import Tortoise
DATABASE_URL = 'use your own' # I'm using postgres
DATABASE_MODELS = ['app.auth.models.rbac',]
# Fixture
@pytest.fixture
async def db():
await Tortoise.init(
db_url=DATABASE_URL,modules={'models': DATABASE_MODELS}
)
await Tortoise.generate_schemas()
# Test
param = [
('user.create',['AdminGroup','NoaddGroup']),('page.create',['DataGroup'])
]
@pytest.mark.parametrize('perm,out',param)
@pytest.mark.asyncio
async def test_permissions_get_groups(db,perm,out):
groups = await Permission.get_groups(perm)
assert Counter(groups) == Counter(out)
模型(简化)
class Group(models.Model):
name = fields.CharField(max_length=191,index=True,unique=True)
permissions: models.ManyToManyRelation['Permission'] = \
fields.ManyToManyField('models.Permission',related_name='groups',through='auth_group_permissions',backward_key='group_id')
class Meta:
table = 'auth_group'
class Permission(models.Model):
code = fields.CharField(max_length=191,unique=True)
class Meta:
table = 'auth_permission'
@classmethod
async def get_groups(cls,code):
groups = await Group.filter(permissions__code=code).values('name')
return [i.get('name') for i in groups]
我正在尝试手动启动事件循环,但不确定如果它没有关闭会产生什么后果。这有点令人困惑,真的。如果您对我的固定装置的外观有任何其他选择,那么我全神贯注。
解决方法
关于在 Tortoise ORM 中运行测试的文档似乎有点过时。在单元测试部分,它提到了有关使用 initializer()
和 finalizer()
的一些内容,但这些只会带来更多问题。看起来真正的解决方案比看起来更简单。
固定装置
from fastapi.testclient import TestClient
app = FastAPI()
# Fixtures
@pytest.fixture
def client():
with TestClient(app) as tc:
yield tc
@pytest.fixture
def loop(client):
yield client.task.get_loop()
和测试
param = [
('user.create',['AdminGroup','NoaddGroup']),('page.create',['DataGroup'])
]
@pytest.mark.parametrize('perm,out',param)
def test_sample(loop,perm,out):
async def ab():
groups = await Permission.get_groups(perm)
assert Counter(groups) == Counter(out)
loop.run_until_complete(ab())
请注意,@pytest.mark.asyncio
和 db 固定装置已被删除,后者被 loop 固定装置取代。在某种程度上,这更有意义。此解决方案将自身附加到 FastAPI 数据库连接,而不是像我最初所做的那样启动自己的连接。
我第一次使用它时,我真的发誓。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。