Test with sqlalchemy

test_sqla.py

import os

import pytest
from sqlalchemy import Column, Integer, create_engine, event, String
from sqlalchemy.orm import declarative_base, Session

Base = declarative_base()


class Model(Base):
    __tablename__ = "model"

    id = Column(Integer(), primary_key=True)
    name = Column(String(), unique=True)


@pytest.fixture(scope="session")
def engine():
    uri = os.getenv("DB_URI")
    engine = create_engine(uri)
    Base.metadata.create_all(bind=engine)
    yield engine
    Base.metadata.drop_all(bind=engine)


@pytest.fixture
def session(engine):
    connection = engine.connect()
    trans = connection.begin()

    session = Session(bind=connection)
    nested = connection.begin_nested()

    @event.listens_for(session, "after_transaction_end")
    def end_savepoint(session, transaction):
        nonlocal nested
        if not nested.is_active:
            nested = connection.begin_nested()

    yield session
    session.expunge_all()
    trans.rollback()
    connection.close()


model = Model(id=1, name="name111")


@pytest.mark.parametrize("x", [1, 2])
def test_x(x, session):
    print(session)
    print("test_x", x)
    session.add(model)
    session.commit()
    session.commit()
    assert False