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_by_parent_and_type(self):
couchConnector = CouchDbConnector(self.db)
doc = {
'_id': '123',
'type': 'father',
'parent': None,
}
couchConnector.saveDocument(doc)
doc = {
'_id': '456',
'type': 'child',
'parent': '123',
}
couchConnector.saveDocument(doc)
doc = {
'_id': '789',
def test_save_Document(self):
couchConnector = CouchDbConnector(self.db)
doc = {
'_id': '123',
'data': 'some data'
}
couchConnector.saveDocument(doc)
doc_from_db = self.db.get('123')
self.assertNotEquals(
doc_from_db,
None,
"Document should be retrieved")
self.assertEquals(
doc_from_db.get('data'),
'some data',
def test_get_Document(self):
couchConnector = CouchDbConnector(self.db)
doc = {
'_id': '123',
'data': 'some data'
}
couchConnector.saveDocument(doc)
doc_retrieved = couchConnector.getDocument('123')
self.assertNotEquals(
doc_retrieved,
None,
"Document should be retrieved")
self.assertEquals(
doc_retrieved.get('data'),
'some data',
def test_remove_Document(self):
couchConnector = CouchDbConnector(self.db)
doc = {
'_id': '123',
'data': 'some data'
}
couchConnector.saveDocument(doc)
couchConnector.remove('123')
try:
doc_from_db = self.db.get('123')
except ResourceNotFound:
doc_from_db = None
self.assertEquals(
doc_from_db,
None,
def __init__(self, db, seq_num=0):
super(CouchDbConnector, self).__init__(type=DBTYPE.COUCHDB)
self.db = db
self.saves_counter = 0
self.mutex = threading.Lock()
self._docs = {}
try:
vmanager = ViewsManager()
vmanager.addViews(self.db)
self._compactDatabase()
except restkit.Unauthorized:
getLogger(self).warn(
"You're not authorized to upload views to this database")
self.seq_num = self.db.info()['update_seq']
def _loadDb(self, dbname):
db = self.__serv.get_db(dbname)
seq = db.info()['update_seq']
self.dbs[dbname] = CouchDbConnector(db, seq_num=seq)
return self.dbs[dbname]
def _create(self, name):
db = self.__serv.create_db(name.lower())
return CouchDbConnector(db)