MagicStack / MagicStack/asyncpg

error in pytest sqlalchemy.exc.InterfaceError

Abierto
#1,220 1 comentario 0 reacciones 0 asignados Ver en GitHub

Nadie ha tomado este issue todavía.

Lenguaje dominante
Python
Estrellas
8.1k
Forks
468
Métricas de merge de PR
Sin PR fusionados en 30 d

Descripción

FAILED tests/test_users.py::test_get_all_users - sqlalchemy.exc.InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) : cannot perform operation: another ope...
FAILED tests/test_users.py::test_get_user_by_id - sqlalchemy.exc.InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) : cannot perform operation: another ope...

db.py
```
from typing import AsyncGenerator

from sqlalchemy import NullPool
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from config import settings

engine = create_async_engine(settings.db.url)
async_session_maker = async_sessionmaker(
bind=engine, class_=AsyncSession, expire_on_commit=False
)

async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
async with async_session_maker() as session:
yield session

test_engine = create_async_engine(settings.db.test_url, poolclass=NullPool)
test_async_session_maker = async_sessionmaker(
bind=test_engine, class_=AsyncSession, expire_on_commit=False
)

async def get_test_async_session() -> AsyncGenerator[AsyncSession, None]:
async with test_async_session_maker() as session:
yield session
```

conftest.py
```
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app import app
from core.database import get_async_session, get_test_async_session, test_engine
from core.database.models import Base

@pytest_asyncio.fixture(scope="function")
async def init_db():
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)

@pytest_asyncio.fixture(scope="function")
async def session(init_db) -> AsyncSession:
async for session in get_test_async_session():
try:
yield session
except IntegrityError:
await session.rollback()
raise
finally:
await session.rollback()
await session.close()

@pytest_asyncio.fixture(scope="function")
async def client(init_db) -> AsyncClient:
async def override_get_async_session():
async for session in get_test_async_session():
yield session

app.dependency_overrides[get_async_session] = override_get_async_session

transport = ASGITransport(app=app)
client = AsyncClient(transport=transport, base_url="http://testserver/api/v1")
yield client
await client.aclose()

async def user_authentication_headers(client: AsyncClient, username: str, password: str):
json = {"username": username, "password": password}
response = await client.post("/users/login", json=json)
data = response.json()
auth_token = data["access_token"]
headers = {"Authorization": f"Bearer {auth_token}"}
return headers
```

test_users.py
```
@pytest.mark.asyncio
async def test_get_all_users(client: AsyncClient, session: AsyncSession):
UserFactory._meta.sqlalchemy_session = session
superuser = UserFactory(is_superuser=True)
session.add(superuser)
await session.commit()

headers = await user_authentication_headers(client, superuser.username, "password123")
response = await client.get(
"/users",
headers=headers
)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)

@pytest.mark.asyncio
async def test_get_user_by_id(client: AsyncClient, session: AsyncSession):
UserFactory._meta.sqlalchemy_session = session
user = UserFactory(username="singleuser", is_active=True)
session.add(user)
await session.commit()

headers = await user_authentication_headers(client, user.username, "password123")
response = await client.get(
f"/users/{user.id}",
headers=headers
)
assert response.status_code == 200
data = response.json()
assert data["username"] == "singleuser"
```

Guía de contribución

No hay ninguna guía de contribución indexada para este repositorio

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Línea de trabajo

Empieza ejecutando tests/test_users.py y compara la configuración de la sesión async en db.py con los fixtures y la sobrescritura de dependencias en conftest.py. Céntrate en los casos fallidos test_get_all_users y test_get_user_by_id. Se considera terminado cuando ambas pruebas se completen sin sqlalchemy.exc.InterfaceError.

Escrito por el modelo de indexación a partir del texto del issue.

Evaluación

Stack tecnológico
postgresql, python, sqlalchemy
Área
databases
Tipo de issue
Error
Dificultad
3/5
Tiempo estimado
1-2 días
Estado de actividad
Estancado
Claridad
Necesita aclaración
Aptitud para principiantes
25/100

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.