test_datamanager.py

from hypothesis import given, settings, strategies as st
import mock

from zope.sqlalchemy.datamanager import _SESSION_STATE
from zope.sqlalchemy.datamanager import SessionDataManager


def ok(*args, **kwargs):
    pass


def fail(*args, **kwargs):
    raise RuntimeError("foo")


ok_or_fail = st.sampled_from((ok, fail))
active_or_changed = st.sampled_from(('active', 'changed'))

@given(status=active_or_changed,
       close=ok_or_fail,
       commit=ok_or_fail,
       flush=ok_or_fail,
       expire_all=ok_or_fail)
@settings(max_examples=1000)
def test_datamanager_always_clears_session(status, close, commit, flush, expire_all):
    _SESSION_STATE.clear()

    tm = mock.Mock()

    tx = mock.Mock(spec_set=['commit'])
    tx.commit.side_effect = commit

    session = mock.Mock(spec_set=['close',
                                  'expire_on_commit',
                                  'expire_all',
                                  'flush',
                                  'transaction'])
    session.transaction._iterate_self_and_parents = None
    session.transaction._iterate_parents.return_value = (tx,)
    session.expire_on_commit = True
    session.expire_all.side_effect = expire_all
    session.close.side_effect = close
    session.flush.side_effect = flush

    trans = mock.sentinel.trans

    dm = SessionDataManager(session, status, tm)

    try:
        voted = False
        try:
            dm.tpc_begin(trans)
            dm.commit(trans)
            dm.tpc_vote(trans)
            voted = True
            dm.tpc_finish(trans)
        except:
            if not voted:
                dm.abort(trans)
            dm.tpc_abort(trans)
    except:
        pass

    assert _SESSION_STATE == {}