Reputation: 187
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!
Upvotes: 0
Views: 583
Reputation: 1611
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...
Upvotes: 1