Method test_basic

test/orm/test_versioning.py:199–254

(self)

    @provision.allow_stale_updates
    def test_basic(self):
        Foo = self.classes.Foo

        s1 = self._fixture()
        f1 = Foo(value="f1")
        f2 = Foo(value="f2")
        s1.add_all((f1, f2))
        s1.commit()

        f1.value = "f1rev2"
        with conditional_sane_rowcount_warnings(update=True):
            s1.commit()

        s2 = fixture_session()
        f1_s = s2.get(Foo, f1.id)
        f1_s.value = "f1rev3"
        with conditional_sane_rowcount_warnings(update=True):
            s2.commit()

        f1.value = "f1rev3mine"

        # Only dialects with a sane rowcount can detect the
        # StaleDataError
        if testing.db.dialect.supports_sane_rowcount:
            assert_raises_message(
                sa.orm.exc.StaleDataError,
                r"UPDATE statement on table 'version_table' expected "
                r"to update 1 row\(s\); 0 were matched.",
                s1.commit,
            ),
            s1.rollback()
        else:
            with conditional_sane_rowcount_warnings(update=True):
                s1.commit()

        # new in 0.5 !  don't need to close the session
        f1 = s1.get(Foo, f1.id)
        f2 = s1.get(Foo, f2.id)

        f1_s.value = "f1rev4"
        with conditional_sane_rowcount_warnings(update=True):
            s2.commit()

        s1.delete(f1)
        s1.delete(f2)

        if testing.db.dialect.supports_sane_multi_rowcount:
            assert_raises_message(
                sa.orm.exc.StaleDataError,
                r"DELETE statement on table 'version_table' expected "
                r"to delete 2 row\(s\); 1 were matched.",
                s1.commit,
            )
        else:
            with conditional_sane_rowcount_warnings(delete=True):
                s1.commit()

    def test_multiple_updates(self):