如何测试编辑数据库的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_会话与delete post视图中的db_会话不同

登录功能的代码为：

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_会话是delete_post视图中提到的db_会话

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_会话已替换为db.session

2. database.py文件仅用于启动空白烧瓶SQLAlchemy实例

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

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