Jak łączyć urządzenia py.test z Flask-SQLAlchemy i PostgreSQL?

Question

Staram się pisać py.test oprawy do zarządzania bazą danych mojej aplikacji, które maksymalizują prędkość, wspiera pytest-xdist równoległość testów i izoluje testy od siebie nawzajem.

Używam Flask-SQLAlchemy 2.1 przeciwko bazie danych PostgreSQL 9.4.

Oto ogólny zarys tego, co próbuję wykonać:

1. $ py.test -n 3 obraca się trzy sesje testowe do uruchamiania testów.

2. W ramach każdej sesji urządzenie py.test uruchamia się raz, aby skonfigurować transakcję, utworzyć tabele bazy danych, a następnie po zakończeniu sesji wycofuje transakcję. Tworzenie tabel bazy danych musi się odbywać w ramach transakcji PostgreSQL, która jest widoczna tylko dla danej sesji testowej, w przeciwnym razie równoległe sesje testowe utworzone przez pytest-xdist powodują konflikty między sobą.

3. Drugie urządzenie py.test uruchamiane dla każdego testu łączy się z istniejącą transakcją w celu wyświetlenia utworzonych tabel, utworzenia zagnieżdżonego punktu zapisu, uruchomienia testu, a następnie wycofania do zagnieżdżonego punktu zapisu.

4. Idealnie, te urządzenia pytestowe obsługują testy, które wywołują db.session.rollback(). Istnieje potencjalny przepis na dokonanie tego na końcu tego SQLAlchemy doc.

5. Idealnie uchwyty pytest powinna przynieść przedmiot db, nie tylko dlatego, że sesję ludzie mogą pisać testy bez konieczności pamiętania użyć sesji, który jest inny niż standardowego db.session oni wykorzystać w całej aplikacji.

Oto co mam do tej pory:

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 

Oto najbardziej użyteczne odniesienia, które znalazłem podczas googlowania:

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

Podobny problem próbowałem połączyć z urządzeniami yield. Niestety zgodnie z doc nie można połączyć więcej niż jednego poziomu yield.

Ale może być w stanie znaleźć pracę wokół używając 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