| 630 | eq_(finalizer.call_count, 0) |
| 631 | |
| 632 | def test_conn_reusable(self): |
| 633 | conn = self.db.connect() |
| 634 | |
| 635 | conn.execute(select(1)) |
| 636 | |
| 637 | eq_(self.dbapi.connect.mock_calls, [self.mock_connect]) |
| 638 | |
| 639 | self.dbapi.shutdown() |
| 640 | |
| 641 | with expect_raises(tsa.exc.DBAPIError): |
| 642 | conn.execute(select(1)) |
| 643 | |
| 644 | assert not conn.closed |
| 645 | assert conn.invalidated |
| 646 | |
| 647 | eq_([c.close.mock_calls for c in self.dbapi.connections], [[call()]]) |
| 648 | |
| 649 | # trans was autobegin. they have to call rollback |
| 650 | with expect_raises(tsa.exc.PendingRollbackError): |
| 651 | conn.execute(select(1)) |
| 652 | |
| 653 | # ok |
| 654 | conn.rollback() |
| 655 | |
| 656 | # now we are good |
| 657 | # test reconnects |
| 658 | conn.execute(select(1)) |
| 659 | assert not conn.invalidated |
| 660 | |
| 661 | eq_( |
| 662 | [c.close.mock_calls for c in self.dbapi.connections], |
| 663 | [[call()], []], |
| 664 | ) |
| 665 | |
| 666 | def test_invalidated_close(self): |
| 667 | conn = self.db.connect() |