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

Solution

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.

Comments

[Feulo]

Nice! it worked, is there a way to use the pytest's fixtures to mock a database? So tests won't need to have a real connection with the database and I can test if all methods are working fine.

[Tryph]

@Feulo It is probably possible to use a fake database mock using pytest and the unittest.mock library, but this is something I never did.

[Feulo]

I'll take a look on the unnitest.mock documentation. Thanks!

[lorey]

Used this and it seems like it's is not able to handle session.commit calls during the tests as there is no nested transaction. This means tests might be commit to the database. I might be wrong as I have not fully grasped everything. But this worked for me: gist.github.com/kissgyorgy/e2365f25a213de44b9a2

[jon]

@Tryph What if you have an intermediate Interface that inherits from Base? In order to extend some functionality of the SQLAlchemy ORM. Many thanks!

[Tryph]

@jon I don't see why using a subclass of Base would prevent using this, but I never subclassed Base... In addition I cannot give a better answer with so few detail. Feel free to post a new question :)

[Ham]

@lorey Yes. By wrapping session in nested transaction, you can call session.commit() at anywhere of the functions you want to test , since the code sample (in the gist link) never calls transaction.commit() , the changes made by session.commit() never affects corresponding database tables, then rollback before the outer transaction ends.