Pytest Failing Tests Even Though The Test Is Actually Working and Performing CRUD Operations Properly

Question

Using Pytest 6.2.5 and SQLAlchemy 1.4 async engine with Postgres backend.

With Pytest, I am testing a function that drops a table from the database. The test fails even though the table actually does get dropped from the database! This is confirmed by querying the database directly with postgreSQL.

The logic of the test is: drop the table from the db, then check for the existence of the table by retrieving the list of tables in the db and asserting that the dropped table does not appear.

In conftest.py this fixture calls the function get_table_names() in the main script that returns a dict of all the table names in the db.

@pytest.fixture(scope='function')
def table_names():
    return DB.get_table_names()

Then in test.py we run the test:

@pytest.mark.asyncio
def test_drop_table(table_names):
    asyncio.run(DB.delete_table('table_name'))
    assert 'table_name' not in table_names

When running the test with a print statement that prints out dict table_names we can see that the dict does contain all the table names. Even though the table does get dropped from the db, we get this error:

table_names = dict_keys(['a_table', 'another_table', 'lots_of_tables']) # <- prints all tables accurately!

...

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/usr/lib/python3.7/asyncio/runners.py:43: in run
    return loop.run_until_complete(main)
/usr/lib/python3.7/asyncio/base_events.py:584: in run_until_complete
    return future.result()
../src/file.py:294: in delete_table
    await conn.run_sync(Base.metadata.drop_all(engine, [table], checkfirst=True))
../../../../env3/lib/python3.7/site-packages/sqlalchemy/ext/asyncio/engine.py:536: in run_sync
    return await greenlet_spawn(fn, conn, *arg, **kw)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

fn = None, _require_await = False
args = (<sqlalchemy.future.engine.Connection object at 0x7f3c9ffe4f60>,)
kwargs = {}
context = <_AsyncIoGreenlet object at 0x7f3c9f6bee08 (otid=0x7f3ca46717e0) dead>
switch_occurred = False

    async def greenlet_spawn(
        fn: Callable, *args, _require_await=False, **kwargs
    ) -> Any:
        """Runs a sync function ``fn`` in a new greenlet.
    
        The sync function can then use :func:`await_` to wait for async
        functions.
    
        :param fn: The sync callable to call.
        :param \\*args: Positional arguments to pass to the ``fn`` callable.
        :param \\*\\*kwargs: Keyword arguments to pass to the ``fn`` callable.
        """
    
        context = _AsyncIoGreenlet(fn, greenlet.getcurrent())
        # runs the function synchronously in gl greenlet. If the execution
        # is interrupted by await_, context is not dead and result is a
        # coroutine to wait. If the context is dead the function has
        # returned, and its result can be returned.
        switch_occurred = False
        try:
>           result = context.switch(*args, **kwargs)
E           TypeError: 'NoneType' object is not callable

../../../../env3/lib/python3.7/site-packages/sqlalchemy/util/_concurrency_py3k.py:123: TypeError

This seems to be the heart of the error TypeError: 'NoneType' object is not callable. I had thought this was referring to the list of table names that we're asserting against, but when I comment out this line and run the function purely to drop the database like this:

@pytest.mark.asyncio
def test_drop_table():
    asyncio.run(DB.delete_table('table_name'))

We still receive the same error even though function is working, the table gets dropped from the database!

EDIT: Thanks to suggestions below I have also tried calling get_table_names directly like this:

@pytest.mark.asyncio 
def test_drop_table():     
    asyncio.run(delete_table('table_name')) 
    tables = DB.get_table_names()     
    assert 'table_name' not in tables

But still received the same error!

switch_occurred = False try: > result = context.switch(*args, **kwargs) E TypeError: 'NoneType' object is not callable

All the other tests are passing properly.

Here are the functions in the main script that are being called:

Setup:

class Db:

    def __init__(self, config):
        self._engine = create_async_engine('postgresql+asyncpg://postgres@localhost:5432/db', echo=True, future=True)
        self._session = sessionmaker(self._engine, expire_on_commit=False, class_=AsyncSession)
        self._meta = MetaData()
        self._cache = self.cache()

    @property
    def meta(self):
        self._meta.reflect(bind=sync_engine)
        return self._meta

    @property
    def cache(self):
        Base = automap_base()
        Base.prepare(sync_engine, reflect=True)
        return Base

    def get_table_names(self):
        return self.meta.tables.keys()

    async def delete_table(self, table_name, sync_engine):
        table =  self.meta.tables[table_name]
        async with self._engine.begin() as conn:
            await conn.run_sync(Base.metadata.drop_all(sync_engine, [table], checkfirst=True))
        return

In the main script this runs on the regular database. For the tests, this class is pointed to the test_db.

I'm not sure where else to look to debug the cause of this failing test, would greatly appreciate being pointed in the right direction!

Answer 1

I think you are misunderstanding how pytest fixtures work. Fixtures are prerequesites for tests, that are initialized before the test runs. That means you just freeze the state of your table names before the test is executed and of course the table name of your dropped table will still be in there. You need to call get_table_names in your assert statement and it should work.

Also, you are using pytest.mark.asyncio wrong. It's supposed to be a decorator for async test functions, so you don't have to use asyncio.run(). The idomatic way to write your test would be

@pytest.mark.asyncio
async def test_drop_table():
    await Db.delete_table('table_name')
    assert 'table_name' not in get_table_names()

assuming get_table_names() querys your (test) database and is a sync function. Pytest will provide an event loop for your test and run it.

EDIT: There it is. You're calling Base.metadata.drop_all() instead of passing it. So your table gets dropped, but then None is passed to run_sync(), which results in your exception. You need to call run_sync like

await conn.run_sync(Base.metadata.drop_all, # function object
                               sync_engine, # *args
                                   [table],
                           checkfirst=True, # *kwargs
                    )

I'm a bit confused about your call signatures, since delete_table() seems to take an additional argument, that you don't seem to pass. So, no guarantees...