Understanding how to perform functional testing using transactions with pyramid + SQLAlchemy + pytest + webtest for views that use a postgreSQL database.

[dcs3spp]

Hi,

I am a newbie having difficulty understanding and getting functional testing working with pyramid, SQLAlchemy, pytest and webtest. I am using pyramid 1.10. Hoping that someone is able to advise a way forward or direct me to any useful resources.

I have written the fixture below that creates a SQL Alchemy session for each test and initialises data within a transaction, based upon documentation for functional testing at the wiki that uses unittest.

When the fixture completes the transaction aborts and closes the session. When the next test runs the fixture will create a new transaction and reinitialise the data.

   

@pytest.fixture

def session(request, testapp):

    factory = testapp.app.registry['dbsession_factory']

    engine = factory.kw['bind']

    # create all the tables in db

    Base.metadata.create_all(engine)

    log.info ("Creating root transaction for the test session")

    with transaction.manager as tx:

        from plantoeducate_data.models import get_tm_session

        session=get_tm_session(factory, transaction.manager)

        brief=DocumentTemplateModel ()

        brief.DocumentDescription='brief'

        brief.DocumentTypeID=1

        brief.DocumentFilePath='brief.ott'

        feedback=DocumentTemplateModel ()

        feedback.DocumentDescription='feedback'

        feedback.DocumentTypeID=2

        feedback.DocumentFilePath='feedback.ott'

        session.add_all([brief, feedback])

        #session.flush()

        yield session

        log.info("Rolling back root transaction")

        transaction.abort()

        session.close()

    

I have two tests that use the fixture, listed below:

def test_delete_document(self, testapp, session):

        doc=session.query(DocumentTemplateModel).first()

        import pdb; pdb.set_trace()

        # delete the document

        res = testapp.delete('/documents/templates/{}'.format(doc.DocumentID), status=204)

def test_filter_documents(self, testapp, session):

         res = testapp.get('/documents/templates/1', status=200)

    expectedTemplatesCount = 2

    import pdb; pdb.set_trace()

I think that pyramid_tm creates a session from a SQLAlchemy session factory for each request and hooks up the current active transaction. 

When the first test is run I can see data in the session, however the request session in the view does not see the data. 

When the second test runs there is no data at all in the session that was created by the test fixture.

How do I make the data initialised in the test visible to the view being tested? Is it possible to perform testing by initialising data in SQLAlchemy, making it visible to the request session in the view and then rolling back state in preparation for subsequent test? 

Kind Regards

dcs3spp

[Mikko Ohtamaa]

Hi dc3spp,

For inspiration, you can check out how Websauna doees it:

https://websauna.org/docs/narrative/testing/writing.html#basic-integration-testing-pattern

Pyramid does not provide you test fixtures or relevant plumping for managing transaction lifecycle in functional testing. Your best shot is to copy-paste fixtures from Websauna, but they come with a lot of tidbits like INI driven configuration for development and tests alike.

-Mikko

[dcs3spp]

Hi Mikko,

Cheers will give it a try.....

Kind Regards

dcs3spp

[andi]

Hi dcs3spp, 

 

I remember it was pretty complicated for me to setup something as you describe: one sqla session per test, which will be automatically rolled back. 

 

You are right, that in the TestApp there is a separate transaction handling, which destroys your approach. To solve that problem for me (no idea if this was elegant or not, but it does what you describe) I did the following.

 

Created a TestApp with a custom registry field: 

@pytest.fixture(scope="function")
def _test_app(router: Router) -> TestApp:
    return TestApp(router)


@pytest.fixture(scope="function")
def test_app(_test_app: TestApp, _sqla_session) -> TestApp:
    """wrapper to ensure the fixture-created `sqla_session` will be picked up in `test_app`"""
    _test_app.app.registry.settings['paste.testing'] = True
    _test_app.app.registry['paste.testing.session'] = _sqla_session
    return _test_app

along with the usual sqla fixure

 

