Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_feed_using_limit(self):
"""
Test getting content back for a feed using limit
"""
self.populate_db_with_documents()
feed = Feed(self.db, limit=3)
seq_list = list()
for change in feed:
self.assertSetEqual(set(change.keys()), set(['seq', 'changes', 'id']))
seq_list.append(change['seq'])
self.assertEqual(len(seq_list), 3)
self.assertTrue(str(seq_list[0]).startswith('1'))
self.assertTrue(str(seq_list[1]).startswith('2'))
self.assertTrue(str(seq_list[2]).startswith('3'))
self.assertEqual(feed.last_seq, seq_list[2])
def test_get_raw_content(self):
"""
Test getting raw feed content
"""
self.populate_db_with_documents(3)
feed = Feed(self.db, raw_data=True)
raw_content = list()
for raw_line in feed:
self.assertIsInstance(raw_line, BYTETYPE)
raw_content.append(raw_line)
changes = json.loads(''.join([unicode_(x) for x in raw_content]))
if self.is_couchdb_1x_version() is True:
self.assertSetEqual(
set(changes.keys()), set(['results', 'last_seq']))
else:
self.assertSetEqual(set(changes.keys()), set(['results', 'last_seq', 'pending']))
results = list()
for result in changes['results']:
self.assertSetEqual(set(result.keys()), set(['seq', 'changes', 'id']))
results.append(result)
expected = set(['julia000', 'julia001', 'julia002'])
self.assertSetEqual(set([x['id'] for x in results]), expected)
def test_invalid_non_positive_integer_argument(self):
"""
Test that an invalid integer argument type is caught and an exception is
raised
"""
feed = Feed(self.client, limit=-1)
with self.assertRaises(CloudantArgumentError) as cm:
invalid_feed = [x for x in feed]
self.assertEqual(
str(cm.exception), 'Argument limit must be > 0. Found: -1')
def test_invalid_non_positive_integer_argument(self):
"""
Test that an invalid integer argument type is caught and an exception is
raised
"""
feed = Feed(self.db, limit=-1)
with self.assertRaises(CloudantArgumentError) as cm:
invalid_feed = [x for x in feed]
self.assertEqual(
str(cm.exception), 'Argument limit must be > 0. Found: -1')
def test_invalid_non_positive_integer_argument(self):
"""
Test that an invalid integer argument type is caught and an exception is
raised
"""
feed = Feed(self.client, timeout=-1)
with self.assertRaises(CloudantArgumentError) as cm:
invalid_feed = [x for x in feed]
self.assertEqual(
str(cm.exception), 'Argument timeout must be > 0. Found: -1')
def test_invalid_feed_value(self):
"""
Test that an invalid feed argument value is caught and an exception is
raised
"""
feed = Feed(self.client, feed='foo')
with self.assertRaises(CloudantArgumentError) as cm:
invalid_feed = [x for x in feed]
self.assertTrue(str(cm.exception).startswith(
'Invalid value (foo) for feed option.'))
feed = Feed(self.client, feed='normal')
with self.assertRaises(CloudantArgumentError) as cm:
invalid_feed = [x for x in feed]
self.assertTrue(str(cm.exception).startswith(
'Invalid value (normal) for feed option.'))
def test_get_feed_using_timeout(self):
"""
Test getting content back for a feed using timeout
"""
self.populate_db_with_documents()
feed = Feed(self.db, feed='continuous', timeout=100)
changes = list()
for change in feed:
self.assertSetEqual(set(change.keys()), set(['seq', 'changes', 'id']))
changes.append(change)
expected = set(['julia{0:03d}'.format(i) for i in range(100)])
self.assertSetEqual(set([x['id'] for x in changes]), expected)
self.assertTrue(str(feed.last_seq).startswith('100'))
# Compare continuous with normal
normal = Feed(self.db)
self.assertSetEqual(
set([x['id'] for x in changes]), set([n['id'] for n in normal]))
def test_get_continuous_feed(self):
"""
Test getting content back for a "continuous" feed
"""
self.populate_db_with_documents()
feed = Feed(self.db, feed='continuous')
changes = list()
for change in feed:
self.assertSetEqual(set(change.keys()), set(['seq', 'changes', 'id']))
changes.append(change)
if len(changes) == 100:
feed.stop()
expected = set(['julia{0:03d}'.format(i) for i in range(100)])
self.assertSetEqual(set([x['id'] for x in changes]), expected)
self.assertIsNone(feed.last_seq)
# Compare continuous with normal
normal = Feed(self.db)
self.assertSetEqual(
set([x['id'] for x in changes]), set([n['id'] for n in normal]))
def test_invalid_argument(self):
"""
Test that an invalid argument is caught and an exception is raised
"""
feed = Feed(self.client, foo='bar')
with self.assertRaises(CloudantArgumentError) as cm:
invalid_feed = [x for x in feed]
self.assertEqual(str(cm.exception), 'Invalid argument foo')
feed = Feed(self.client, style='all_docs')
with self.assertRaises(CloudantArgumentError) as cm:
invalid_feed = [x for x in feed]
self.assertEqual(str(cm.exception), 'Invalid argument style')
def test_get_feed_using_filter(self):
"""
Test getting content back for a feed using filter
"""
self.populate_db_with_documents(6)
ddoc = DesignDocument(self.db, '_design/ddoc001')
ddoc['filters'] = {
'even_docs': 'function(doc, req){if (doc.age % 2 != 0){return false;} return true;}'
}
ddoc.create()
feed = Feed(self.db, filter='ddoc001/even_docs')
changes = list()
for change in feed:
self.assertSetEqual(set(change.keys()), set(['seq', 'changes', 'id']))
changes.append(change)
expected = set(['julia000', 'julia002', 'julia004'])
self.assertSetEqual(set([x['id'] for x in changes]), expected)
self.assertTrue(str(feed.last_seq).startswith('7'))