Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
# This is not the preferred way of dealing with query index
# views but it works best for this test.
data = {
'_id': '_design/ddoc001',
'language': 'query',
'views': {
'view001': {'map': {'fields': {'name': 'asc', 'age': 'asc'}},
'reduce': '_count',
'options': {'def': {'fields': ['name', 'age']},
'w': 2}
}
}
}
self.db.create_document(data)
ddoc = DesignDocument(self.db, '_design/ddoc001')
ddoc.fetch()
with self.assertRaises(CloudantDesignDocumentException) as cm:
ddoc.update_view(
'view001',
'function (doc) {\n emit(doc._id, 1);\n}'
)
err = cm.exception
self.assertEqual(
str(err),
'Cannot update a query index view using this method.'
)
def test_document_iteration_returns_valid_documents(self):
"""
This test will check that the __iter__ method returns documents that are
valid Document or DesignDocument objects and that they can be managed
remotely. In this test we will delete the documents as part of the test
to ensure that remote management is working as expected and confirming
that the documents are valid.
"""
self.populate_db_with_documents(3)
with DesignDocument(self.db, '_design/ddoc001') as ddoc:
ddoc.add_view('view001', 'function (doc) {\n emit(doc._id, 1);\n}')
docs = []
ddocs = []
for doc in self.db:
# A valid document must have a document_url
self.assertEqual(
doc.document_url,
'/'.join((self.db.database_url, doc['_id']))
)
if isinstance(doc, DesignDocument):
self.assertEqual(doc['_id'], '_design/ddoc001')
ddocs.append(doc)
elif isinstance(doc, Document):
self.assertTrue(
doc['_id'] in ['julia000', 'julia001', 'julia002']
)
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_view_callable_view_result(self):
"""
Test that by referencing the .result attribute the view callable
method is invoked and the data returned is wrapped as a Result.
"""
self.populate_db_with_documents()
ddoc = DesignDocument(self.db, 'ddoc001')
ddoc.add_view(
'view001',
'function (doc) {\n emit(doc._id, 1);\n}'
)
ddoc.save()
view = ddoc.get_view('view001')
rslt = view.result
self.assertIsInstance(rslt, Result)
ids = []
# rslt[:3] limits the Result to the first 3 elements
for row in rslt[:3]:
ids.append(row['id'])
expected = ['julia000', 'julia001', 'julia002']
self.assertTrue(all(x in ids for x in expected))
def test_invalid_limit(self):
"""
Test limit translation fails when a non-integer value is used.
"""
msg = 'Argument limit not instance of expected type:'
with self.assertRaises(CloudantArgumentError) as cm:
python_to_couch({'limit': True})
self.assertTrue(str(cm.exception).startswith(msg))
def test_constructor_with_query_skip_limit_options_skip_limit(self):
"""
Ensure that options skip and/or limit override the values in the query
callable if present when constructing a QueryResult
"""
query = Query(self.db, skip=10, limit=10)
result = QueryResult(query, skip=100, limit=100)
self.assertIsInstance(result, QueryResult)
self.assertDictEqual(result.options, {'skip': 100, 'limit': 100})
self.assertEqual(result._ref, query)
def test_map_setter_failure(self):
"""
Test that the map setter fails if a dict is not supplied
"""
try:
self.view.map = 'function (doc) {\n emit(doc._id, 1);\n}'
self.fail('Above statement should raise an Exception')
except CloudantArgumentError as err:
self.assertEqual(
str(err),
'The map property must be a dictionary.'
)
def test_constructor_with_query_skip_limit(self):
"""
Test instantiating a QueryResult when query callable already has
skip and/or limit
"""
query = Query(self.db, skip=10, limit=10)
result = QueryResult(query)
self.assertIsInstance(result, QueryResult)
self.assertDictEqual(result.options, {'skip': 10, 'limit': 10})
self.assertEqual(result._ref, query)
def test_map_setter(self):
"""
Test that the map setter works
"""
ddoc = DesignDocument(self.db, 'ddoc001')
view = View(ddoc, 'view001')
self.assertIsNone(view.get('map'))
view.map = 'function (doc) {\n emit(doc._id, 1);\n}'
self.assertEqual(
view.get('map'),
'function (doc) {\n emit(doc._id, 1);\n}'
)
def test_constructor(self):
"""
Test instantiating a View
"""
ddoc = DesignDocument(self.db, 'ddoc001')
view = View(
ddoc,
'view001',
'function (doc) {\n emit(doc._id, 1);\n}',
'_count',
dbcopy='{0}-copy'.format(self.db.database_name)
)
self.assertEqual(view.design_doc, ddoc)
self.assertEqual(view.view_name, 'view001')
self.assertIsInstance(view['map'], _Code)
self.assertEqual(
view['map'],
'function (doc) {\n emit(doc._id, 1);\n}'
)
self.assertIsInstance(view['reduce'], _Code)
self.assertEqual(view['reduce'], '_count')
self.assertEqual(