@pytest.fixture(scope="session")
def _sqla_session(pyramid_config, sqla_engine) -> Session:
    """
    Depending on this fixture is comparable to a integration test, which has nothing more than
    the orm properly defined. Which is helpful, but not the full application configured.
    """
    session_factory = get_session_factory(sqla_engine, db_session)
    return session_factory()


@pytest.fixture(scope="function")
def sqla_session(_sqla_session: Session, test_app):
    """
    wrap a transaction
    """
    # TODO andi: magically there is a transaction active here. why.
    # t = _sqla_session.begin()
    yield _sqla_session

    # this is the important `rollback`
    _sqla_session.transaction.rollback()

 

Then I have something, that probably `pyramid_tm` does under the hood: a request listener, which handles a “per request session”. 

 

def add_tm_session(req):
    # this property is set in `webtest.app.TestApp#do_request`
    if 'paste.testing' in req.environ and req.environ['paste.testing'] is True:
        from pyramid.threadlocal import get_current_registry

        registry = get_current_registry()
        # in case of integration testing, we set this registry key with the `sqla_session` from
        # around. this allows us to operate in the same session, so we don't need any commits.
        return registry.get('paste.testing.session', None)
    # request.tm is the transaction manager used by pyramid_tm
    return get_tm_session(session_factory, req.tm)

# make request.dbsession available for use in Pyramid
config.add_request_method(add_tm_session, 'dbsession', reify=True)

You can see the “magic” in there. 

 

What I found in my documentation was this link (another was dead already), which I found helpful: https://gist.github.com/inklesspen/4504383

 

The good thing about this pattern for me was, that I could apply this to Celery jobs in tests or Spyne integration - aka pretty flexible. 

 

Hope this helps. 

 

Andi

 

--
You received this message because you are subscribed to the Google Groups "pylons-discuss" group.
To unsubscribe from this group and stop receiving emails from it, send an email to pylons-discus...@googlegroups.com.
To post to this group, send email to pylons-...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/pylons-discuss/bf247ef1-1a48-4934-9bce-205d166a5c64%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

 

[dcs3spp]

Hi Andi,

Thankyou for sharing the fixtures and approach, much appreciated. So the SQLAlchemy test session is shared in the registry which triggers a bypass of the pyramid_tm session for the request when testing.

That would solve the problem I have been encountering since, with this approach the session from testing is shared with web request thread and rolled back per test. 

I think that the difficulties I have been facing were as a result of me using two separate SQLAlchemy sessions, one for the test and one per pyramid web request. My approach in earlier code posting was an attempt to try to bind the sessions to an outer transaction for the test and an inner nested transaction for the web requested. However, I think they were separate transaction managers. I quickly became confused and lost concerning the effects of nested transactions. Presumably, once the inner transaction web request commits it would be difficult to rollback to the outer test transaction anyway.

In summary, like your idea will give it a try....

Thanks again for sharing the code and concept. 

Kind Regards

dcs3spp

[andi]

 

Welcome dcs3spp!

 

One thing I noted in your snippet was you where using pdb (or ipdb). That looks like you’re not using an IDE that is able to debug ;)

 

I would  really recommend using PyCharm for such tinkering, cause even the community version has a well integrated debugger and supports py.test (once you configured your project to use it – one click in the configuration).

 

Best, Andi

To view this discussion on the web visit https://groups.google.com/d/msgid/pylons-discuss/8daf90a7-3485-4a1c-8cbf-caa9153a054f%40googlegroups.com.

[dcs3spp]

Cheers Andi will have a look at PyCharm :)

[Michael Merickel]

Andi, I think this is a fantastic approach. The key here is to override request.dbsession to a mocked out version which is not connected to pyramid_tm at all, giving you full control of commit/rollback from the outside. One extra step would be to disable pyramid_tm entirely by setting environ['tm.active'] = True which would let you mock out request.tm if you were using it for any other purposes. This can be done from the TestApp constructor (you could also use this for the dbsession to avoid using the get_current_request threadlocal) via TestApp(app, extra_environ={'tm.active': True, 'dbsession': dbsession}).

