Skip to content

Instantly share code, notes, and snippets.

@jose-lpa
Last active September 15, 2020 16:28
Show Gist options
  • Save jose-lpa/1b456890e609452833d10b5c7dcbc861 to your computer and use it in GitHub Desktop.
Save jose-lpa/1b456890e609452833d10b5c7dcbc861 to your computer and use it in GitHub Desktop.
Disposable database with Pytest, with data cleanup per-test for proper isolation
import pytest
from sqlalchemy import create_engine, engine, MetaData
from sqlalchemy.orm import sessionmaker
from sqlalchemy_utils import create_database, database_exists, drop_database
from my_project.database import Base
@pytest.fixture(scope="session", autouse=True)
def disposable_database() -> engine.Connection:
db_engine = create_engine(SQLALCHEMY_DATABASE_URI, pool_pre_ping=True)
if database_exists(db_engine.url):
drop_database(db_engine.url)
create_database(db_engine.url)
with db_engine.connect() as connection:
Base.metadata.create_all(bind=connection)
yield connection
Base.metadata.drop_all(bind=connection)
# Get rid of database entirely after all tests have finished.
drop_database(db_engine.url)
@pytest.fixture(scope="function")
def test_db_session(disposable_database: engine.Connection):
Session = sessionmaker(autocommit=False, bind=disposable_database)
session = Session()
try:
yield session
except Exception: # pylint: disable=broad-except
# Rollback all operations in case unit test went wrong.
session.rollback()
finally:
# In any case, cleanup the test database after every test to remove any
# created data.
meta = MetaData(bind=disposable_database, reflect=True)
for table in meta.sorted_tables:
session.execute(table.delete())
session.commit()
session.close()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment