MagicStack / MagicStack/asyncpg
error in pytest sqlalchemy.exc.InterfaceError
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
- Lee el issue completo y luego la guía de contribución del proyecto.
- Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
- Haz un fork del repositorio y trabaja en una rama.
- 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