We should look at adding this pattern to the cookiecutter to make it more accessible for people.

To view this discussion on the web visit https://groups.google.com/d/msgid/pylons-discuss/A1460691-D307-4B5C-846C-E8CD20BE8FA3%40googlemail.com.

[andi]

 

 

Thanks Michael,

 

I’d be glad to have this appear in a cookiecutter – let me know how I can help.

 

For the record, what I missed to mention (but is actually the bigger selling point for me):

Apart from the automatic rollback functionality, that approach allows putting anything to the session, no matter where. I.e. you can `add` and `flush` something in a test, before calling the `TestApp` in the same test, which will have access to what was `add`ed to that session. Still without anything being committed. You can imagine how easy writing integration tests is in the end.

 

Best, Andi

 

 

From: <pylons-...@googlegroups.com> on behalf of Michael Merickel <mmer...@gmail.com>
Reply-To: <pylons-...@googlegroups.com>
Date: Tuesday, 6. November 2018 at 16:58
To: <pylons-...@googlegroups.com>
Subject: Re: [pylons-discuss] Understanding how to perform functional testing using transactions with pyramid + SQLAlchemy + pytest + webtest for views that use a postgreSQL database.

 

Andi, I think this is a fantastic approach. The key here is to override request.dbsession to a mocked out version which is not connected to pyramid_tm at all, giving you full control of commit/rollback from the outside. One extra step would be to disable pyramid_tm entirely by setting environ['tm.active'] = True which would let you mock out request.tm if you were using it for any other purposes. This can be done from the TestApp constructor (you could also use this for the dbsession to avoid using the get_current_request threadlocal) via TestApp(app, extra_environ={'tm.active': True, 'dbsession': dbsession}).

To view this discussion on the web visit https://groups.google.com/d/msgid/pylons-discuss/CAKdhhwFcagH3GzLMdJZ2qCa%3D9csv5nEzcRR47%2Bg5k%2BYFMH1kRA%40mail.gmail.com.

[dcs3spp]

Hi,

Just a quick followup message. Thanks again Andi for sharing your idea and thanks Michael for the idea of using the testapp environment to pass in session and/or request transaction manager. Your suggestions have really helped a Python and Pyramid newbie! Fingers crossed, I think I have managed to get it working now with some tweaking here and there....bit pushed for time so quickly pasting the fixtures, request method and sample test.

Here are the pytest fixtures.....

@pytest.fixture(scope="session")
def settings():
    """ Return a dictionary of application settings defined in test.ini, including expanded environment variables.
    :return: Dictionary of application settings defined in test.ini.
    """
    testconfigfile = project_path(pytest.config.rootdir.strpath, 'test.ini')
    settings = get_appsettings(testconfigfile, name='main')
    settings = load_environment(testconfigfile, settings)
    conf = {'global_config': {'__file__': testconfigfile}}
    settings.update(conf)

    return settings


@pytest.fixture(scope="function")
def testapp(settings, session):
    """ Wrap `webtest.TestApp` request within a doomed transaction that aborts when a test completes.

    Attempts to commit will raise a `DoomedTransaction` exception. Doomed transactions can only be aborted.
    The test app instance is initialised with environment tm.active set to True to instruct pyramid_tm to bypass
    creating transaction for request.

    :param settings: dictionary of application settings contained within `test.ini`.
    :param session: SQLAlchemy session.
    :return: webtest.TestApp object.
    """
    testing.setUp()

    log.info("testapp fixture :: starting a doomed transaction")

    with transaction.manager as tx:
        tx.doom()
        from plantoeducate_data.models import set_tm_session
        app = main(**settings)
        testapp = webtest.TestApp(app, extra_environ={'db.session': session, 'tm.active': True})
        set_tm_session(session, transaction.manager)

        yield testapp

        log.info("testapp fixture :: aborting transaction")

        tx.abort()
        testing.tearDown()


