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_delete_design_document_success_with_encoded_url(self):
"""
Test that we can remove a design document from the remote
database successfully when the document id requires an encoded url.
"""
ddoc = DesignDocument(self.db, '_design/http://example.com')
ddoc.create()
self.assertTrue(ddoc.exists())
ddoc.delete()
self.assertFalse(ddoc.exists())
self.assertEqual(ddoc, {'_id': '_design/http://example.com'})
def test_get_search_disk_size(self):
"""
Test retrieval of search_disk_size endpoint from the DesignDocument.
"""
self.populate_db_with_documents(100)
ddoc = DesignDocument(self.db, '_design/ddoc001')
ddoc.add_search_index(
'search001',
'function (doc) {\n index("default", doc._id); '
'if (doc._id) {index("name", doc.name, {"store": true}); }\n}'
)
ddoc.save()
ddoc_remote = DesignDocument(self.db, '_design/ddoc001')
ddoc_remote.fetch()
# Make a request to the search index to ensure it is built
self.db.get_search_result('_design/ddoc001', 'search001', query='name:julia*')
search_disk_size = ddoc_remote.search_disk_size('search001')
self.assertEqual(
sorted(search_disk_size.keys()), ['name', 'search_index'],
'The search disk size should contain only keys "name" and "search_index"')
self.assertEqual(
search_disk_size['name'], '_design/ddoc001/search001',
'The search index "name" should be correct.')
self.assertEqual(
sorted(search_disk_size['search_index'].keys()), ['disk_size'],
'The search index should contain only key "disk_size"')
def test_add_a_show_function(self):
"""
Test that adding a show function adds a show object to
the DesignDocument dictionary.
"""
ddoc = DesignDocument(self.db, '_design/ddoc001')
self.assertEqual(ddoc.get('shows'), {})
ddoc.add_show_function(
'show001',
'function(head, req) { provides(\'html\', function() '
'{var html = \'<ol>\\n\'; while (row = getRow()) '
'{ html += \'<li>\' + row.key + \':\' + row.value + \'</li>\\n\';} '
'html += \'</ol>\'; return html; }); }'
)
self.assertListEqual(list(ddoc.get('shows').keys()), ['show001'])
self.assertEqual(
ddoc.get('shows'),
{'show001': 'function(head, req) { provides(\'html\', function() '
'{var html = \'<ol>\\n\'; while (row = getRow()) '</ol>
def test_create_a_search_index_no_kwargs(self):
"""
Test that a TEXT index is created in the remote database.
"""
index = TextIndex(self.db, 'ddoc001', 'index001')
index.create()
self.assertEqual(index.design_document_id, '_design/ddoc001')
self.assertEqual(index.name, 'index001')
with DesignDocument(self.db, index.design_document_id) as ddoc:
self.assertEquals(ddoc['_id'], index.design_document_id)
self.assertTrue(ddoc['_rev'].startswith('1-'))
self.assertEquals(ddoc['language'], 'query')
self.assertEquals(ddoc['lists'], {})
self.assertEquals(ddoc['shows'], {})
self.assertEquals(ddoc['views'], {})
index = ddoc['indexes']['index001']
self.assertEquals(index['analyzer']['default'], 'keyword')
self.assertEquals(index['analyzer']['fields']['$default'], 'standard')
self.assertEquals(index['analyzer']['name'], 'perfield')
self.assertEquals(index['index']['default_analyzer'], 'keyword')
self.assertEquals(index['index']['default_field'], {})
self.assertEquals(index['index']['fields'], 'all_fields')
self.assertEquals(index['index']['selector'], {})
def test_retrieve_view_url(self):
"""
Test the retrieval of the View url
"""
ddoc = DesignDocument(self.db, 'ddoc001')
view = View(ddoc, 'view001')
self.assertEqual(
view.url,
'/'.join((ddoc.document_url, '_view/view001'))
)
def test_fetch_search_index(self):
"""
Ensure that the document fetch from the database returns the
DesignDocument format as expected when retrieving a design document
containing search indexes.
"""
ddoc = DesignDocument(self.db, '_design/ddoc001')
search_index = ('function (doc) {\n index("default", doc._id); '
'if (doc._id) {index("name", doc.name, '
'{"store": true}); }\n} ')
ddoc.add_search_index('search001', search_index)
ddoc.add_search_index('search002', search_index, 'simple')
ddoc.add_search_index('search003', search_index, 'standard')
ddoc.save()
ddoc_remote = DesignDocument(self.db, '_design/ddoc001')
self.assertNotEqual(ddoc_remote, ddoc)
ddoc_remote.fetch()
self.assertEqual(ddoc_remote, ddoc)
self.assertTrue(ddoc_remote['_rev'].startswith('1-'))
self.assertEqual(ddoc_remote, {
'_id': '_design/ddoc001',
'_rev': ddoc['_rev'],
'options': {'partitioned': False},
'indexes': {
'search001': {'index': search_index},
'search002': {'index': search_index, 'analyzer': 'simple'},
'search003': {'index': search_index, 'analyzer': 'standard'}
},
'views': {},
'lists': {},
'shows': {}
def test_geospatial_index(self):
"""
Test retrieval and query of Cloudant Geo indexes from the DesignDocument.
"""
ddoc = DesignDocument(self.db, '_design/ddoc001')
ddoc['st_indexes'] = {
'geoidx': {
'index': 'function(doc) { '
'if (doc.geometry && doc.geometry.coordinates) { '
'st_index(doc.geometry);}} '
}
}
ddoc.save()
ddoc_remote = DesignDocument(self.db, '_design/ddoc001')
self.assertNotEqual(ddoc_remote, ddoc)
ddoc_remote.fetch()
self.assertEqual(ddoc_remote, {
'_id': '_design/ddoc001',
'_rev': ddoc['_rev'],
'st_indexes': ddoc['st_indexes'],
'indexes': {},
def test_search_index_save_fails_with_existing_text_index(self):
"""
Tests that save fails when language is query and both a search index
and a text index exist in the design document.
"""
ddoc = DesignDocument(self.db, '_design/ddoc001')
ddoc['language'] = 'query'
ddoc['indexes']['index001'] = {
'index': {'index_array_lengths': True,
'fields': [{'name': 'name', 'type': 'string'},
{'name': 'age', 'type': 'number'}],
'default_field': {'enabled': True, 'analyzer': 'german'},
'default_analyzer': 'keyword',
'selector': {}},
'analyzer': {'name': 'perfield','default': 'keyword',
'fields': {'$default': 'german'}}}
ddoc.save()
self.assertTrue(ddoc['_rev'].startswith('1-'))
search_index = ('function (doc) {\n index("default", doc._id); '
'if (doc._id) {index("name", doc.name, '
'{"store": true}); }\n}')
ddoc.add_search_index('search001', search_index)