Блог

Использование Pytest 6.2.5 и асинхронного движка SQLAlchemy 1.4 с серверной частью Postgres.

С помощью Pytest я тестирую функцию, которая удаляет таблицу из базы данных. Тест завершается неудачно, даже если таблица действительно удаляется из базы данных! Это подтверждается путем запроса базы данных непосредственно с помощью PostgreSQL.

Логика теста такова: удалите таблицу из базы данных, затем проверьте наличие таблицы, получив список таблиц в базе данных и подтвердив, что удаленная таблица не отображается.

В conftest.py этом приспособлении вызывается функция get_table_names() в основном скрипте, которая возвращает a dict всех имен таблиц в базе данных.

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

Затем test.py мы запускаем тест:

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

При запуске теста с print инструкцией, которая выводится, dict table_names мы видим, что dict она содержит все имена таблиц. Несмотря на то, что таблица действительно удаляется из базы данных, мы получаем эту ошибку:

 table_names = dict_keys(['a_table', 'another_table', 'lots_of_tables']) # lt;- 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 = (lt;sqlalchemy.future.engine.Connection object at 0x7f3c9ffe4f60gt;,) kwargs = {} context = lt;_AsyncIoGreenlet object at 0x7f3c9f6bee08 (otid=0x7f3ca46717e0) deadgt; switch_occurred = False   async def greenlet_spawn(  fn: Callable, *args, _require_await=False, **kwargs  ) -gt; 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: gt; 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   

Похоже, в этом и заключается суть ошибки TypeError: 'NoneType' object is not callable . Я думал, что это относится к списку имен таблиц, против которых мы утверждаем, но когда я комментирую эту строку и запускаю функцию исключительно для удаления базы данных, как это:

 @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: gt; 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  

В основном скрипте это выполняется в обычной базе данных. Для тестов это class указано на test_db .

Я не уверен, где еще искать причину этого неудачного теста, был бы очень признателен, если бы мне указали правильное направление!

Я думаю, вы неправильно понимаете, как работают светильники pytest. Приспособления-это предварительные условия для тестов, которые инициализируются перед запуском теста. Это означает, что вы просто замораживаете состояние имен ваших таблиц до выполнения теста, и, конечно, имя таблицы вашей удаленной таблицы все равно будет там. Вам нужно позвонить get_table_names в свое assert заявление, и оно должно сработать.

Кроме того, вы используете pytest.mark.asyncio неправильно. Он должен быть декоратором для асинхронных тестовых функций, поэтому вам не нужно его использовать asyncio.run() . Идоматичным способом написания вашего теста было бы

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

предполагая get_table_names() , что запрашивает вашу (тестовую) базу данных и является функцией синхронизации. Pytest предоставит цикл событий для вашего теста и запустит его.

ПРАВКА: Вот оно. Ты звонишь Base.metadata.drop_all() вместо того, чтобы передать его. Таким образом, ваша таблица удаляется , но затем None передается run_sync() , что приводит к вашему исключению. Тебе нужно позвонить run_sync , как

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

Я немного смущен вашими подписями вызовов, поскольку delete_table() , похоже, требуется дополнительный аргумент, который вы, похоже, не передаете. Так что никаких гарантий…