(self, close_resets_only, method)
| 1680 | @testing.combinations(True, False, "default", argnames="close_resets_only") |
| 1681 | @testing.variation("method", ["close", "reset"]) |
| 1682 | def test_session_close_resets_only(self, close_resets_only, method): |
| 1683 | users, User = self.tables.users, self.classes.User |
| 1684 | self.mapper_registry.map_imperatively(User, users) |
| 1685 | |
| 1686 | if close_resets_only is True: |
| 1687 | kw = {"close_resets_only": True} |
| 1688 | elif close_resets_only is False: |
| 1689 | kw = {"close_resets_only": False} |
| 1690 | else: |
| 1691 | eq_(close_resets_only, "default") |
| 1692 | kw = {} |
| 1693 | |
| 1694 | s = fixture_session(**kw) |
| 1695 | u1 = User() |
| 1696 | s.add(u1) |
| 1697 | assertions.in_(u1, s) |
| 1698 | |
| 1699 | if method.reset: |
| 1700 | s.reset() |
| 1701 | elif method.close: |
| 1702 | s.close() |
| 1703 | else: |
| 1704 | method.fail() |
| 1705 | |
| 1706 | assertions.not_in(u1, s) |
| 1707 | |
| 1708 | u2 = User() |
| 1709 | if method.close and close_resets_only is False: |
| 1710 | with expect_raises_message( |
| 1711 | exc.InvalidRequestError, |
| 1712 | "This Session has been permanently closed and is unable " |
| 1713 | "to handle any more transaction requests.", |
| 1714 | ): |
| 1715 | s.add(u2) |
| 1716 | assertions.not_in(u2, s) |
| 1717 | else: |
| 1718 | s.add(u2) |
| 1719 | assertions.in_(u2, s) |
| 1720 | |
| 1721 | |
| 1722 | class SessionSchemaTranslateTest( |
nothing calls this directly
no test coverage detected