How use pytest to unit test sqlalchemy orm classes

Question

I want to write some py.test code to test 2 simple sqlalchemy ORM classes that were created based on this Tutorial. The problem is, how do I set a the database in py.test to a test database and rollback all changes when the tests are done? Is it possible to mock the database and run tests without actually connect to de database?

here is the code for my classes:

from sqlalchemy import create_engine, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker, relationship

eng = create_engine('mssql+pymssql://user:pass@host/my_database')

Base = declarative_base(eng)
Session = sessionmaker(eng)
intern_session = Session()

class Author(Base):
    __tablename__ = "Authors"

    AuthorId = Column(Integer, primary_key=True)
    Name = Column(String)  
    Books = relationship("Book")

    def add_book(self, title):
        b = Book(Title=title, AuthorId=self.AuthorId)
        intern_session.add(b)
        intern_session.commit()

class Book(Base):
    __tablename__ = "Books"

    BookId = Column(Integer, primary_key=True)
    Title = Column(String)      
    AuthorId = Column(Integer, ForeignKey("Authors.AuthorId"))    

    Author = relationship("Author")                           

Answer 1

I usually do that this way:

1. I do not instantiate engine and session with the model declarations, instead I only declare a Base with no bind:

   Base = declarative_base()
   

   and I only create a session when needed with

   engine = create_engine('<the db url>')
   db_session = sessionmaker(bind=engine)
   

   You can do the same by not using the intern_session in your add_book method but rather use a session parameter.

   def add_book(self, session, title):
       b = Book(Title=title, AuthorId=self.AuthorId)
       session.add(b)
       session.commit()
   

   It makes your code more testable since you can now pass the session of your choice when you call the method. And you are no more stuck with a session bound to a hardcoded database url.

2. Add a custom --dburl option to pytest using its pytest_addoption hook.

   Simply add this to your top-level conftest.py:

   def pytest_addoption(parser):
       parser.addoption('--dburl',
                        action='store',
                        default='<if needed, whatever your want>',
                        help='url of the database to use for tests')
   

   Now you can run pytest --dburl <url of the test database>

3. Then you can retrieve the dburl option from the request fixture

   • From a custom fixture:

     @pytest.fixture()
     def db_url(request):
         return request.config.getoption("--dburl")
         # ...
     

   • Inside a test:

     def test_something(request):
         db_url = request.config.getoption("--dburl")
         # ...
     

---

At this point you are able to:

• get the test db_url in any test or fixture
• use it to create an engine
• create a session bound to the engine
• pass the session to a tested method

It is quite a mess to do this in every test, so you can make a usefull usage of pytest fixtures to ease the process.

Below are some fixtures I use:

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker


@pytest.fixture(scope='session')
def db_engine(request):
    """yields a SQLAlchemy engine which is suppressed after the test session"""
    db_url = request.config.getoption("--dburl")
    engine_ = create_engine(db_url, echo=True)

    yield engine_

    engine_.dispose()


@pytest.fixture(scope='session')
def db_session_factory(db_engine):
    """returns a SQLAlchemy scoped session factory"""
    return scoped_session(sessionmaker(bind=db_engine))


@pytest.fixture(scope='function')
def db_session(db_session_factory):
    """yields a SQLAlchemy connection which is rollbacked after the test"""
    session_ = db_session_factory()

    yield session_

    session_.rollback()
    session_.close()

Using the db_session fixture you can get a fresh and clean db_session for each single test. When the test ends, the db_session is rollbacked, keeping the database clean.