Pytest@parameterize在第一次测试后由于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.asynciodb夹具已被移除,后者被循环固定装置所取代。在某种程度上,这更有意义。这个解决方案将自己连接到FastAPI数据库连接,而不是像我最初所做的那样启动自己的数据库连接。

我第一次把它用起来时,我真的发誓了。

最新更新