由于 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.asynciodb 固定装置已被删除,后者被 loop 固定装置取代。在某种程度上,这更有意义。此解决方案将自身附加到 FastAPI 数据库连接,而不是像我最初所做的那样启动自己的连接。

我第一次使用它时,我真的发誓。