def setUp(self):
self.api = API(create_auth())
- self.api.retry_count = 2
- self.api.retry_delay = 5
def testidcursoritems(self):
items = list(Cursor(self.api.user_timeline).items(25))
self.assertEqual(len(pages), 5)
def testcursorcursoritems(self):
- items = list(Cursor(self.api.friends_ids).items())
- self.assert_(len(items) > 0)
+ items = list(Cursor(self.api.friends_ids).items(10))
+ self.assertEqual(len(items), 10)
- items = list(Cursor(self.api.followers_ids, 'twitter').items(30))
- self.assert_(len(items) == 30)
+ items = list(Cursor(self.api.followers_ids, 'twitter').items(10))
+ self.assertEqual(len(items), 10)
def testcursorcursorpages(self):
- pages = list(Cursor(self.api.friends_ids).pages())
- self.assert_(len(pages) > 0)
+ pages = list(Cursor(self.api.friends_ids).pages(1))
+ self.assert_(len(pages) == 1)
- pages = list(Cursor(self.api.followers_ids, 'twitter').pages(5))
- self.assert_(len(pages) == 5)
+ pages = list(Cursor(self.api.followers_ids, 'twitter').pages(1))
+ self.assert_(len(pages) == 1)
BaseIterator.__init__(self, method, args, kargs)
self.max_id = kargs.get('max_id')
self.since_id = kargs.get('since_id')
+ self.count = 0
def next(self):
"""Fetch a set of items with IDs less than current set."""
+ if self.limit and self.limit == self.count:
+ raise StopIteration
+
# max_id is inclusive so decrement by one
# to avoid requesting duplicate items.
max_id = self.since_id - 1 if self.max_id else None
raise StopIteration
self.max_id = data.max_id
self.since_id = data.since_id
+ self.count += 1
return data
def prev(self):
"""Fetch a set of items with IDs greater than current set."""
+ if self.limit and self.limit == self.count:
+ raise StopIteration
+
since_id = self.max_id
data = self.method(since_id = since_id, *self.args, **self.kargs)
if len(data) == 0:
raise StopIteration
self.max_id = data.max_id
self.since_id = data.since_id
+ self.count += 1
return data
class PageIterator(BaseIterator):