Menu
I'm trying to learn something about testing my flask app. In order to do that, I am using pytest and sqlalchemy.
I want to test a template, whose delivers route some SQL content. So in my opinion I need a testClient for testing the route itself and a DB fixture to manage the DB stuff included in the route.
Here is my fixture:
import pytest
from config import TestingConfig
from application import create_app, db
# ###########################
# ## functional tests
# ###########################
@pytest.fixture(scope='module')
def test_client():
app = create_app(TestingConfig)
# Flask provides a way to test your application by exposing the Werkzeug
# test Client and handling the context locals for you.
testing_client = app.test_client()
with app.app_context():
db.create_all()
yield testing_client # this is where the testing happens!
db.drop_all()
And this is my basic test:
def test_home_page(test_client):
"""
GIVEN a Flask application
WHEN the '/' page is requested (GET)
THEN check the response is valid and contains rendered content
"""
response = test_client.get('/')
assert response.status_code == 200
assert "SOME CONTENT" in response.data
Running my test fails with:
=================================================================================================== test session starts ===================================================================================================
platform linux -- Python 3.5.2, pytest-3.8.0, py-1.5.4, pluggy-0.7.1
rootdir: /home/dakkar/devzone/private/, inifile:
collected 2 items
tests/test_main.py
SETUP M test_client
tests/test_main.py::test_home_page (fixtures used: test_client)F
tests/test_main.py::test_valid_order_message (fixtures used: test_client).
TEARDOWN M test_client
======================================================================================================== FAILURES =========================================================================================================
_____________________________________________________________________________________________________ test_home_page ______________________________________________________________________________________________________
self = <sqlalchemy.engine.base.Connection object at 0x7f1c3f29b630>, dialect = <sqlalchemy.dialects.sqlite.pysqlite.SQLiteDialect_pysqlite object at 0x7f1c3f2c4ba8>
constructor = <bound method DefaultExecutionContext._init_compiled of <class 'sqlalchemy.dialects.sqlite.base.SQLiteExecutionContext'>>
statement = 'SELECT sum("order".col2_count) AS orders_col2, sum("order".col1_count) AS orders_col1, count("order".id) AS orders_count \nFROM "order"', parameters = ()
args = (<sqlalchemy.dialects.sqlite.base.SQLiteCompiler object at 0x7f1c3f29b6d8>, [immutabledict({})]), conn = <sqlalchemy.pool._ConnectionFairy object at 0x7f1c3f29b550>
context = <sqlalchemy.dialects.sqlite.base.SQLiteExecutionContext object at 0x7f1c3f29b6a0>
def _execute_context(self, dialect, constructor,
statement, parameters,
*args):
"""Create an :class:`.ExecutionContext` and execute, returning
a :class:`.ResultProxy`."""
try:
try:
conn = self.__connection
except AttributeError:
# escape "except AttributeError" before revalidating
# to prevent misleading stacktraces in Py3K
conn = None
if conn is None:
conn = self._revalidate_connection()
context = constructor(dialect, self, conn, *args)
except BaseException as e:
self._handle_dbapi_exception(
e,
util.text_type(statement), parameters,
None, None)
if context.compiled:
context.pre_exec()
cursor, statement, parameters = context.cursor, \
context.statement, \
context.parameters
if not context.executemany:
parameters = parameters[0]
if self._has_events or self.engine._has_events:
for fn in self.dispatch.before_cursor_execute:
statement, parameters = \
fn(self, cursor, statement, parameters,
context, context.executemany)
if self._echo:
self.engine.logger.info(statement)
self.engine.logger.info(
"%r",
sql_util._repr_params(parameters, batches=10)
)
evt_handled = False
try:
if context.executemany:
if self.dialect._has_events:
for fn in self.dialect.dispatch.do_executemany:
if fn(cursor, statement, parameters, context):
evt_handled = True
break
if not evt_handled:
self.dialect.do_executemany(
cursor,
statement,
parameters,
context)
elif not parameters and context.no_parameters:
if self.dialect._has_events:
for fn in self.dialect.dispatch.do_execute_no_params:
if fn(cursor, statement, context):
evt_handled = True
break
if not evt_handled:
self.dialect.do_execute_no_params(
cursor,
statement,
context)
else:
if self.dialect._has_events:
for fn in self.dialect.dispatch.do_execute:
if fn(cursor, statement, parameters, context):
evt_handled = True
break
if not evt_handled:
self.dialect.do_execute(
cursor,
statement,
parameters,
> context)
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:1193:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <sqlalchemy.dialects.sqlite.pysqlite.SQLiteDialect_pysqlite object at 0x7f1c3f2c4ba8>, cursor = <sqlite3.Cursor object at 0x7f1c3f2c2ce0>
statement = 'SELECT sum("order".col2_count) AS orders_col2, sum("order".col1_count) AS orders_col1, count("order".id) AS orders_count \nFROM "order"', parameters = ()
context = <sqlalchemy.dialects.sqlite.base.SQLiteExecutionContext object at 0x7f1c3f29b6a0>
def do_execute(self, cursor, statement, parameters, context=None):
> cursor.execute(statement, parameters)
E sqlite3.OperationalError: no such table: order
venv/lib/python3.5/site-packages/sqlalchemy/engine/default.py:509: OperationalError
The above exception was the direct cause of the following exception:
test_client = <FlaskClient <Flask 'application'>>
def test_home_page(test_client):
"""
GIVEN a Flask application
WHEN the '/' page is requested (GET)
THEN check the response is valid and contains rendered content
"""
> response = test_client.get('/')
tests/test_main.py:7:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
venv/lib/python3.5/site-packages/werkzeug/test.py:830: in get
return self.open(*args, **kw)
venv/lib/python3.5/site-packages/flask/testing.py:200: in open
follow_redirects=follow_redirects
venv/lib/python3.5/site-packages/werkzeug/test.py:803: in open
response = self.run_wsgi_app(environ, buffered=buffered)
venv/lib/python3.5/site-packages/werkzeug/test.py:716: in run_wsgi_app
rv = run_wsgi_app(self.application, environ, buffered=buffered)
venv/lib/python3.5/site-packages/werkzeug/test.py:923: in run_wsgi_app
app_rv = app(environ, start_response)
venv/lib/python3.5/site-packages/flask/app.py:2309: in __call__
return self.wsgi_app(environ, start_response)
venv/lib/python3.5/site-packages/flask/app.py:2295: in wsgi_app
response = self.handle_exception(e)
venv/lib/python3.5/site-packages/flask/app.py:1741: in handle_exception
reraise(exc_type, exc_value, tb)
venv/lib/python3.5/site-packages/flask/_compat.py:35: in reraise
raise value
venv/lib/python3.5/site-packages/flask/app.py:2292: in wsgi_app
response = self.full_dispatch_request()
venv/lib/python3.5/site-packages/flask/app.py:1815: in full_dispatch_request
rv = self.handle_user_exception(e)
venv/lib/python3.5/site-packages/flask/app.py:1718: in handle_user_exception
reraise(exc_type, exc_value, tb)
venv/lib/python3.5/site-packages/flask/_compat.py:35: in reraise
raise value
venv/lib/python3.5/site-packages/flask/app.py:1813: in full_dispatch_request
rv = self.dispatch_request()
venv/lib/python3.5/site-packages/flask/app.py:1799: in dispatch_request
return self.view_functions[rule.endpoint](**req.view_args)
application/main/routes.py:20: in index
func.count(Order.id).label("orders_count")
venv/lib/python3.5/site-packages/sqlalchemy/orm/query.py:2947: in one
ret = self.one_or_none()
venv/lib/python3.5/site-packages/sqlalchemy/orm/query.py:2917: in one_or_none
ret = list(self)
venv/lib/python3.5/site-packages/sqlalchemy/orm/query.py:2988: in __iter__
return self._execute_and_instances(context)
venv/lib/python3.5/site-packages/sqlalchemy/orm/query.py:3011: in _execute_and_instances
result = conn.execute(querycontext.statement, self._params)
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:948: in execute
return meth(self, multiparams, params)
venv/lib/python3.5/site-packages/sqlalchemy/sql/elements.py:269: in _execute_on_connection
return connection._execute_clauseelement(self, multiparams, params)
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:1060: in _execute_clauseelement
compiled_sql, distilled_params
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:1200: in _execute_context
context)
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:1413: in _handle_dbapi_exception
exc_info
venv/lib/python3.5/site-packages/sqlalchemy/util/compat.py:265: in raise_from_cause
reraise(type(exception), exception, tb=exc_tb, cause=cause)
venv/lib/python3.5/site-packages/sqlalchemy/util/compat.py:248: in reraise
raise value.with_traceback(tb)
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:1193: in _execute_context
context)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <sqlalchemy.dialects.sqlite.pysqlite.SQLiteDialect_pysqlite object at 0x7f1c3f2c4ba8>, cursor = <sqlite3.Cursor object at 0x7f1c3f2c2ce0>
statement = 'SELECT sum("order".col2_count) AS orders_col2, sum("order".col1_count) AS orders_col1, count("order".id) AS orders_count \nFROM "order"', parameters = ()
context = <sqlalchemy.dialects.sqlite.base.SQLiteExecutionContext object at 0x7f1c3f29b6a0>
def do_execute(self, cursor, statement, parameters, context=None):
> cursor.execute(statement, parameters)
E sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: order [SQL: 'SELECT sum("order".col2_count) AS orders_col2, sum("order".col1_count) AS orders_col1, count("order".id) AS orders_count \nFROM "order"'] (Background on this error at: http://sqlalche.me/e/e3q8)
venv/lib/python3.5/site-packages/sqlalchemy/engine/default.py:509: OperationalError
=========================================================================================== 1 failed, 1 passed in 0.52 seconds ============================================================================================
which tells me: db.create_all() does not create all tables in my testing database. Any hint, what I am doing wrong here?
Some additional info:
More Debugging: I followed this guide here: https://xvrdm.github.io/2017/07/03/testing-flask-sqlalchemy-database-with-pytest/
this is where thing become strange:
Link from above:
>>> db.engine.table_names() # Check the tables currently on the engine
[] # no table found
>>> db.create_all() # Create the tables according to defined models
>>> db.engine.table_names()
['users'] # Now table 'users' is found
What happenes in my project:
>>> db.engine.table_names()
[]
>>> db.create_all()
>>> db.engine.table_names()
[]
>>>
Snipplet from models.py:
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class Order(db.Model):
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), index=True, unique=True)
773
For each test that has test_dbin its argument list pytest first runs Base.metadata.create_all(bind=engine), then yields to the test code, and afterwards makes sure that Base.metadata.drop_all(bind=engine)gets run, even when the tests fail. The full code:
After setting up your basic test structure, pytest makes it really easy to write tests and provides a lot of flexibility for running the tests. pytest satisfies the key aspects of a good test environment: pytest is incredible! I highly recommend using it for testing any application or script written in Python.
This can be achieved with the following fixture: @pytest.fixture() def test_db(): Base.metadata.create_all(bind=engine) yield Base.metadata.drop_all(bind=engine) And then use it in your tests like so:
Even when checking code coverage, arguments can still be passed to pytest: This article provides a guide for testing Flask applications, focusing on: Patrick is a software engineer from the San Francisco Bay Area with experience in C++, Python, and JavaScript. His favorite areas of teaching are Vue and Flask.
You must be using flask-sqlalchemy, behind the scenes, it uses the declarative extension to define your models.
By subclassing a sqlalchemy declarative base class, sqlalchemy will generate Table and mapper for you, newly created table info stores in the corresponding Metadata obj. db.create_all() actually is metadata.create_all(), which will only create tables stored in the metadata.
Therefore, before you try to create a table with metadata.create_all, you have to store the info of that table into the metadata registry first, which equals to define a declarative base subclass. In python, this means to get your class definition code executed, which in turns, import the module the classes defined.
51
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With
