如何测试编辑数据库的 Flask 视图?

Question

我正在尝试编写测试以涵盖我正在构建的网站的大部分功能，但是在运行测试时不断收到以下错误。

Traceback (most recent call last):
File "tests.py", line 291, in test_delete_post_page_li
response = c.get('/delete_post/1', follow_redirects=True)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/werkzeug/test.py", line 1006, in get
return self.open(*args, **kw)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/testing.py", line 227, in open
follow_redirects=follow_redirects,
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/werkzeug/test.py", line 970, in open
response = self.run_wsgi_app(environ.copy(), buffered=buffered)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/werkzeug/test.py", line 861, in run_wsgi_app
rv = run_wsgi_app(self.application, environ, buffered=buffered)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/werkzeug/test.py", line 1096, in run_wsgi_app
app_rv = app(environ, start_response)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/app.py", line 2464, in __call__
return self.wsgi_app(environ, start_response)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/app.py", line 2450, in wsgi_app
response = self.handle_exception(e)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/app.py", line 1867, in handle_exception
reraise(exc_type, exc_value, tb)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/_compat.py", line 39, in reraise
raise value
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/app.py", line 2447, in wsgi_app
response = self.full_dispatch_request()
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/app.py", line 1952, in full_dispatch_request
rv = self.handle_user_exception(e)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/app.py", line 1821, in handle_user_exception
reraise(exc_type, exc_value, tb)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/_compat.py", line 39, in reraise
raise value
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/app.py", line 1950, in full_dispatch_request
rv = self.dispatch_request()
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/flask/app.py", line 1936, in dispatch_request
return self.view_functions[rule.endpoint](**req.view_args)
File "/home/kody/Projects/lifeLongLearning/app/blogs/views.py", line 182, in delete_post
db_session.delete(post)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/sqlalchemy/orm/scoping.py", line 162, in do
return getattr(self.registry(), name)(*args, **kwargs)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2018, in delete
self._delete_impl(state, instance, head=True)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2030, in _delete_impl
to_attach = self._before_attach(state, obj)
File "/home/kody/Projects/lifeLongLearning/venv/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2417, in _before_attach
% (state_str(state), state.session_id, self.hash_key)
sqlalchemy.exc.InvalidRequestError: Object '<Post at 0x7fa75bb2ec50>' is already attached to session '19' (this is '4')

测试代码为：

class LoggedDatabaseTests(TestCase):
############################
#### setup and teardown ####
############################
def create_app(self):
app.config.from_object('config.TestConfiguration')
return app
# executed prior to each test
def setUp(self):
self.engine = create_engine(app.config['SQLALCHEMY_DATABASE_URI'])
self.db_session = scoped_session(sessionmaker(autocommit=False,
autoflush=False,
bind=self.engine))
Base.query = self.db_session.query_property()
Base.metadata.create_all(bind=self.engine)
# executed after each test
def tearDown(self):
self.db_session.close()
self.db_session.remove()
self.db_session.rollback()
Base.metadata.drop_all(self.engine)

def test_delete_post_page_li(self):
p_cat = PostCategory(name='froots')
self.db_session.add(p_cat)
self.db_session.commit()
post = Post(name='Hello', content='3fhskajlga', category_id=1, category=p_cat)
self.db_session.add(post)
self.db_session.commit()
with app.test_client() as c :
login(c, '*****', '*****')
response = c.get('/delete_post/1', follow_redirects=True)
self.assertEqual(response.status_code, 302)
assert post not in self.db_session

测试代码中提到的db_session与删除帖子视图中db_session不同。

登录函数的代码为：

def login(client, username, password):
return client.post('/login', data=dict(
username=username,
password=password
), follow_redirects=True)

登录视图为：

@auth.route('/login', methods=['GET', 'POST'])
def login():
form = LoginForm()
if form.validate_on_submit():
if check_password_hash(passwrd, form.password.data) and form.username.data == 'LLLRocks':
session['logged_in'] = True
return redirect(url_for('other.home'))
# load login template
return render_template('login.html', form=form, title='Login')

删除视图为：

#
# Delete Post
# Description:
#   This is a view that will delete a post. The id that is passed in is that of the
#   post that will be deleted.
#
@blogs.route('/delete_post/<int:id>', methods=['GET', 'POST'])
def delete_post(id):
"""
Delete a post from the database
"""
# check if user is logged in
if not session.get('logged_in'):
return redirect(url_for('other.home'))
post = Post.query.get(id)
db_session.delete(post)
db_session.commit()
db_session.close()
db_session.remove()
db_session.rollback()
# redirect to the home page
return redirect(url_for('other.home'))

