(self)
| 395 | @testing.requires.independent_connections |
| 396 | @engines.close_open_connections |
| 397 | def test_transaction(self): |
| 398 | User, users = self.classes.User, self.tables.users |
| 399 | |
| 400 | self.mapper_registry.map_imperatively(User, users) |
| 401 | conn1 = testing.db.connect() |
| 402 | conn2 = testing.db.connect() |
| 403 | |
| 404 | sess = Session(bind=conn1) |
| 405 | u = User(name="x") |
| 406 | sess.add(u) |
| 407 | sess.flush() |
| 408 | assert ( |
| 409 | conn1.exec_driver_sql("select count(1) from users").scalar() == 1 |
| 410 | ) |
| 411 | assert ( |
| 412 | conn2.exec_driver_sql("select count(1) from users").scalar() == 0 |
| 413 | ) |
| 414 | sess.commit() |
| 415 | assert ( |
| 416 | conn1.exec_driver_sql("select count(1) from users").scalar() == 1 |
| 417 | ) |
| 418 | |
| 419 | assert ( |
| 420 | testing.db.connect() |
| 421 | .exec_driver_sql("select count(1) from users") |
| 422 | .scalar() |
| 423 | == 1 |
| 424 | ) |
| 425 | sess.close() |
| 426 | |
| 427 | |
| 428 | class SessionUtilTest(_fixtures.FixtureTest): |
nothing calls this directly
no test coverage detected