Best practices to test a webapp using scoped_session with defined scope

[José Luis Lafuente]

Maybe not the best title, but I'll try to explain the problem.

I'm using sqlalchemy and pyramid on my app. I'm using a scoped_session on every request:


from pyramid.threadlocal import get_current_request
Session = scoped_session(sessionmaker(), scopefunc=get_current_request)


I want to write some functional tests, I'm using py.test for it:


import pytest
from myapp import Session, Model

@pytest.fixture(scope='function')
def webapp(request):

    from myapp import main
    from webtest import TestApp

    app = main({}, **settings)
    Session.begin(subtransactions=True)

    def fin():
        Session.rollback()
        Session.remove()
    request.addfinalizer(fin)

    return TestApp(app)

def test_search(webapp):
    Session.add(Model(name='dummy'))
    res = webapp.get('/', status=200)
    assert res.body == 'something'


The problem I have with this code is inside the web request to '/'. I query Session for the new model, but nothing is found. However, this assert will success:

assert Session.query(Model).filter_by(name='dummy').one().name == 'Aa'

I think the problem is related to the argument 'scopefunc=get_current_request' of the scoped_session, and the fact that the point where I add the model have a different scope that the web request.
To solve it, I'm patching the session this way:

...
from sqlalchemy.util import ThreadLocalRegistry

@pytest.fixture(scope='function')
def webapp(request, monkeypatch):
    monkeypatch.setattr('myapp.Session.registry', ThreadLocalRegistry(Session.session_factory))
...

Is it a correct way? Are there better ways to add models to the database for a functional test, rolling back them afterwards?

Thanks,

JL

[Mike Bayer]

I'm not sure why you need to do something with "thread local" inside of a test, as I wouldn't expect to support test suites that run concurrently in multiple threads.     If you're looking for a test fixture where the thing surrounding all of it is "the transaction", you should create the transaction using the Core API, then associate all sessions with that connection/transaction.  The example at http://docs.sqlalchemy.org/en/rel_1_0/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites illustrates this approach in the general sense.  In this case, with the global scoped_session() you probably can just patch in an alternate sessionmaker that includes the target Connection as the bind, or even call myapp.Session.configure(bind=my_connection), though you'd have to revert "bind" to its former value afterwards.

Thanks,

JL

--
You received this message because you are subscribed to the Google Groups "sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email to sqlalchemy+...@googlegroups.com.
To post to this group, send email to sqlal...@googlegroups.com.
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.

[José Luis Lafuente]

Thanks for your answer.

I want to run the tests sequentially. The reason to set the session
registry to ThreadLocalRegistry was just to simulate the
scoped_session creation with scopefunc set to None, just as here:

https://bitbucket.org/zzzeek/sqlalchemy/src/148924821bf8c8537cf8ece740b26a61571215fb/lib/sqlalchemy/orm/scoping.py?at=rel_1_0#cl-44

I think the main problem using scoped_session with scopedfunc is that
when I add a model for testing, I'm not inside a web request, and
scopedfunc ( in my case set to
pyramid.threadlocal.get_current_request) returns None. So in the
registry of the scoped_session, there is now one session. When later
I'm inside the web request, the scoped_session's registry calls again
to scopedfunc, but now gets a new result and creates a second session
in the registry. The query doesn't return anything because the insert
was done in another session.

Not sure if this description is correct, but is what makes senses to
me after debugging and looking at SQLAlchemy code.

If it is correct, an alternative would be this patch:

monkeypatch.setattr('myapp.Session.registry.scopefunc', lambda: None)

to enforce the use of the same session. Does it make sense?

I also tried to patch myapp.Session directly, but since is a global
object I find it hacky. About the example you provided, I don't see
how it applies to my case, because my setUp is called before any
transaction.

> You received this message because you are subscribed to a topic in the
> Google Groups "sqlalchemy" group.
> To unsubscribe from this topic, visit
> https://groups.google.com/d/topic/sqlalchemy/zTuZExRXZHE/unsubscribe.
> To unsubscribe from this group and all its topics, send an email to

[Mike Bayer]

On 5/11/15 2:04 PM, José Luis Lafuente wrote:
> Thanks for your answer.
>
> I want to run the tests sequentially. The reason to set the session
> registry to ThreadLocalRegistry was just to simulate the
> scoped_session creation with scopefunc set to None, just as here:
>
> https://bitbucket.org/zzzeek/sqlalchemy/src/148924821bf8c8537cf8ece740b26a61571215fb/lib/sqlalchemy/orm/scoping.py?at=rel_1_0#cl-44
>
> I think the main problem using scoped_session with scopedfunc is that
> when I add a model for testing, I'm not inside a web request, and
> scopedfunc ( in my case set to
> pyramid.threadlocal.get_current_request) returns None. So in the
> registry of the scoped_session, there is now one session. When later
> I'm inside the web request, the scoped_session's registry calls again
> to scopedfunc, but now gets a new result and creates a second session
> in the registry. The query doesn't return anything because the insert
> was done in another session.

OK, I'm not currently reading this too deeply because running
transactional tests with pyramid and a scoped_session + request scope
is a solved problem. Take a look at my example application at
https://bitbucket.org/zzzeek/pycon2014_atmcraft/ which does all of these
things using idiomatic Pyramid test fixtures.