由于 Tortoise ORM 中的事件循环关闭,Pytest @parametrize 在第一次测试后失败
Pytest @parametrize fails after first test due to closed event loop in Tortoise ORM
在 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, index=True, 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 数据库连接,而不是启动您自己的数据库连接,这是我最初所做的。
我第一次让这个工作时,我真的发誓了。
在 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, index=True, 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 数据库连接,而不是启动您自己的数据库连接,这是我最初所做的。
我第一次让这个工作时,我真的发誓了。