0

在 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]

我正在考虑尝试手动启动事件循环，但不确定如果不关闭会产生什么后果。这有点令人困惑，真的。如果您对我的固定装置的外观有任何替代方案，那么我会全神贯注。

4

1 回答 1

1

似乎关于在 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固定装置以及被循环固定装置替换的固定装置。在某种程度上，这更有意义。该解决方案将自身附加到 FastAPI 数据库连接，而不是像我最初所做的那样启动您自己的。

我第一次让这个工作时，我真的发誓。

于 2021-04-01T15:35:51.313 回答