No easy way to enforce versioning when using nested transactions

[rooterkyberian]

To speed up tests I'm using nested transactions - this allows me to rollback all changes after I'm done testing, and move onto next test without reloading whole database, migrations and fixtures.

https://github.com/kvesteri/sqlalchemy-continuum/blob/345be0361790f1ddd82e3f7870ad64775f98382d/sqlalchemy_continuum/manager.py#L365

Caused my quite a headache as I couldn't figure out why sqlalchemy-continuum fails to work for me during tests (it always stored only the latest version).

It would be nice for this check to be easily disabled, maybe by using if self.options['ignore_nested'] and session.transaction.nested: instead?

[rooterkyberian]

For now I'm using ugly pytest fixture which contains a copy pasted function without this check. I expect it to break at if sqlachemy_continuum changes how clear method looks like or if it gets installed in more places that I don't manually replace after monkey patching so I don't think this is a viable solution.

@pytest.fixture
def sqlalchemy_continuum_nested_mock():
    """
    Patch sqlalchemy-continuum so it creates new version on nested transaction commit

    See https://github.com/kvesteri/sqlalchemy-continuum/issues/204 for details
    """
    import sqlalchemy_continuum
    assert sqlalchemy_continuum.__version__ == '1.3.6', (
        f'Please review this patch as it was untested against sqlalchemy_continuum {sqlalchemy_continuum.__version__}',
    )

    versioning_manager = sqlalchemy_continuum.versioning_manager

    def mock_clear(self, session):
        conn = self.session_connection_map.pop(session, None)
        if conn is None:
            return

        if conn in self.units_of_work:
            uow = self.units_of_work[conn]
            uow.reset(session)
            del self.units_of_work[conn]

        for connection in dict(self.units_of_work).keys():
            if connection.closed or conn.connection is connection.connection:
                uow = self.units_of_work[connection]
                uow.reset(session)
                del self.units_of_work[connection]

    def reinstall_clear_ref():
        versioning_manager.remove_session_tracking(_db.session)
        versioning_manager.session_listeners.update({
            'after_commit': getattr(versioning_manager, 'clear'),
            'after_rollback': getattr(versioning_manager, 'clear'),
        })
        versioning_manager.track_session(_db.session)

    with patch.object(versioning_manager, 'clear', lambda *args: mock_clear(versioning_manager, *args)):
        reinstall_clear_ref()
        yield
    reinstall_clear_ref()