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



@RqL

Использую asyncpg, вот код
сам тест:
import unittest
from models.create import create_city
from models.conn import pool_instance


class CreateTables(unittest.IsolatedAsyncioTestCase):
    async def test_create_city1(self):
        pool = await pool_instance.get_pool()
        await create_city("Moscow")
        async with pool.acquire() as conn:
            city = await conn.fetchrow("SELECT * FROM cities WHERE name="Moscow"")
            self.assertEqual(city["name"], "Moscow")
    
    async def test_create_city2(self):
        pool = await pool_instance.get_pool()
        await create_city("SPB")
        async with pool.acquire() as conn:
            city = await conn.fetchrow("SELECT * FROM cities WHERE name="SPB"")
            self.assertEqual(city["name"], "SPB")

models.create.create_city

async def create_city(name: str) -> None:
    pool = await pool_instance.get_pool()
    async with pool.acquire() as conn:
        await conn.execute('INSERT INTO cities(name) VALUES($1)', name)

models.conn

from config import DB_DSN
from asyncpg import create_pool

class ConnectionPool:
    _pool = None

    async def get_pool(self):
        if self._pool is None:
            self._pool = await create_pool(dsn=DB_DSN)
        return self._pool

pool_instance = ConnectionPool()

Если вынести эти функции отдельно и запустить через asyncio.gather, то ошибки не падают. Ошибка в написании самих тестов или в использовании пула? Как можно исправить тесты или может быть само взаимодействие с базой?


Решения вопроса 0


Ответы на вопрос 1



@va_k

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

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *