

def test_cursor_close_w_failed_rowproc(self):
    User = self.classes.User
    s = Session()
    q = s.query(User)
    ctx = q._compile_context()
    cursor = mock.Mock()
    q._entities = [mock.Mock(row_processor=mock.Mock(side_effect=Exception('boom')))]
    assert_raises(Exception, list, loading.instances(q, cursor, ctx))
    assert cursor.close.called, "Cursor wasn't closed"

dataset/ETHPy150Open zzzeek/sqlalchemy/test/orm/test_loading.py InstancesTest.test_cursor_close_w_failed_rowproc/original