@pytest.fixture(scope="session")
def session(settings):
    """ Return a SQLAlchemy session for the database referenced by `sqlalchemy.url` within test.ini file.

    When tests have been run the session is closed and the associated engine is disposed.

    :param settings: dictionary of application settings contained within `test.ini`.
    :return: SQLAlchemy session object.
    """
    engine = get_engine(settings, prefix='sqlalchemy.')
    factory = get_session_factory(engine)

    Base.metadata.create_all(engine)

    session = factory()

    yield session

    session.close()
    engine.dispose()

and here is the request method.....

def set_tm_session(session, transaction_manager):
    zope.sqlalchemy.register(session, transaction_manager=transaction_manager)
    return session


def includeme(config):

...

# use pyramid_tm to hook the transaction lifecycle to the request
config.include('pyramid_tm')

session_factory = get_session_factory(get_engine(filtered_settings))
config.registry['dbsession_factory'] = session_factory

def add_tm_session(request):
    """ Request method that returns a SQLAlchemy session for a request.

    The SQLAlchemy session is managed by a Zope transaction, unless the request has been generated from a
    webtest.TestApp instance for functional testing. In this case:

    - Inspect request.environ dictionary for the SQLAlchemy session referenced by key db.session. Remove the
        session from the request's environment dictionary and return the session.
    - Use the session factory to generate and return a new SQLAlchemy session if there is no entry for db.session
        in the request environment dictionary.

    When functional testing it is the responsibility of the test client to configure the Zope transaction
    manager for the SQLAlchemy session. For example:

        with transaction.manager as tx:
            set_tm_session(session, tx)

    The webtest.TestApp instance should configure the environment dictionary as follows:
        `testapp = webtest.TestApp(app, extra_environ={'db.session': session, 'tm.active': True})`

    Setting tm.active to True causes the pyramid_tm tween to bypass generating a transaction for the SQLAlchemy
    session on the request.

    :param request: Pyramid Request instance
    :return: SQLAlchemy session.
    """
    if 'paste.testing' in request.environ and request.environ['paste.testing'] is True:
        if 'db.session' in request.environ:# and 'db.tm' in request.environ:
            dbsession = request.environ['db.session']
            del request.environ['db.session']
            return dbsession
            #return set_tm_session(request.environ['db.session'], request.environ['db.tm'])
        else:
            return session_factory()

    return get_tm_session(session_factory, request.tm)w


config.add_request_method(add_tm_session, 'dbsession', reify = True)

and a sample test fixture.....

def test_filter_documents_feedback(self, testapp, documents):

    res = testapp.get('/documents/templates/2', status=200)
    log.info("test_filter_documents_feedback :: completed filter request")

    expected_feedback_templates_count = 1
    expected_templates_count = 2

    assert(len(res.json['data']) == expected_feedback_templates_count)
    assert(documents.query(DocumentTemplateModel).count() == expected_templates_count)

The testapp fixture provides access to a webtest.TestApp instance wrapped by a transaction. The webtest.TestApp instance stores a SQLAlchemy session in the extra_environ dict referenced by key db.session which is available for the pyramid request method to grab and make available on each request.  

The documents fixture populates some data in the session and then yields the session to the test.

Thanks again!!!

Kind regards

dcs3spp

[Steve Piercy]

On 11/6/18 at 6:28 PM, pylons-...@googlegroups.com ('andi'
via pylons-discuss) pronounced:

>I’d be glad to have this appear in a cookiecutter – let me know how I can help.

Please submit a PR to the master branch of:
https://github.com/pylons/pyramid-cookiecutter-starter

Details for contributing:
https://github.com/Pylons/pyramid-cookiecutter-starter/blob/latest/CONTRIBUTING.md

--steve

------------------------
Steve Piercy, Eugene, OR