#------------------------------------------------------------------------------ # Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved. #------------------------------------------------------------------------------ """Module for testing Simple Oracle Document Access (SODA) Collections""" import TestEnv import cx_Oracle class TestCase(TestEnv.BaseTestCase): def __testSkip(self, coll, numToSkip, expectedContent): filterSpec = {'$orderby': [{'path': 'name', 'order': 'desc'}]} doc = coll.find().filter(filterSpec).skip(numToSkip).getOne() content = doc.getContent() if doc is not None else None self.assertEqual(content, expectedContent) def testInvalidJson(self): "test inserting invalid JSON value into SODA collection" invalidJson = "{testKey:testValue}" sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoInvalidJSON") doc = sodaDatabase.createDocument(invalidJson) self.assertRaises(cx_Oracle.IntegrityError, coll.insertOne, doc) coll.drop() def testInsertDocuments(self): "test inserting documents into a SODA collection" sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoInsertDocs") coll.find().remove() valuesToInsert = [ { "name" : "George", "age" : 47 }, { "name" : "Susan", "age" : 39 }, { "name" : "John", "age" : 50 }, { "name" : "Jill", "age" : 54 } ] insertedKeys = [] for value in valuesToInsert: doc = coll.insertOneAndGet(value) insertedKeys.append(doc.key) self.connection.commit() self.assertEqual(coll.find().count(), len(valuesToInsert)) for key, value in zip(insertedKeys, valuesToInsert): doc = coll.find().key(key).getOne() self.assertEqual(doc.getContent(), value) coll.drop() def testSkipDocuments(self): "test skipping documents in a SODA collection" sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoSkipDocs") coll.find().remove() valuesToInsert = [ { "name" : "Anna", "age" : 62 }, { "name" : "Mark", "age" : 37 }, { "name" : "Martha", "age" : 43 }, { "name" : "Matthew", "age" : 28 } ] for value in valuesToInsert: coll.insertOne(value) self.connection.commit() self.__testSkip(coll, 0, valuesToInsert[3]) self.__testSkip(coll, 1, valuesToInsert[2]) self.__testSkip(coll, 3, valuesToInsert[0]) self.__testSkip(coll, 4, None) self.__testSkip(coll, 125, None) def testReplaceDocument(self): "test replace documents in SODA collection" sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoReplaceDoc") coll.find().remove() content = {'name': 'John', 'address': {'city': 'Sydney'}} doc = coll.insertOneAndGet(content) newContent = {'name': 'John', 'address': {'city':'Melbourne'}} coll.find().key(doc.key).replaceOne(newContent) self.connection.commit() self.assertEqual(coll.find().key(doc.key).getOne().getContent(), newContent) coll.drop() def testSearchDocumentsWithContent(self): "test search documents with content using $like and $regex" sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoSearchDocContent") coll.find().remove() data = [ {'name': 'John', 'address': {'city': 'Bangalore'}}, {'name': 'Johnson', 'address': {'city': 'Banaras'}}, {'name': 'Joseph', 'address': {'city': 'Bangalore'}}, {'name': 'Jibin', 'address': {'city': 'Secunderabad'}}, {'name': 'Andrew', 'address': {'city': 'Hyderabad'}}, {'name': 'Matthew', 'address': {'city': 'Mumbai'}} ] for value in data: coll.insertOne(value) self.connection.commit() filterSpecs = [ ({'name': {'$like': 'And%'}}, 1), ({'name': {'$like': 'J%n'}}, 3), ({'name': {'$like': '%hn%'}}, 2), ({'address.city': {'$like': 'Ban%'}}, 3), ({'address.city': {'$like': '%bad'}}, 2), ({'address.city': {'$like': 'Hyderabad'}}, 1), ({'address.city': {'$like': 'China%'}}, 0), ({'name': {'$regex': 'Jo.*'}}, 3), ({'name': {'$regex': '.*[ho]n'}}, 2), ({'name': {'$regex': 'J.*h'}}, 1), ({'address.city': {'$regex': 'Ba.*'}}, 3), ({'address.city': {'$regex': '.*bad'}}, 2), ({'address.city': {'$regex': 'Hyderabad'}}, 1), ({'name': {'$regex': 'Js.*n'}}, 0) ] for filterSpec, expectedCount in filterSpecs: self.assertEqual(coll.find().filter(filterSpec).count(), expectedCount, filterSpec) coll.drop() def testDocumentRemove(self): "test removing documents" sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoRemoveDocs") coll.find().remove() data = [ {'name': 'John', 'address': {'city': 'Bangalore'}}, {'name': 'Johnson', 'address': {'city': 'Banaras'}}, {'name': 'Joseph', 'address': {'city': 'Mangalore'}}, {'name': 'Jibin', 'address': {'city': 'Secunderabad'}}, {'name': 'Andrew', 'address': {'city': 'Hyderabad'}}, {'name': 'Matthew', 'address': {'city': 'Mumbai'}} ] docs = [coll.insertOneAndGet(v) for v in data] coll.find().key(docs[3].key).remove() self.assertEqual(coll.find().count(), len(data) - 1) searchResults = coll.find().filter({'name': {'$like': 'Jibin'}}) self.assertEqual(searchResults.count(), 0) coll.find().filter({'name': {'$like': 'John%'}}).remove() self.assertEqual(coll.find().count(), len(data) - 3) coll.find().filter({'name': {'$regex': 'J.*'}}).remove() self.assertEqual(coll.find().count(), len(data) - 4) self.connection.commit() coll.drop() def testCreateAndDropIndex(self): "test create and drop Index" indexName = "cxoTestIndexes_ix_1" indexSpec = { 'name': indexName, 'fields': [ { 'path': 'address.city', 'datatype': 'string', 'order': 'asc' } ] } sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoTestIndexes") coll.find().remove() self.connection.commit() coll.dropIndex(indexName) coll.createIndex(indexSpec) self.assertRaises(cx_Oracle.DatabaseError, coll.createIndex, indexSpec) self.assertEqual(coll.dropIndex(indexName), True) self.assertEqual(coll.dropIndex(indexName), False) coll.drop() def testGetDocuments(self): "test getting documents from Collection" self.connection.autocommit = True sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoTestGetDocs") coll.find().remove() data = [ {'name': 'John', 'address': {'city': 'Bangalore'}}, {'name': 'Johnson', 'address': {'city': 'Banaras'}}, {'name': 'Joseph', 'address': {'city': 'Mangalore'}}, {'name': 'Jibin', 'address': {'city': 'Secunderabad'}}, {'name': 'Andrew', 'address': {'city': 'Hyderabad'}} ] insertedKeys = list(sorted(coll.insertOneAndGet(v).key for v in data)) fetchedKeys = list(sorted(d.key for d in coll.find().getDocuments())) self.assertEqual(fetchedKeys, insertedKeys) coll.drop() def testCursor(self): "test fetching documents from a cursor" self.connection.autocommit = True sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoFindViaCursor") coll.find().remove() data = [ {'name': 'John', 'address': {'city': 'Bangalore'}}, {'name': 'Johnson', 'address': {'city': 'Banaras'}}, {'name': 'Joseph', 'address': {'city': 'Mangalore'}}, ] insertedKeys = list(sorted(coll.insertOneAndGet(v).key for v in data)) fetchedKeys = list(sorted(d.key for d in coll.find().getCursor())) self.assertEqual(fetchedKeys, insertedKeys) coll.drop() def testMultipleDocumentRemove(self): "test removing multiple documents using multiple keys" sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoRemoveMultipleDocs") coll.find().remove() data = [ {'name': 'John', 'address': {'city': 'Bangalore'}}, {'name': 'Johnson', 'address': {'city': 'Banaras'}}, {'name': 'Joseph', 'address': {'city': 'Mangalore'}}, {'name': 'Jibin', 'address': {'city': 'Secunderabad'}}, {'name': 'Andrew', 'address': {'city': 'Hyderabad'}}, {'name': 'Matthew', 'address': {'city': 'Mumbai'}} ] docs = [coll.insertOneAndGet(v) for v in data] keys = [docs[i].key for i in (1, 3, 5)] numRemoved = coll.find().keys(keys).remove() self.assertEqual(numRemoved, len(keys)) self.assertEqual(coll.find().count(), len(data) - len(keys)) self.connection.commit() coll.drop() def testDocumentVersion(self): "test using version to get documents and remove them" sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoDocumentVersion") coll.find().remove() content = {'name': 'John', 'address': {'city': 'Bangalore'}} insertedDoc = coll.insertOneAndGet(content) key = insertedDoc.key version = insertedDoc.version doc = coll.find().key(key).version(version).getOne() self.assertEqual(doc.getContent(), content) newContent = {'name': 'James', 'address': {'city': 'Delhi'}} replacedDoc = coll.find().key(key).replaceOneAndGet(newContent) newVersion = replacedDoc.version doc = coll.find().key(key).version(version).getOne() self.assertEqual(doc, None) doc = coll.find().key(key).version(newVersion).getOne() self.assertEqual(doc.getContent(), newContent) self.assertEqual(coll.find().key(key).version(version).remove(), 0) self.assertEqual(coll.find().key(key).version(newVersion).remove(), 1) self.assertEqual(coll.find().count(), 0) self.connection.commit() coll.drop() def testGetCursor(self): "test keys with GetCursor" sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoKeysWithGetCursor") coll.find().remove() data = [ {'name': 'John', 'address': {'city': 'Bangalore'}}, {'name': 'Johnson', 'address': {'city': 'Banaras'}}, {'name': 'Joseph', 'address': {'city': 'Mangalore'}}, {'name': 'Jibin', 'address': {'city': 'Secunderabad'}}, {'name': 'Andrew', 'address': {'city': 'Hyderabad'}}, {'name': 'Matthew', 'address': {'city': 'Mumbai'}} ] docs = [coll.insertOneAndGet(v) for v in data] keys = [docs[i].key for i in (2, 4, 5)] fetchedKeys = [d.key for d in coll.find().keys(keys).getCursor()] self.assertEqual(list(sorted(fetchedKeys)), list(sorted(keys))) self.connection.commit() coll.drop() def testCreatedOn(self): "test createdOn attribute of Document" sodaDatabase = self.connection.getSodaDatabase() coll = sodaDatabase.createCollection("cxoCreatedOn") coll.find().remove() data = {'name': 'John', 'address': {'city': 'Bangalore'}} doc = coll.insertOneAndGet(data) self.assertEqual(doc.createdOn, doc.lastModified) if __name__ == "__main__": TestEnv.RunTestCases()