Как можно избежать ошибки использования одного подключения при открытии нескольких соединений в одном пуле (в unittest)?

Улучшенная формулировка:

У меня есть код, использующий asyncpg. В нем есть набор тестов, в которых используется функция create_city для создания записей в базе данных. Я также использую ConnectionPool для управления пулом подключений к базе данных. Однако, когда я запускаю тесты с помощью asyncio.gather, ошибки не возникают. Я подозреваю, что ошибка может быть связана с написанием самих тестов или с использованием пула подключений. Можете ли вы помочь мне исправить тесты или проблему с взаимодействием с базой данных?
  • 21 ноября 2023 г. 13:13
Ответы на вопрос 2
Проблема может быть связана с тем, что вы используете одно подключение к базе данных во всех тестах, что может привести к неожиданной взаимозависимости между тестами и ошибкам. Чтобы избежать этой проблемы, вы можете предоставлять каждому тесту свое собственное подключение к базе данных. Вот как вы можете это сделать.

1. Создайте класс для тестирования, который унаследует `unittest.TestCase`.

```python
import asyncio
import asyncpg
import unittest

class MyTestCase(unittest.TestCase):
    async def asyncSetUp(self):
        self.pool = await asyncpg.create_pool(...)
        self.connection = await self.pool.acquire()
        self.transaction = self.connection.transaction()

        # Нужно для того, чтобы тесты могли использовать
        # асинхронные методы внутри себя
        await self.transaction.start()

    async def asyncTearDown(self):
        await self.transaction.rollback()
        await self.pool.release(self.connection)
```

2. Ваша функция `create_city` должна использовать подключение, предоставленное текущим тестом.

```python
async def create_city(self, name):
    # Используйте self.connection для выполнения запросов
    result = await self.connection.execute(...)
    return result
```

3. Ваши тесты должны использовать экземпляр `MyTestCase`, и вы должны переопределить метод `setUp` и `tearDown`, чтобы использовать асинхронные методы.

```python
class MyCityTest(MyTestCase):
    async def setUp(self):
        await super().setUp()
        # Создайте новую таблицу или очистите существующую перед каждым тестом
        await self.connection.execute("CREATE TABLE cities (name text)")

    async def tearDown(self):
        # Удалите таблицу или выполните другие операции после каждого теста
        await self.connection.execute("DROP TABLE cities")
        await super().tearDown()

    async def test_create_city(self):
        await self.create_city("Moscow")
        # Напишите дополнительные проверки здесь

    async def test_another_test(self):
        # Напишите другие тесты здесь

if __name__ == '__main__':
    unittest.main()
```

Теперь каждый тест будет использовать свое собственное подключение к базе данных и его транзакцию, что поможет избежать ошибок при запуске нескольких тестов одновременно.
Падает потому что тесты запускаются в разных лупах, а вы передаёте между ними объект с сылкой на луп из первого теста.
Похожие вопросы