pythonpytestfastapitortoise-orm

Pytest @parametrize fails after first test due to closed event loop in Tortoise ORM


In FastAPI, running a test which uses @pytest.mark.parametrize goes through but only for the first set of values. The second and succeeding ones do not. Regardless of the test data being run they all have the same error.

RuntimeError: Event loop is closed

If @pytest.mark.parametrize has 3 types of data to test then the error above comes up 2x since only the first test would work. I'm guessing after the first test it thinks everything is all done and closes the event loop.

I'v tried changing the fixture's scope but only leads to

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)

The test

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)

The models (simplified)

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]

I'm looking at trying to start the event loop manually but am not sure what the consequences of that are if it isn't closed. It's a bit confusing, really. If you have any alternatives on how my fixture should look then I'm all ears.


Solution

  • It seems the documentation on running tests in Tortoise ORM is a bit off. In the unit test section it mentioned something about using initializer() and finalizer() but these brought only more problems. It seems the real solution is simpler than it looks.

    Fixtures

    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()
    

    And the test

    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())
    

    Take note that @pytest.mark.asyncio was removed as well as the db fixture which is replaced by the loop fixture. In a way this makes more sense. This solution attaches itself to the FastAPI database connection instead of starting your own which was what I initially did.

    First time I got this to work I literally swore.