Method test_versioncheck

test/orm/test_versioning.py:348–384

query.with_lockmode performs a 'version check' on an already loaded instance

(self)

    @provision.allow_stale_updates
    @engines.close_open_connections
    def test_versioncheck(self):
        """query.with_lockmode performs a 'version check' on an already loaded
        instance"""

        Foo = self.classes.Foo

        s1 = self._fixture()
        f1s1 = Foo(value="f1 value")
        s1.add(f1s1)
        s1.commit()

        s2 = fixture_session()
        f1s2 = s2.get(Foo, f1s1.id)
        f1s2.value = "f1 new value"
        with conditional_sane_rowcount_warnings(update=True):
            s2.commit()

        # load, version is wrong
        assert_raises_message(
            sa.orm.exc.StaleDataError,
            r"Instance .* has version id '\d+' which does not "
            r"match database-loaded version id '\d+'",
            s1.get,
            Foo,
            f1s1.id,
            with_for_update=dict(read=True),
        )

        # reload it - this expires the old version first
        s1.refresh(f1s1, with_for_update={"read": True})

        # now assert version OK
        s1.get(Foo, f1s1.id, with_for_update=dict(read=True))

        # assert brand new load is OK too
        s1.close()
        s1.get(Foo, f1s1.id, with_for_update=dict(read=True))

    def test_versioncheck_not_versioned(self):
        """ensure the versioncheck logic skips if there isn't a