python - 带有数据库回滚的路由的 Flask 功能 pytest 测试

Question

我正在尝试使用 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

Answer 1

是的，您可以使用 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()