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_index_usage_via_query(self):
"""
Test that a query will warn if the indexes that exist do not satisfy the
query selector.
"""
index = Index(self.db, 'ddoc001', 'index001', fields=['name'])
index.create()
self.populate_db_with_documents(100)
result = self.db.get_query_result(fields=['name', 'age'],
selector={'age': {'$eq': 6}}, raw_result=True)
self.assertTrue(str(result['warning']).lower().startswith("no matching index found"))
def test_retrieve_index_url(self):
"""
Test constructing the Index url
"""
index = Index(self.db)
self.assertEqual(
index.index_url,
'/'.join((self.db.database_url, '_index'))
)
def test_create_an_index_using_design_prefix(self):
"""
Test that a JSON index is created correctly in the remote database when
the ddoc id is already prefixed by '_design/'
"""
index = Index(self.db, '_design/ddoc001', 'index001', fields=['name', 'age'])
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.assertIsInstance(ddoc.get_view(index.name), QueryIndexView)
self.assertEquals(ddoc['_id'], index.design_document_id)
self.assertTrue(ddoc['_rev'].startswith('1-'))
self.assertEquals(ddoc['indexes'], {})
self.assertEquals(ddoc['language'], 'query')
self.assertEquals(ddoc['lists'], {})
self.assertEquals(ddoc['shows'], {})
self.assertListEqual(list(ddoc['views'].keys()), [index.name])
def test_create_fails_due_to_index_name_validation(self):
"""
Ensure that if the index name is not a string the create call fails.
"""
index = Index(self.db, 'ddoc001', ['index001'], fields=['name', 'age'])
with self.assertRaises(CloudantArgumentError) as cm:
index.create()
err = cm.exception
self.assertEqual(
str(err),
'The index name: [\'index001\'] is not a string.'
)
def test_constructor_with_args(self):
"""
Test instantiating an Index by passing in arguments. As a side effect
this test also tests the design_document_id, name, type, and definition
property methods.
"""
index = Index(self.db, 'ddoc-id', 'index-name', foo={'bar': 'baz'})
self.assertIsInstance(index, Index)
self.assertEqual(index.design_document_id, 'ddoc-id')
self.assertEqual(index.name, 'index-name')
self.assertEqual(index.type, 'json')
self.assertEqual(index.definition, {'foo': {'bar': 'baz'}})
def test_create_an_index_using_ddoc_index_name(self):
"""
Test that a JSON index is created in the remote database.
"""
index = Index(self.db, 'ddoc001', 'index001', fields=['name', 'age'])
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.assertIsInstance(ddoc.get_view(index.name), QueryIndexView)
self.assertEquals(ddoc['_id'], index.design_document_id)
self.assertTrue(ddoc['_rev'].startswith('1-'))
self.assertEquals(ddoc['indexes'], {})
self.assertEquals(ddoc['language'], 'query')
self.assertEquals(ddoc['lists'], {})
self.assertEquals(ddoc['shows'], {})
self.assertListEqual(list(ddoc['views'].keys()), ['index001'])
def test_deleting_index_without_index_name(self):
"""
Tests that deleting an index without an index name provided fails as
expected.
"""
ddoc = DesignDocument(self.db, '_design/ddoc001')
index = Index(self.db, 'ddoc001', fields=['name', 'age'])
self.assertFalse(ddoc.exists())
with self.assertRaises(CloudantArgumentError) as cm:
index.delete()
err = cm.exception
self.assertEqual(
str(err),
'Deleting an index requires an index name be provided.'
)
def all_docs(self, **kwargs):
"""
Return an `Index` object referencing all documents in the database.
You can treat it like an iterator:
for doc in db.all_docs():
print doc
"""
return Index(self._make_url('_all_docs'), session=self._session, **kwargs)
def index(self, path, **kwargs):
"""
Create a `Index` object referencing the function at `path`. For example:
index = doc.index('_view/index-name')
# refers to /DB/_design/DOC/_view/index-name
"""
opts = dict(self.opts, **kwargs)
return Index(self._make_url(path), session=self._session, **opts)