In real project development, testing is always a must-have feature and
when dealing code with complex business logic, we would be better to
test against a real database. The following snippet will illustrate a way to
set up real database testing with py.test
for a Flask project.
Before we begin, let me give you a simple introduction of py.test
.
py.test
is a python test framework which helps you write test cases in
a more elegant way. Unlike python’s builtin unittest
package
which organizes tests (and test suites) by class, py.test
allows writing test
with a simple function and without importing some external packages:
def test_the_ultimate_question():
assert inquiry_the_answer_for_the_ultimate_question() == 42
Then, we type py.test
in the console and everything will work by default.
Besides the benefit of writing less boilerplate code, py.test
also comes with
a handful of useful plugins, for instance:
pytest-cov
: a plugin for generating code coverage reportpytest-bdd
: a plugin for writing BDD style testsWhen talking about writing test, most of the work is to organize the data for the component we are going to test. For example, to test an account debit service, we may setup an account, like an account with $5 balance, for the debit test case; after the debit function ran, we should check if the account is in the expected state with proper data. There are numerous ways to deal with the test data, one way is to mock these data while the other is to use real data (i.e. setup a real database account record for the test). We won’t go any further on this topic because it will make a long post to explain them thoroughly.
Suppose we are building a simple api service, which responses a quote on each HTTP call, and the quotes are read from the database (if there are no available quotes, a default quote will be returned). Let’s go straight to the code:
db = SQLAlchemy()
class Quote(db.Model):
"""The quote database model."""
id = db.Column(db.Integer, primary_key=True)
message = db.Column(db.Text)
api_bp = Blueprint('api', __name__)
@api_bp.route('/hello', methods=('GET', ))
def api_hello():
# read from database with random order
quote = db.session.query(Quote).order_by(sql_func.rand()).first()
if not quote:
# default case
return jsonify(message='hello, world')
return jsonify(message=quote.message)
And this is the flask application setup flow:
def create_app(**configs) -> Flask:
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv(
'SQLALCHEMY_DATABASE_URI'
)
app.config.update(**configs)
db.init_app(app)
app.register_blueprint(api_bp, url_prefix='/api')
return app
Next step is to write some tests for this api. Before we start, we have to define the test database & test flask application:
import pathlib
import alembic
from alembic.config import Config as AlembicConfig
import pytest
from quote_app import create_app
# the application scoped database instance
from quote_app import db as _db
_project_root = pathlib.Path(__file__).parent.parent
@pytest.fixture(scope='session')
def app(request):
"""Create flask application (session scope)."""
configs = {
'SERVER_NANE': 'test',
'TESTING': True,
}
app = create_app(**configs)
# prepare the application context
ctx = app.app_context()
ctx.push()
def teardown():
ctx.pop()
request.addfinalizer(teardown)
return app
@pytest.fixture(scope='session')
def create_db_with_alembic(app, request):
# bind the app the database instance
_db.app = app
alembic_config = AlembicConfig(_project_root / 'alembic.ini')
# override settings & run migrations
alembic_config.set_section_option(
'alembic', 'sqlalchemy.url',
app.config['SQLALCHEMY_DATABASE_URI']
)
alembic.command.upgrade(alembic_config, 'head')
def teardown():
_db.session.remove()
_db.drop_all()
# drop alembic's revisions table, so next time we can re-run these
# migrations
_db.session.execute('drop table if exists alembic_version')
request.addfinalizer(teardown)
return _db
@pytest.fixture(scope='session')
def create_db_with_sqlalchemy(app, request):
# bind the app the database instance
_db.app = app
_db.create_all()
def teardown():
_db.session.remove()
_db.drop_all()
request.addfinalizer(teardown)
return _db
@pytest.fixture
def db(
# change this parameter to use the other implementation
create_db_with_alembic,
# create_db_with_sqlalchemy,
request,
):
db = create_db_with_alembic
# db = create_db_with_sqlalchemy
def cleanup_tables():
# sort tables with topology order, ensure children tables are
# deleted first
for table in db.metadata.sorted_tables[::-1]:
db.session.execute(table.delete())
db.session.commit()
cleanup_tables()
request.addfinalizer(cleanup_tables)
return db
@pytest.fixture
def api_client(db, app):
"""Create api request client."""
return app.test_client()
Now turn to the real testing logic:
from quote_app import Quote
def test_api_hello__no_quote(api_client):
resp = api_client.get('/api/hello')
assert resp.status_code == 200
assert resp.json['message'] == 'hello, world'
def test_api_hello__single_quote(db, api_client):
quote = Quote()
quote.mesage = 'test-mesage'
db.session.add(quote)
db.session.commit()
resp = api_client.get('/api/hello')
assert resp.status_code == 200
assert resp.json['message'] == quote.message
Finally, run the tests:
$ py.test
platform linux -- Python 3.7.4, pytest-5.0.1, py-1.8.0, pluggy-0.12.0
rootdir: /b4fun/snippet/pytest_real_database
collected 2 items
test/test_api_hello.py .. [100%]
========================= 2 passed in 1.14 seconds ============================
With the help of py.test, we can create a basic test environment for a flask-sqlalchemy based web application project easily.
A full example of code can be check out from the GitHub repo.