python - Pytest:设置测试客户端和数据库

Question

我想学习一些关于测试我的 Flask 应用程序的知识。为此，我使用了 pytest 和 sqlalchemy。

我想测试一个模板，它传递路由一些 SQL 内容。所以在我看来，我需要一个 testClient 来测试路由本身，以及一个 DB fixture 来管理路由中包含的 DB 内容。

这是我的固定装置:

import pytest
from config import TestingConfig
from application import create_app, db


# ###########################
# ## functional tests
# ###########################


@pytest.fixture(scope='module')
def test_client():
    app = create_app(TestingConfig)

    # Flask provides a way to test your application by exposing the Werkzeug 
    # test Client and handling the context locals for you.
    testing_client = app.test_client()

    with app.app_context():

        db.create_all()

        yield testing_client  # this is where the testing happens!

        db.drop_all()

这是我的基本测试:

def test_home_page(test_client):
    """
    GIVEN a Flask application
    WHEN the '/' page is requested (GET)
    THEN check the response is valid and contains rendered content
    """
    response = test_client.get('/')
    assert response.status_code == 200
    assert "SOME CONTENT" in response.data

运行我的测试失败:

=================================================================================================== test session starts ===================================================================================================
platform linux -- Python 3.5.2, pytest-3.8.0, py-1.5.4, pluggy-0.7.1
rootdir: /home/dakkar/devzone/private/, inifile:
collected 2 items                                                                                                                                                                                                         

tests/test_main.py 
    SETUP    M test_client
        tests/test_main.py::test_home_page (fixtures used: test_client)F
        tests/test_main.py::test_valid_order_message (fixtures used: test_client).
    TEARDOWN M test_client

======================================================================================================== FAILURES =========================================================================================================
_____________________________________________________________________________________________________ test_home_page ______________________________________________________________________________________________________

self = <sqlalchemy.engine.base.Connection object at 0x7f1c3f29b630>, dialect = <sqlalchemy.dialects.sqlite.pysqlite.SQLiteDialect_pysqlite object at 0x7f1c3f2c4ba8>
constructor = <bound method DefaultExecutionContext._init_compiled of <class 'sqlalchemy.dialects.sqlite.base.SQLiteExecutionContext'>>
statement = 'SELECT sum("order".col2_count) AS orders_col2, sum("order".col1_count) AS orders_col1, count("order".id) AS orders_count \nFROM "order"', parameters = ()
args = (<sqlalchemy.dialects.sqlite.base.SQLiteCompiler object at 0x7f1c3f29b6d8>, [immutabledict({})]), conn = <sqlalchemy.pool._ConnectionFairy object at 0x7f1c3f29b550>
context = <sqlalchemy.dialects.sqlite.base.SQLiteExecutionContext object at 0x7f1c3f29b6a0>

    def _execute_context(self, dialect, constructor,
                         statement, parameters,
                         *args):
        """Create an :class:`.ExecutionContext` and execute, returning
            a :class:`.ResultProxy`."""

        try:
            try:
                conn = self.__connection
            except AttributeError:
                # escape "except AttributeError" before revalidating
                # to prevent misleading stacktraces in Py3K
                conn = None
            if conn is None:
                conn = self._revalidate_connection()

            context = constructor(dialect, self, conn, *args)
        except BaseException as e:
            self._handle_dbapi_exception(
                e,
                util.text_type(statement), parameters,
                None, None)

        if context.compiled:
            context.pre_exec()

        cursor, statement, parameters = context.cursor, \
            context.statement, \
            context.parameters

        if not context.executemany:
            parameters = parameters[0]

        if self._has_events or self.engine._has_events:
            for fn in self.dispatch.before_cursor_execute:
                statement, parameters = \
                    fn(self, cursor, statement, parameters,
                       context, context.executemany)

        if self._echo:
            self.engine.logger.info(statement)
            self.engine.logger.info(
                "%r",
                sql_util._repr_params(parameters, batches=10)
            )

        evt_handled = False
        try:
            if context.executemany:
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_executemany:
                        if fn(cursor, statement, parameters, context):
                            evt_handled = True
                            break
                if not evt_handled:
                    self.dialect.do_executemany(
                        cursor,
                        statement,
                        parameters,
                        context)
            elif not parameters and context.no_parameters:
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_execute_no_params:
                        if fn(cursor, statement, context):
                            evt_handled = True
                            break
                if not evt_handled:
                    self.dialect.do_execute_no_params(
                        cursor,
                        statement,
                        context)
            else:
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_execute:
                        if fn(cursor, statement, parameters, context):
                            evt_handled = True
                            break
                if not evt_handled:
                    self.dialect.do_execute(
                        cursor,
                        statement,
                        parameters,
>                       context)

venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:1193: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <sqlalchemy.dialects.sqlite.pysqlite.SQLiteDialect_pysqlite object at 0x7f1c3f2c4ba8>, cursor = <sqlite3.Cursor object at 0x7f1c3f2c2ce0>
statement = 'SELECT sum("order".col2_count) AS orders_col2, sum("order".col1_count) AS orders_col1, count("order".id) AS orders_count \nFROM "order"', parameters = ()
context = <sqlalchemy.dialects.sqlite.base.SQLiteExecutionContext object at 0x7f1c3f29b6a0>

    def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       sqlite3.OperationalError: no such table: order

venv/lib/python3.5/site-packages/sqlalchemy/engine/default.py:509: OperationalError

The above exception was the direct cause of the following exception:

test_client = <FlaskClient <Flask 'application'>>

    def test_home_page(test_client):
        """
        GIVEN a Flask application
        WHEN the '/' page is requested (GET)
        THEN check the response is valid and contains rendered content
        """