database.py 文件如下。此文件中的db_session是delete_post视图中提到的db_session。

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
# Need to connect to the new database
engine = create_engine('mysql+mysqldb://****:******@******/****', convert_unicode=True, pool_recycle=3600, pool_pre_ping=True)
db_session = scoped_session(sessionmaker(autocommit=False,
autoflush=False,
bind=engine))
Base = declarative_base()
Base.query = db_session.query_property()
def init_db():
# import all modules here that might define models so that
# they will be registered properly on the metadata.  Otherwise
# you will have to import them first before calling init_db()
import app.models
Base.metadata.create_all(bind=engine)

有一天，我将深入研究文档，但在此之前，我很抱歉我的无知。如果我错过了发布任何重要代码，请告诉我，我会立即发布。

Answer 1

我已经能够让代码工作，下面我将发布我如何让它工作。

测试代码现在如下所示。

from app.database import db
from config import TestConfiguration
from app import create_app as c_app
class TestingWhileLoggedIn(TestCase):
def create_app(self):
app = c_app(TestConfiguration)
return app
# executed prior to each test
def setUp(self):
self.app_context = self.app.app_context()
self.app_context.push()
db.create_all()
login(self.client, 'LLLRocks', 'h0ngk0ng')
# executed after each test
def tearDown(self):
db.session.remove()
db.drop_all()
self.app_context.pop()
logout(self.client)
def test_delete_post_page_li(self):
p_cat = PostCategory(name='froots')
db.session.add(p_cat)
db.session.commit()
post = Post(name='Hello', content='3fhskajlga', category_id=1, category=p_cat)
db.session.add(post)
db.session.commit()
response = self.client.get('/delete_post/1', follow_redirects=False)
self.assertEqual(response.status_code, 302)
deleted_post = Post.query.filter_by(name='Hello').first()
self.assertEqual(deleted_post, None)
assert post not in db.session

配置文件现在看起来像

import os
from os.path import abspath, dirname, join
# _cwd = dirname(abspath(__file__))
_basedir = os.path.abspath(os.path.dirname(__file__))

TOP_LEVEL_DIR = os.path.abspath(os.curdir)
class Config(object) :
pass
class BaseConfiguration(object):
SQLALCHEMY_TRACK_MODIFICATIONS = False

class ProductionConfiguration(BaseConfiguration):
SQLALCHEMY_DATABASE_URI = '***************'
SQLALCHEMY_POOL_PRE_PING = True
SQLALCHEMY_ENGINE_OPTIONS = {'pool_recycle' : 3600}
SECRET_KEY = '************'
UPLOAD_FOLDER = TOP_LEVEL_DIR + '/app/static'

class TestConfiguration(BaseConfiguration):
TESTING = True
WTF_CSRF_ENABLED = False
SECRET_KEY = '************'
SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(_basedir, 'testing.sqlite')

database.py 文件

如下

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

并且正在导入的create_app函数是

def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)
db.init_app(app)@blogs.route('/delete_post/<int:id>', methods=['GET', 'POST'])
def delete_post(id):
"""
Delete a post from the database
"""
# check if user is logged in
if not session.get('logged_in'):
return redirect(url_for('other.home'))
post = Post.query.get(id)
db.session.delete(post)
db.session.commit()
# redirect to the home page
return redirect(url_for('other.home'))
from app import models
from .blogs import blogs as blogs_blueprint
app.register_blueprint(blogs_blueprint)
from .auth import auth as auth_blueprint
app.register_blueprint(auth_blueprint)
from .other import other as other_blueprint
app.register_blueprint(other_blueprint)
from .worksheets import worksheets as worksheets_blueprint
app.register_blueprint(worksheets_blueprint)


@app.teardown_appcontext
def shutdown_session(exception=None):
db.session.close()
db.session.remove()
db.session.rollback()

return app

正在测试的视图如下所示

@blogs.route('/delete_post/<int:id>', methods=['GET', 'POST'])
def delete_post(id):
"""
Delete a post from the database
"""
# check if user is logged in
if not session.get('logged_in'):
return redirect(url_for('other.home'))
post = Post.query.get(id)
db.session.delete(post)
db.session.commit()
# redirect to the home page
return redirect(url_for('other.home'))

对问题中的代码所做的更改如下所示。

1. db_session被替换为 db.session

2. database.py 文件仅启动一个空白的 Flask-SQLAlchemy 实例

3. 测试的设置是使用 init 中定义的create_app函数进行的。测试配置已传递给它，因此它使用测试数据库。然后，使用 db.drop_all 和db.create_all Flask-SQLAlchemy 函数来创建和清理数据库。
4. 登录和注销已添加到测试的设置和拆卸中，但这与原始问题无关。

解决问题的是将代码更改为使用Flask-SQLAlchemy，而不仅仅是SQLAlchemy。