Viewing file: test_paginate.py (6.52 KB) -rw-r--r-- Select action/file-type: (+) | (+) | (+) | Code (+) | Session (+) | (+) | SDB (+) | (+) | (+) | (+) | (+) | (+) |
# Copyright 2006 Divmod, Inc. See LICENSE file for details
""" This module contains tests for the L{axiom.store.ItemQuery.paginate} method. """
from twisted.trial.unittest import TestCase
from axiom.store import Store from axiom.item import Item from axiom.attributes import integer, compoundIndex
from axiom.test.util import QueryCounter
class SingleColumnSortHelper(Item): mainColumn = integer(indexed=True) other = integer() compoundIndex(mainColumn, other)
class MultiColumnSortHelper(Item): columnOne = integer() columnTwo = integer() compoundIndex(columnOne, columnTwo)
class CrossTransactionIteration(TestCase):
def test_separateTransactions(self): """ Verify that 'paginate' is iterable in separate transactions. """ s = Store() b1 = SingleColumnSortHelper(store=s, mainColumn=1) b2 = SingleColumnSortHelper(store=s, mainColumn=2) b3 = SingleColumnSortHelper(store=s, mainColumn=3) itr = s.transact(lambda : iter(s.query(SingleColumnSortHelper).paginate())) self.assertIdentical(s.transact(itr.next), b1) self.assertEquals(s.transact(lambda : (itr.next(), itr.next())), (b2, b3)) self.assertRaises(StopIteration, lambda : s.transact(itr.next))
def test_moreItemsNotMoreWork(self): """ Verify that each step of a paginate does not become more work as items are added. """ s = Store() self._checkEfficiency(s.query(SingleColumnSortHelper))
def test_moreItemsNotMoreWorkSorted(self): """ Verify that each step of a paginate does not become more work as more items are added even if a sort is given. """ s = Store() self._checkEfficiency(s.query(SingleColumnSortHelper, sort=SingleColumnSortHelper.mainColumn.ascending))
def test_moreItemsNotMoreWorkRestricted(self): s = Store() self._checkEfficiency(s.query(SingleColumnSortHelper, SingleColumnSortHelper.other == 6, sort=SingleColumnSortHelper.mainColumn.ascending))
def _checkEfficiency(self, qry): s = qry.store mnum = [0] def more(): mnum[0] += 1 SingleColumnSortHelper(store=s, mainColumn=mnum[0], other=6) for i in range(5): more()
qc = QueryCounter(s) # Sanity check: calling paginate() shouldn't do _any_ DB work. L = [] m = qc.measure( # Let's also keep the page-size to 1, forcing the implementation to # get exactly 1 item each time. (Otherwise the first N items will # take a fixed amount of work, the next 10, and so on, but each # subsequent item will take 0, breaking our attempt to measure # below) lambda : L.append(qry.paginate(pagesize=1))) self.assertEquals(m, 0) y = L.pop() g = iter(y) # startup costs a little more, so ignore that # s.debug = True what = qc.measure(g.next) # 1 oneunit = qc.measure(g.next) # 2 otherunit = qc.measure(g.next) self.assertEquals(otherunit, oneunit) # 3 # Now, make some more data
for i in range(3): more() # and make sure that doesn't increase the amount of work self.assertEquals(qc.measure(g.next), oneunit) # 4 self.assertEquals(qc.measure(g.next), oneunit) # 5 self.assertEquals(qc.measure(g.next), oneunit) # 6
# one more sanity check - we're at the end. self.assertEquals(g.next().mainColumn, 7) self.assertEquals(g.next().mainColumn, 8) self.assertEquals(list(g), [])
def test_storeIDTiebreaker(self): """ Verify that items whose sort column are identical are all returned and deterministically ordered. """ s = Store() x = [SingleColumnSortHelper(store=s, mainColumn=1234) for nothing in range(10)] first = SingleColumnSortHelper(store=s, mainColumn=1233) last = SingleColumnSortHelper(store=s, mainColumn=1235) # This is sensitive to page size, so let's test it at lots of places # where edge-cases are likely to develop in the implementation. for pagesize in range(1, 30) + [1000]: # The ordering here in the asserts might look a little weird - that we # ascend by storeID in both cases regardless of the order of the sort, # but it's intentional. The storeID is merely to be a tiebreaker to # provide a stable sort. You could be sorting by any number of # compound columns, 'ascending' for your particular column might mean # something odd or contradictory to 'ascending' for storeID's # 'ascending'. If you want guaranteed stability on storeID, do that. self.assertEqual( list(s.query( SingleColumnSortHelper, sort=SingleColumnSortHelper.mainColumn.descending ).paginate(pagesize=pagesize)), [last] + x + [first])
self.assertEqual( list(s.query( SingleColumnSortHelper, sort=SingleColumnSortHelper.mainColumn.ascending ).paginate(pagesize=pagesize)), [first] + x + [last])
def test_moreThanOneColumnSort(self): """ Verify that paginate works with queries that have complex sort expressions.
Note: it doesn't. """ s = Store()
x = MultiColumnSortHelper(store=s, columnOne=1, columnTwo=9) y1 = MultiColumnSortHelper(store=s, columnOne=2, columnTwo=1) y2 = MultiColumnSortHelper(store=s, columnOne=2, columnTwo=2) y3 = MultiColumnSortHelper(store=s, columnOne=2, columnTwo=3) y4 = MultiColumnSortHelper(store=s, columnOne=2, columnTwo=4) z = MultiColumnSortHelper(store=s, columnOne=3, columnTwo=5) self.assertEquals(list( s.query(MultiColumnSortHelper, sort=[MultiColumnSortHelper.columnOne.ascending, MultiColumnSortHelper.columnTwo.ascending] ).paginate(pagesize=1)), [x, y1, y2, y3, y4, z])
test_moreThanOneColumnSort.todo = ( "There's no use-case for this yet, but it would be a consistent " "extension of the API.")
|