如何为测试设置Flask应用和SQLAlchemy？

Question

在Flask中，通常的做法是这样开始：

from flask import Flask
from flaskext.sqlalchemy import SQLAlchemy
app = Flask(__name__)
SQLALCHEMY_DATABASE_URI = 'something'
app.config.from_object(__name__)
db = SQLAlchemy(app)

然后在任何地方导入并使用app和db。但是，当你这样创建db时，它会从应用程序中获取配置，而一旦获取了这个配置，似乎就无法再更改了。Flask官网上有一些关于创建应用工厂的页面，但如果我这样做，仍然不清楚如何在各处使用app和db。

我该如何写一个脚本来测试我的Flask应用程序，使用不同的数据库？我应该如何构建我的应用程序以实现这一点？我需要使用module吗？

Answer 1

首先，不要直接在你的脚本里创建Flask应用，而是使用一个叫做应用工厂的方式。这意味着你需要写一个函数，这个函数接收你的配置文件作为参数，然后返回一个已经创建好的应用对象。接着，你可以创建一个全局的SQLAlchemy对象，不需要参数，然后在创建应用的时候对它进行配置，具体可以参考这里。

db = SQLAlchemy()

def create_app(configfile):
    app = Flask(__name__)

    app.config.from_pyfile(config, silent=True)
    db.init_app(app)

    # create routes, etc.

    return app

要运行这个应用，你只需要做类似下面的操作：

app = create_app('config.py')
app.run()

如果你想运行单元测试，可以做类似下面的事情：

class Test(TestCase):
    def setUp(self):
        # init test database, etc.
        app = create_app('test_config.py')
        self.app = app.test_client()
    def tearDown(self):
        # delete test database, etc.

在我的情况下，我是直接使用SQLAlchemy和scoped_session，而不是Flask-SQLAlchemy。我做的方式是一样的，只不过用了懒加载的SQLAlchemy设置。

Answer 2

你不想在导入时就连接数据库。可以在导入时配置你的应用，因为在测试之前你总是可以调整配置。下面的例子中，你的数据库连接会放在一些使用应用配置的函数后面，这样在单元测试时，你可以把数据库连接指向一个不同的文件，然后在设置中明确连接。

假设你有一个名为 myapp 的包，里面有一个 myapp.py 文件，内容如下：

# myapp/myapp.py
from __future__ import with_statement
from sqlite3 import dbapi2 as sqlite3
from contextlib import closing
from flask import Flask, request, session, g, redirect, url_for, abort, \
     render_template, flash

# configuration
DATABASE = '/tmp/flaskr.db'
DEBUG = True
SECRET_KEY = 'development key'
USERNAME = 'admin'
PASSWORD = 'default'

# create our little application :)
app = Flask(__name__)
app.config.from_object(__name__)
app.config.from_envvar('MYAPP_SETTINGS', silent=True)

def connect_db():
    """Returns a new connection to the database."""
    return sqlite3.connect(app.config['DATABASE'])


def init_db():
    """Creates the database tables."""
    with closing(connect_db()) as db:
        with app.open_resource('schema.sql') as f:
            db.cursor().executescript(f.read())
        db.commit()


@app.before_request
def before_request():
    """Make sure we are connected to the database each request."""
    g.db = connect_db()


@app.after_request
def after_request(response):
    """Closes the database again at the end of the request."""
    g.db.close()
    return response

@app.route('/')
def show_entries():
    cur = g.db.execute('select title, text from entries order by id desc')
    entries = [dict(title=row[0], text=row[1]) for row in cur.fetchall()]
    return render_template('show_entries.html', entries=entries)

if __name__=="__main__":
    app.run()

你的测试文件 myapp/test_myapp.py 会是这样的：

import os
import myapp
import unittest
import tempfile

class MyappTestCase(unittest.TestCase):

    def setUp(self):
        self.db_fd, myapp.app.config['DATABASE'] = tempfile.mkstemp()
        self.app = myapp.app.test_client()
        myapp.init_db()

    def tearDown(self):
        os.close(self.db_fd)
        os.unlink(myapp.app.config['DATABASE'])

    def test_empty_db(self):
        rv = self.app.get('/')
        assert 'No entries here so far' in rv.data

当然，如果你想使用 SQLAlchemy，你需要相应地更新 connect_db 和 init_db 函数，但希望你能明白这个思路。

Answer 3

你选择使用环境变量的想法是对的。不过，运行单元测试时用错数据库是有风险的。而且，你可能不想在每次请求和每个需要用到数据库的地方都去连接数据库。你可以使用一个配置文件夹和你明确设置的环境变量。这是我目前想到的最好方法。

run.py
shell.py

config/__init__.py
config/test.py
config/postgres.py
...

main/__init__.py
main/someapp/__init__.py
main/someapp/models.py

...
main/tests/__init__.py
main/tests/testutils.py

所以，配置文件可能是：

# config/test.py
SQLALCHEMY_DATABASE_URI = "sqlite://"

还有

# config/postgres.py
SQLALCHEMY_DATABASE_URI = 'postgresql://user:pw@localhost/somedb'

这样，我可以在我的基础测试类中明确设置数据库：

import os
from flask.ext.testing import TestCase

os.environ["DIAG_CONFIG_MODULE"] = "config.test"
from main import app, db


class SQLAlchemyTest(TestCase):

    def create_app(self):
        return app

    def setUp(self):
        db.create_all()

    def tearDown(self):
        db.session.remove()
        db.drop_all()

然后，对于我来说，main/__init__.py是：

import os

from flask import Flask, render_template, g
from flask.ext.sqlalchemy import SQLAlchemy

# by default, let's use a DB we don't care about
# but, we can override if we want
config_obj = os.environ.get("DIAG_CONFIG_MODULE", "config.test")
app = Flask(__name__)
app.config.from_object(config_obj)
db = SQLAlchemy(app)

@app.before_request
def before_request():
    g.db = db
    g.app = app

# ...
@app.route('/', methods=['GET'])
def get():
    return render_template('home.html')
# ...    
from main.someapp.api import mod as someappmod
app.register_blueprint(someappmod)

接着，在其他文件中，当我知道想要运行哪个配置时，可能会是：

# run.py
import os
os.environ["DIAG_CONFIG_MODULE"] = "config.postgres"
from main import app
app.run(debug=True)

还有

# shell.py
import os
os.environ["DIAG_CONFIG_MODULE"] = "config.postgres"

from main import app, db
from main.symdiag.models import *
from main.auth.models import *
print sorted(k for k in locals().keys() if not k.startswith("_"))
import IPython
IPython.embed()

也许……这是目前为止最好的方法 :P。