Làm thế nào để kết hợp đồ đạc py.test với Flask-SQLAlchemy và PostgreSQL?

Question

Tôi đang cố gắng viết các đồ đạc py.test để quản lý cơ sở dữ liệu của ứng dụng nhằm tối đa hóa tốc độ, hỗ trợ pytest-xdist song song các thử nghiệm và cách ly các thử nghiệm với nhau.

Tôi đang sử dụng Flask-SQLAlchemy 2.1 so với cơ sở dữ liệu PostgreSQL 9.4.

Dưới đây là phác thảo chung về những gì tôi đang cố gắng để hoàn thành:

1. $ py.test -n 3 quay lên ba buổi thử nghiệm cho chạy thử nghiệm.

2. Trong mỗi phiên, một lịch thi đấu py.test chạy một lần để thiết lập giao dịch, tạo bảng cơ sở dữ liệu và sau đó vào cuối phiên, nó quay lại giao dịch. Việc tạo các bảng cơ sở dữ liệu cần xảy ra trong một giao dịch PostgreSQL chỉ hiển thị với phiên kiểm tra cụ thể đó, nếu không các phiên kiểm tra song song được tạo ra bởi pytest-xdist sẽ gây xung đột với nhau.

3. Một vật cố py.test thứ hai chạy cho mọi thử nghiệm kết nối với giao dịch hiện có để xem các bảng đã tạo, tạo một điểm lưu trữ lồng nhau, chạy kiểm tra, sau đó cuộn trở lại điểm lưu trữ lồng nhau.

4. Lý tưởng nhất, các đồ đạc tinh tế nhất này hỗ trợ các thử nghiệm gọi db.session.rollback(). Có một công thức tiềm năng để hoàn thành công việc này ở cuối số SQLAlchemy doc này.

5. Lý tưởng nhất là đồ đạc pytest nên năng suất đối tượng db, không chỉ là phiên để folks có thể viết bài kiểm tra mà không cần phải nhớ để sử dụng một phiên đó là khác nhau hơn so với tiêu chuẩn db.session họ sử dụng trong suốt ứng dụng.

Dưới đây là những gì tôi có cho đến nay:

import pytest 

# create_app() is my Flask application factory 
# db is just 'db = SQLAlchemy()' + 'db.init_app(app)' within the create_app() function 
from app import create_app, db as _db 


@pytest.yield_fixture(scope='session', autouse=True) 
def app(): 
    '''Session-wide test application''' 
    a = create_app('testing') 
    with a.app_context(): 
     yield a 

@pytest.yield_fixture(scope='session') 
def db_tables(app): 
    '''Session-wide test database''' 
    connection = _db.engine.connect() 
    trans = connection.begin() # begin a non-ORM transaction 

    # Theoretically this creates the tables within the transaction 
    _db.create_all() 
    yield _db 
    trans.rollback() 
    connection.close() 

@pytest.yield_fixture(scope='function') 
def db(db_tables): 
    '''db session that is joined to existing transaction''' 

    # I am quite sure this is broken, but it's the general idea 

    # bind an individual Session to the existing transaction 
    db_tables.session = db_tables.Session(bind=db_tables.connection) 

    # start the session in a SAVEPOINT... 
    db_tables.session.begin_nested() 

    # yield the db object, not just the session so that tests 
    # can be written transparently using the db object 
    # without requiring someone to understand the intricacies of these 
    # py.test fixtures or having to remember when to use a session that's 
    # different than db.session 
    yield db_tables 

    # rollback to the savepoint before the test ran 
    db_tables.session.rollback() 
    db_tables.session.remove() # not sure this is needed 

Đây là tài liệu tham khảo hữu ích nhất mà tôi đã tìm thấy trong khi googling:

http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites

http://koo.fi/blog/2015/10/22/flask-sqlalchemy-and-postgresql-unit-testing-with-transaction-savepoints/

https://github.com/mitsuhiko/flask-sqlalchemy/pull/249

Answer 1

Tôi gặp sự cố tương tự khi cố gắng kết hợp các đồ đạc yield. Rất tiếc, theo số doc bạn không thể kết hợp nhiều cấp độ yield.

Nhưng bạn có thể tìm một công việc xung quanh bằng request.finalizer:

@pytest.fixture(scope='session', autouse=True) 
def app(): 
    '''Session-wide test application''' 
    a = create_app('testing') 
    with a.app_context(): 
     return a 

@pytest.fixture(scope='session') 
def db_tables(request, app): 
    '''Session-wide test database''' 
    connection = _db.engine.connect() 
    trans = connection.begin() # begin a non-ORM transaction 

    # Theoretically this creates the tables within the transaction 
    _db.create_all() 
    def close_db_session(): 
     trans.rollback() 
     connection.close() 
    request.addfinalizer(close_db_session) 
    return _db