MagicStack / MagicStack/asyncpg

error in pytest sqlalchemy.exc.InterfaceError

Ouverte
#1,220 1 commentaire 0 réactions 0 personnes assignées Voir sur GitHub

Personne n'a encore pris cette issue.

Langage dominant
Python
Étoiles
8.1k
Forks
468
Métriques de merge des PR
Aucune PR mergée en 30 j

Description

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"
```

Guide de contribution

Aucun guide de contribution indexé pour ce dépôt

Par où commencer

  1. Lisez l'issue en entier, puis le guide de contribution du projet.
  2. Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
  3. Forkez le dépôt et travaillez sur une branche.
  4. Ouvrez une pull request qui référence le numéro de l'issue.

Piste de recherche

Commencez par exécuter tests/test_users.py et comparez la configuration de la session async dans db.py avec les fixtures et la surcharge de dépendance dans conftest.py. Concentrez-vous sur les cas en échec test_get_all_users et test_get_user_by_id. C’est terminé lorsque les deux tests s’exécutent sans sqlalchemy.exc.InterfaceError.

Rédigé par le modèle d'indexation à partir du texte de l'issue.

Évaluation

Stack technique
postgresql, python, sqlalchemy
Domaine
databases
Type d'issue
Bug
Difficulté
3/5
Temps estimé
1-2 jours
Activité
À l'abandon
Clarté
À clarifier
Accessibilité débutants
25/100

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.