Pytest Проваливает Тесты, Даже Если Тест Действительно Работает и Выполняет Операции CRUD Должным Образом

#python #sqlalchemy #pytest #python-asyncio

Вопрос:

Использование 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 .

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

Ответ №1:

Я думаю, вы неправильно понимаете, как работают светильники 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() , похоже, требуется дополнительный аргумент, который вы, похоже, не передаете. Так что никаких гарантий…

Комментарии:

1. Спасибо за объяснение! Как отмечалось в моем вопросе, даже когда я удаляю оператор assert и просто запускаю drop_table функцию, я все равно получаю ту же самую ошибку. Но я просто попробовал ваше предложение и позвонил get_table_names так: @pytest.mark.asyncio def test_drop_table(): asyncio.run(delete_table('table_name')) tables = DB.get_collection_names() assert 'table_name' not in tables Но все равно получил ту же ошибку! ‘switch_occurred = Ложная попытка: gt; результат = контекст.переключатель(*args, **kwargs) E Ошибка типа: объект ‘NoneType’ не вызывается`

2. Извините, я немного поторопился с ответом. 'NoneType' object is not callable плохо пахнет, как вызов функции вместо передачи объекта функции в качестве параметра. Хотя не могу сказать без всего кода. Если вам нужна помощь, вы должны опубликовать минимальный пример, который мы можем скопировать, вставить и запустить, который воспроизводит ошибку или, по крайней мере, опубликует код для всех вызываемых функций в обратной трассировке.

3. Спасибо, обновил мой вопрос кодом! Я согласен с вами по поводу запаха ошибки 🙂 Это одна из причин, по которой я сбит с толку возвращаемой ошибкой, даже когда я удалил assert инструкцию из теста.