>       response = test_client.get('/')

tests/test_main.py:7: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
venv/lib/python3.5/site-packages/werkzeug/test.py:830: in get
    return self.open(*args, **kw)
venv/lib/python3.5/site-packages/flask/testing.py:200: in open
    follow_redirects=follow_redirects
venv/lib/python3.5/site-packages/werkzeug/test.py:803: in open
    response = self.run_wsgi_app(environ, buffered=buffered)
venv/lib/python3.5/site-packages/werkzeug/test.py:716: in run_wsgi_app
    rv = run_wsgi_app(self.application, environ, buffered=buffered)
venv/lib/python3.5/site-packages/werkzeug/test.py:923: in run_wsgi_app
    app_rv = app(environ, start_response)
venv/lib/python3.5/site-packages/flask/app.py:2309: in __call__
    return self.wsgi_app(environ, start_response)
venv/lib/python3.5/site-packages/flask/app.py:2295: in wsgi_app
    response = self.handle_exception(e)
venv/lib/python3.5/site-packages/flask/app.py:1741: in handle_exception
    reraise(exc_type, exc_value, tb)
venv/lib/python3.5/site-packages/flask/_compat.py:35: in reraise
    raise value
venv/lib/python3.5/site-packages/flask/app.py:2292: in wsgi_app
    response = self.full_dispatch_request()
venv/lib/python3.5/site-packages/flask/app.py:1815: in full_dispatch_request
    rv = self.handle_user_exception(e)
venv/lib/python3.5/site-packages/flask/app.py:1718: in handle_user_exception
    reraise(exc_type, exc_value, tb)
venv/lib/python3.5/site-packages/flask/_compat.py:35: in reraise
    raise value
venv/lib/python3.5/site-packages/flask/app.py:1813: in full_dispatch_request
    rv = self.dispatch_request()
venv/lib/python3.5/site-packages/flask/app.py:1799: in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
application/main/routes.py:20: in index
    func.count(Order.id).label("orders_count")
venv/lib/python3.5/site-packages/sqlalchemy/orm/query.py:2947: in one
    ret = self.one_or_none()
venv/lib/python3.5/site-packages/sqlalchemy/orm/query.py:2917: in one_or_none
    ret = list(self)
venv/lib/python3.5/site-packages/sqlalchemy/orm/query.py:2988: in __iter__
    return self._execute_and_instances(context)
venv/lib/python3.5/site-packages/sqlalchemy/orm/query.py:3011: in _execute_and_instances
    result = conn.execute(querycontext.statement, self._params)
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:948: in execute
    return meth(self, multiparams, params)
venv/lib/python3.5/site-packages/sqlalchemy/sql/elements.py:269: in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:1060: in _execute_clauseelement
    compiled_sql, distilled_params
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:1200: in _execute_context
    context)
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:1413: in _handle_dbapi_exception
    exc_info
venv/lib/python3.5/site-packages/sqlalchemy/util/compat.py:265: in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb, cause=cause)
venv/lib/python3.5/site-packages/sqlalchemy/util/compat.py:248: in reraise
    raise value.with_traceback(tb)
venv/lib/python3.5/site-packages/sqlalchemy/engine/base.py:1193: in _execute_context
    context)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <sqlalchemy.dialects.sqlite.pysqlite.SQLiteDialect_pysqlite object at 0x7f1c3f2c4ba8>, cursor = <sqlite3.Cursor object at 0x7f1c3f2c2ce0>
statement = 'SELECT sum("order".col2_count) AS orders_col2, sum("order".col1_count) AS orders_col1, count("order".id) AS orders_count \nFROM "order"', parameters = ()
context = <sqlalchemy.dialects.sqlite.base.SQLiteExecutionContext object at 0x7f1c3f29b6a0>

    def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: order [SQL: 'SELECT sum("order".col2_count) AS orders_col2, sum("order".col1_count) AS orders_col1, count("order".id) AS orders_count \nFROM "order"'] (Background on this error at: http://sqlalche.me/e/e3q8)

venv/lib/python3.5/site-packages/sqlalchemy/engine/default.py:509: OperationalError
=========================================================================================== 1 failed, 1 passed in 0.52 seconds ============================================================================================

这告诉我:db.create_all() 不会在我的测试数据库中创建所有表。任何提示，我在这里做错了什么？

一些附加信息:

• 目前正在使用 sqlite
• 数据库文件本身在文件系统中以 0byte 创建

---

更多调试:我在这里遵循了本指南:https://xvrdm.github.io/2017/07/03/testing-flask-sqlalchemy-database-with-pytest/

这就是事情变得奇怪的地方:

上面的链接:

>>> db.engine.table_names()  # Check the tables currently on the engine
[]                           # no table found
>>> db.create_all()          # Create the tables according to defined models
>>> db.engine.table_names()
['users']                    # Now table 'users' is found

我的项目中发生了什么:

>>> db.engine.table_names()
[]
>>> db.create_all()
>>> db.engine.table_names()
[]
>>>

来自 models.py 的片段:

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()


class Order(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), index=True, unique=True)

Answer 1

您必须使用 flask-sqlalchemy，在幕后，它使用声明式扩展来定义您的模型。

通过子类化 sqlalchemy declarative base类，sqlalchemy 将生成 Table和 mapper ，新创建的表信息存储在相应的 Metadata 中对象。 db.create_all() actually is metadata.create_all()，它只会 create tables stored in the metadata .

因此，在尝试使用metadata.create_all 创建表之前，您必须先将该表的信息存储到metadata 注册表中，这等于定义声明性基子类。在 Python 中，这意味着执行您的类定义代码，进而导入模块定义的类。