test that version number can be bumped. Ensures that the UPDATE or DELETE is against the last committed version of version_id_col, not the modified state.
(self)
| 309 | ) |
| 310 | |
| 311 | def test_bump_version(self): |
| 312 | """test that version number can be bumped. |
| 313 | |
| 314 | Ensures that the UPDATE or DELETE is against the |
| 315 | last committed version of version_id_col, not the modified |
| 316 | state. |
| 317 | |
| 318 | """ |
| 319 | |
| 320 | Foo = self.classes.Foo |
| 321 | |
| 322 | s1 = self._fixture() |
| 323 | f1 = Foo(value="f1") |
| 324 | s1.add(f1) |
| 325 | s1.commit() |
| 326 | eq_(f1.version_id, 1) |
| 327 | f1.version_id = 2 |
| 328 | with conditional_sane_rowcount_warnings(update=True): |
| 329 | s1.commit() |
| 330 | eq_(f1.version_id, 2) |
| 331 | |
| 332 | # skip an id, test that history |
| 333 | # is honored |
| 334 | f1.version_id = 4 |
| 335 | f1.value = "something new" |
| 336 | with conditional_sane_rowcount_warnings(update=True): |
| 337 | s1.commit() |
| 338 | eq_(f1.version_id, 4) |
| 339 | |
| 340 | f1.version_id = 5 |
| 341 | s1.delete(f1) |
| 342 | with conditional_sane_rowcount_warnings(delete=True): |
| 343 | s1.commit() |
| 344 | eq_(s1.query(Foo).count(), 0) |
| 345 | |
| 346 | @provision.allow_stale_updates |
| 347 | @engines.close_open_connections |