我正在尝试使用 pytest
为通过 Neo4j driver 与 Neo4j 图形数据库接口(interface)的 Flask 应用程序编写功能测试.
使用 Movie Database 的玩具示例下面概述了使用从数据库中删除记录的路由。出于测试目的,我想将 session 包装在一个事务中,该事务将回滚而不是提交。
该应用程序具有通过 session.run(...)
在自动提交事务中运行 Cypher 语句的路由。 ,但是我可以在测试期间通过在请求之前强制使用事务来规避此逻辑,
session.begin_transaction()
...
session.rollback_transaction()
我的问题是我不确定如何使用
pytest
来利用这种模式.我是否必须以某种方式将数据库绑定(bind)到客户端?还有我可以使用的固定装置,它可以确保利用客户端的每个测试都包含在可以回滚的事务中?我的应用程序/应用程序.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()
测试/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
测试/test_routes.py:
def test_delete(client):
res = client.delete('/Keanu Reeves/acted_in/The Matrix')
assert res.status_code == 204
最佳答案
是的,您可以使用 fixture 来实现这一点:在您的 conftest.py 中添加一个具有 session 范围的自动使用 fixture ,它将在测试 session 开始时启动一个事务,并在结束时将其回滚。
测试/conftest.py:
from neomodel import db
@pytest.fixture(autouse=True, scope="session")
def setup_neo_test_db():
print("Initializing db transaction")
db.begin()
yield
print("Rolling back")
db.rollback()
关于python - 带有数据库回滚的路由的 Flask 功能 pytest 测试,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/45257116/