如何将内存中的 SQLAlchemy sqlite3 数据库“注入”到 Flask test_client？

Question

我有一个与 SQLAlchemy 一起使用的 Flask 应用程序。我不想使用 Flask-SQLAlchemy 扩展。如果我使用“真实”数据库进行测试并设置指向数据库测试实例的环境变量，那么一切正常。

但是，如果我想指向内存中的 sqlite 数据库进行测试，我遇到了问题。在那种情况下，我可以在我的测试中设置数据，但是当我使用 test_client 在我的应用程序中执行给定路由时，它无法在“服务器端”找到我的数据库表 - 即在 @ 987654322@ 代码在下方显示。必须有一种方法来配置 test_client 以使其工作，但我似乎无法完全弄清楚如何去做。

以下是可能相关的代码的 sn-ps (db.py)：

import os

from sqlalchemy import create_engine

from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

engine = create_engine(os.environ['SQLALCHEMY_URL'])

Session = scoped_session(sessionmaker(bind=engine))

我在这里设置 scoped_session 以便我的数据库访问将是线程本地的。

引导我的应用程序的代码 (__init__.py)：

from flask import Flask

from .db import Session

from .hello import hello_blueprint

app = Flask(__name__)
app.register_blueprint(hello_blueprint)

@app.teardown_appcontext
def cleanup(resp_or_exc):
    Session.remove()

我在这里设置我的应用程序并在每次 Flask 弹出应用程序上下文时注册清理回调。

蓝图中的示例路由 (hello.py)：

import json

from flask import Blueprint

from .db import Session

from .models import Message

hello_blueprint = Blueprint('hello', __name__)

@hello_blueprint.route('/messages')
def messages():
    values = Session.query(Message).all()

    results = []
    for value in values:
        results.append({ 'message': value.message })

    return (json.dumps(results), 200, { 'content_type': 'application/json' })

这里我使用作用域会话从数据库中获取一些数据。

我的Message 模型的定义只是普通的 SQLAlchemy (models.py)：

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy import Column, Integer, String

Base = declarative_base()

class Message(Base):
    __tablename__ = 'messages'
    id = Column(Integer, primary_key=True)
    message = Column(String)

    def __repr__(self):
        return "<Message(message='%s')>" % (self.message)

下面是一个非常原始的 pytest 单元测试，只是为了在一个地方展示问题 (test_hello.py)：

import os 

import json

import pytest

import app

from .models import Message

@pytest.fixture
def client():
    client = app.app.test_client()

    return client

def test_hello(client):
    response = client.get('/')
    data = json.loads(response.data.decode('utf-8'))
    assert data == { 'message': "Hello friend!" }

def test_messages(client):
    with app.app.app_context():
        from sqlalchemy import create_engine
        from sqlalchemy import MetaData
        engine = create_engine('sqlite://')

        from .models import Base
        Base.metadata.create_all(engine)

        print('***metadata tables***')
        print(Base.metadata.tables.keys())

        from sqlalchemy.orm import scoped_session
        from sqlalchemy.orm import sessionmaker

        Session = scoped_session(sessionmaker(bind=engine))

        message = Message(message='Hello there!')

        Session.add(message)
        Session.commit()

        values = Session.query(Message).all()

        results = []
        for value in values:
            results.append({ 'message': value.message })

        # This works, prints : [{"message": "Hello there!"}]
        print('*** result***')    
        print(json.dumps(results))

        # The code below doesn't work. Flask's app.py throws an exception
        # with the following at its root:
        # sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: 
        # messages [SQL: 'SELECT messages.id AS messages_id, messages.message AS messages_message, messages.new_field 
        # AS messages_new_field \nFROM messages'] (Background on this error at: http://sqlalche.me/e/e3q8)

        response = client.get('/messages')

        data =json.loads(response.data.decode('utf-8'))
        assert data == [{'message': 'Hello there!'}]

Answer 1

我不太确定我之前做错了什么，但我设法通过了测试。

单元测试的修改版本如下（test_hello.py）：

import os 

import json

import pytest

import app

from .db import engine
from .db import Session

from .models import Base
from .models import Message

@pytest.fixture
def client():
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)

    client = app.app.test_client()

    return client

def test_messages(client):
    message = Message(message='Hello there!')

    Session.add(message)
    Session.commit()

    response = client.get('/messages')

    data = json.loads(response.data.decode('utf-8'))
    assert data == [{'message': 'Hello there!'}]