I'm trying to use pytest to write functional tests for a Flask application which interfaces with a Neo4j graph database via the Neo4j driver.
The toy example using the Movie Database is outlined below using a route which deletes a record from the database. For testing purposes I would like to wrap the session in a transaction which would be rolled-back rather than committed.
The application has routes which runs Cypher statements within an auto-commit transaction via, session.run(...), however I can circumvent this logic during testing by enforcing the use of a transaction prior to request,
session.begin_transaction()
...
session.rollback_transaction()
My question is I'm unsure how to leverage this pattern using pytest. Do I have to somehow bind the database to the client? Also is there a fixture I can use which will ensure that every test leveraging the client will be wrapped in a transaction which can be rolled back?
myapp/app.py:
from flask import _app_ctx_stack, Flask, Response
from flask_restplus import Api, Resource
from neo4j.v1 import GraphDatabase
class FlaskGraphDatabase(object):
def __init__(self, app=None):
self.app = app
if app is not None:
self.init_app(app)
def init_app(self, app):
@app.teardown_appcontext
def teardown(exception):
ctx = _app_ctx_stack.top
if hasattr(ctx, 'neo4j_session'):
ctx.neo4j_session.close()
if hasattr(ctx, 'neo4j_driver'):
ctx.neo4j_driver.close()
@property
def driver(self):
ctx = _app_ctx_stack.top
if ctx is not None:
if not hasattr(ctx, 'neo4j_driver'):
ctx.neo4j_driver = GraphDatabase.driver('bolt://localhost:7687')
return ctx.neo4j_driver
@property
def session(self):
ctx = _app_ctx_stack.top
if ctx is not None:
if not hasattr(ctx, 'neo4j_session'):
ctx.neo4j_session = self.driver.session()
return ctx.neo4j_session
api = Api()
gdb = FlaskGraphDatabase()
@api.route('/<source>/acted_in/<target>')
class Friend(Resource):
def delete(self, source, target):
statement = """
MATCH (s:Person)-[r:ACTED_IN]->(t:Movie)
WHERE s.name = {source} AND t.title = {target}
DELETE r
"""
cursor = gdb.session.run(statement, source=source, target=target)
status = 204 if cursor.summary().counters.contains_updates else 404
return Response(status=status)
def create_app():
app = Flask(__name__)
gdb.init_app(app)
api.init_app(app)
return app
if __name__ == '__main__':
app = create_app()
app.run()
tests/conftest.py:
import pytest
from myapp.app import create_app
@pytest.yield_fixture(scope='session')
def app():
yield create_app()
@pytest.yield_fixture(scope='session')
def client(app):
with app.test_client() as client:
yield client
tests/test_routes.py:
def test_delete(client):
res = client.delete('/Keanu Reeves/acted_in/The Matrix')
assert res.status_code == 204