test_queryset.py 文件源码

def test_paged_result_handling(self):
        if PROTOCOL_VERSION < 2:
            raise unittest.SkipTest(
                "Paging requires native protocol 2+, "
                "currently using: {0}".format(PROTOCOL_VERSION)
            )

        # addresses #225
        class PagingTest(Model):
            id = columns.Integer(primary_key=True)
            val = columns.Integer()
        sync_table(self.conn, PagingTest)

        PagingTest.create(self.conn, id=1, val=1)
        PagingTest.create(self.conn, id=2, val=2)

        with mock.patch.object(self.conn.session, 'default_fetch_size', 1):
            results = PagingTest.objects().find_all(self.conn)

        assert len(results) == 2
        drop_table(self.conn, PagingTest)