query.with_lockmode performs a 'version check' on an already loaded instance
(self)
| 400 | @engines.close_open_connections |
| 401 | @testing.requires.update_nowait |
| 402 | def test_versioncheck_for_update(self): |
| 403 | """query.with_lockmode performs a 'version check' on an already loaded |
| 404 | instance""" |
| 405 | |
| 406 | Foo = self.classes.Foo |
| 407 | |
| 408 | s1 = self._fixture() |
| 409 | f1s1 = Foo(value="f1 value") |
| 410 | s1.add(f1s1) |
| 411 | s1.commit() |
| 412 | |
| 413 | s2 = fixture_session() |
| 414 | f1s2 = s2.get(Foo, f1s1.id) |
| 415 | # not sure if I like this API |
| 416 | s2.refresh(f1s2, with_for_update=True) |
| 417 | f1s2.value = "f1 new value" |
| 418 | |
| 419 | assert_raises( |
| 420 | exc.DBAPIError, s1.refresh, f1s1, with_for_update={"nowait": True} |
| 421 | ) |
| 422 | s1.rollback() |
| 423 | |
| 424 | with conditional_sane_rowcount_warnings(update=True): |
| 425 | s2.commit() |
| 426 | s1.refresh(f1s1, with_for_update={"nowait": True}) |
| 427 | assert f1s1.version_id == f1s2.version_id |
| 428 | |
| 429 | def test_update_multi_missing_broken_multi_rowcount(self): |
| 430 | @util.memoized_property |
nothing calls this directly
no test coverage detected