Method test_cursor_is_closed_on_exhausted

test/orm/test_query.py:5378–5420

test #7274

(self, fetch_method)

    @testing.combinations("fetchmany", "fetchone", "fetchall")
    def test_cursor_is_closed_on_exhausted(self, fetch_method):
        """test #7274"""
        self._eagerload_mappings()

        User = self.classes.User

        sess = fixture_session()

        stmt = select(User).execution_options(yield_per=15)
        result = sess.execute(stmt)

        with mock.patch.object(result.raw, "_soft_close") as mock_close:
            # call assertions are implementation specific.
            # test needs that _soft_close called at least once and without
            # the hard=True flag
            if fetch_method == "fetchmany":
                while True:
                    buf = result.fetchmany(2)
                    if not buf:
                        break
                eq_(mock_close.mock_calls, [mock.call()])
            elif fetch_method == "fetchall":
                eq_(len(result.all()), 4)
                eq_(
                    mock_close.mock_calls, [mock.call(), mock.call(hard=False)]
                )
            elif fetch_method == "fetchone":
                while True:
                    row = result.fetchone()
                    if row is None:
                        break
                eq_(
                    mock_close.mock_calls, [mock.call(), mock.call(hard=False)]
                )
            else:
                assert False

        # soft closed, we can still get an empty result
        eq_(result.all(), [])

        # real closed
        result.close()
        assert_raises(sa.exc.ResourceClosedError, result.all)

    def test_yield_per_close_on_interrupted_iteration_legacy(self):
        """test #8710"""