Использую 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, то ошибки не падают. Ошибка в написании самих тестов или в использовании пула? Как можно исправить тесты или может быть само взаимодействие с базой?
