0

在每个测试功能运行后，我编写了一个 pytest 夹具来回滚数据，这似乎没有任何意义。夹具如下：

@pytest.fixture(scope='function')
def session(db, request):
    """Creates a new database session for a test."""
    connection = db.engine.connect()
    transaction = connection.begin()
    options = dict(bind=connection, binds={})
    session = db.create_scoped_session(options=options)
    db.session = session

    def teardown():
        """rollback data after every function has run"""
        transaction.rollback()
        session.commit()
        connection.close()
        session.remove()
    request.addfinalizer(teardown)
    return session

session可以用作测试函数的参数，但 似乎transaction.rollback()不起作用。我怎样才能修改它以使其工作？

更新：

对于尾巴，原点夹具是

`@pytest.fixture(scope='function')
def session(db, request):
    """Creates a new database session for a test."""
    connection = db.engine.connect()
    transaction = connection.begin()
    options = dict(bind=connection, binds={})
    session = db.create_scoped_session(options=options)
    db.session = session

    def teardown():
        """rollback data after every function has run"""
        transaction.rollback()
        session.commit()
        connection.close()
        session.remove()

    def clear_all_data():
        meta = db.metadata
        for tl in reversed(meta.sorted_tables):
            # print('clear table %s' % tl)
            db.session.execute(tl.delete())

        db.session.commit()

    request.addfinalizer(teardown)
    return session`

和测试功能是

`def test_check_name(client, session, db):
    # name can not be blank
    cur_url = BLUEPREFIX + '/checkname'
    rv = client.post(cur_url).data.decode('utf-8')
    assert get_json_err_no(rv, '15')

    # name can not exist already
    customer = Customer(name='wt', pinyin='wt')
    session.add(customer)
    session.commit()
    rv = client.post(cur_url, data=dict(
        name='wt')).data.decode('utf-8')
    assert get_json_err_no(rv, '14')

    rv = client.post(cur_url, data=dict(
        name='wtf')).data.decode('utf-8')
    assert get_json_err_no(rv, '17')
def test_del_customer(client, session, db):
    cur_url = BLUEPREFIX + '/del'

    # check invalid post
    rv = client.post(cur_url).data.decode('utf-8')
    assert get_json_err_no(rv, '18')

    rv = client.post(cur_url, data=dict(
        customer_id='1a')).data.decode('utf-8')
    assert get_json_err_no(rv, '8')

    print(Customer.query.count())
    rv = client.post(cur_url, data=dict(
        customer_id='1')).data.decode('utf-8')
    assert get_json_err_no(rv, '7')

    customer = Customer(name='wt', pinyin='wt')
    session.add(customer)
    session.commit()
    assert customer.id == 1

    rv = client.post(cur_url, data=dict(
        customer_id=str(customer.id))).data.decode('utf-8')
    assert get_json_err_no(rv, '0')`

我跑py.test -s -v -k "not add"了，结果是

`============================================= test session starts =============================================

...

test_customer.py::test_check_name --------------------------------------------------------------------------------
DEBUG in customer [/home/dyh/svn/hs/hs/views/customer.py:21]:
customer of same name exists !
--------------------------------------------------------------------------------
PASSED
test_customer.py::test_del_customer 1
FAILED
test_helloworld.py::test_answer PASSED

================================================== FAILURES ===================================================
______________________________________________ test_del_customer ______________________________________________

client = <FlaskClient <Flask 'hs'>>, session = <sqlalchemy.orm.scoping.scoped_session object at 0x7f06da034898>
db = <SQLAlchemy engine='sqlite:////tmp/tmp90nyrdqr'>

    def test_del_customer(client, session, db):
        cur_url = BLUEPREFIX + '/del'

        # check invalid post
        rv = client.post(cur_url).data.decode('utf-8')
        assert get_json_err_no(rv, '18')

        rv = client.post(cur_url, data=dict(
            customer_id='1a')).data.decode('utf-8')
        assert get_json_err_no(rv, '8')

        print(Customer.query.count())
        rv = client.post(cur_url, data=dict(
            customer_id='1')).data.decode('utf-8')
>       assert get_json_err_no(rv, '7')
E       assert get_json_err_no('{\n  "err_msg": "\\u64cd\\u4f5c\\u6210\\u529f!",\n  "err_no": "0"\n}', '7')

test_customer.py:90: AssertionError
====================================== 1 tests deselected by '-knot add' ======================================
============================== 1 failed, 2 passed, 1 deselected in 0.06 seconds ===============================`
4

1 回答 1

0

不要使用addfinalizer，而是查看本手册，并使用yield_fixture py.test功能。轻量级示例：

import pytest

@pytest.yield_fixture
def session(db):
    connection = db.engine.connect()
    # ...
    session = db.create_scoped_session()
    yield session
    session.remove()
    # ...
    connection.close()

更新：2021-12-08

重要的：

从pytest-3.0 开始，使用普通夹具装饰器的夹具可以使用yield语句来提供夹具值并执行拆卸代码，就像yield_fixture在以前的版本中一样。

仍支持将函数标记为yield_fixture，但已弃用，不应在新代码中使用。

于 2016-03-18T14:24:07.813 回答