rendered paste body# -*- test-case-name: couchdb.test.test_couchdb -*-# -*- coding: utf-8 -*-## Copyright (C) 2007-2008 Christopher Lenz# Copyright (C) 2008 Federico Stella# All rights reserved.## Redistribution and use in source and binary forms, with or without# modification, are permitted provided that the following conditions# are met:## 1. Redistributions of source code must retain the above copyright# notice, this list of conditions and the following disclaimer.# 2. Redistributions in binary form must reproduce the above copyright# notice, this list of conditions and the following disclaimer in# the documentation and/or other materials provided with the# distribution.# 3. The name of the author may not be used to endorse or promote# products derived from this software without specific prior# written permission.## THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE# ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE# GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER# IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR# OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN# IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.import osfrom twisted.internet import deferfrom twisted.trial import unittestfrom twisted.trial.unittest import tempfileimport twisted.internetimport couchdbtwisted.internet.base.DelayedCall.debug = Truefrom pprint import pprintclass CouchdbServerTestCase(unittest.TestCase): db_list = [] dbname = "twisted-couchdb-test-run" def setUp(self): self.uri = os.environ.get('COUCHDB_URI', 'http://localhost:5984') self.server = couchdb.Server(self.uri) def cb(retval): self.failUnless(retval[u'ok'] == True, "Create Db aborted") self.db = self.server.get(self.dbname) return self.server.create(self.dbname).addCallbacks(cb, lambda _: True) def tearDown(self): def cb(retval): self.failUnless(retval[u'ok'] == True, "Cannot delete db") return self.server.delete(self.dbname).addCallback(cb) def _create_test_docs(self, num): return defer.DeferredList([self.db.set(str(i), {'a': i + 1, 'b': (i + 1) ** 2}) for i in xrange(num)]) def test_list_dbs(self): def update_list(dbs): self.failUnless(isinstance(dbs, list)) self.db_list = dbs return self.server.list().addCallback(update_list) def test_server_len(self): return self.server.len().addCallback(lambda n: self.failUnless(n > 0)) def test_create_duplicate_db(self): return self.failUnlessFailure(self.server.create(self.dbname), couchdb.ResourceConflict) def test_get_db(self): self.assertEquals(type(self.server.get(self.dbname)), couchdb.Database) def test_dbname(self): self.failUnlessRaises(ValueError, self.server.create, "Foo") self.failUnlessRaises(ValueError, self.server.create, "Foo;") def test_db_len_empty(self): return self.db.len().addCallback(self.assertEqual, 0) def test_db_len(self): return self._create_test_docs(4).addCallback(lambda _: self.db.len()).addCallback(self.assertEqual, 4) def test_db_set(self): def cb(retval): self.assertEqual('0', retval['_id']) self.failUnless('_rev' in retval) data = {'a': 1, 'b': 1} return self.db.set('0', data).addCallback(cb) def test_db_doc_non_existent_default(self): return self.db.get("foo").addCallback(lambda r: self.failUnless(r == None)) def test_db_doc_non_existent(self): return self.db.get("foo", False).addCallback(lambda r: self.failUnless(r == False)) def test_db_doc_get(self): data = {'a': 1, 'b': 1} def cb(doc): self.assertEqual('0', doc.id) self.assertEqual(data['_rev'], doc.rev) return self.db.len().addCallback(self.assertEqual, 1) return self.db.set('0', data).addCallback( lambda _: self.db.get('0') ).addCallback(cb) def test_db_doc_delete(self): data = {'a': 1, 'b': 1} return self.db.set('0', data).addCallback( lambda _: self.db.delete('0') ).addCallback( lambda _: self.db.get('0') ).addCallback(lambda r: self.failUnless(r == None)) def test_db_doc_create(self): data = {'a': 1, 'b': 1} def cb(retval): return self.db.get(retval).addCallback(lambda r: self.assertEqual(retval, r.id)) return self.db.create(data).addCallback(cb) def test_db_not_containb(self): return self.db.contains('foo').addCallback(self.failIf) def test_db_contains(self): data = {'a': 1, 'b': 1} return self.db.set('0', data).addCallback( lambda _: self.db.contains('0') ).addCallback(self.failUnless) def test_db_update(self): data = [ couchdb.Document(type='Person', name='John Doe'), couchdb.Document(type='Person', name='Mary Jane'), couchdb.Document(type='City', name='Gotham City') ] def cb(retval): self.failUnless(len([d for d in retval]) == len(data)) return self.db.update(data).addCallback(cb) def test_db_view(self): def cb(retval): for doc in retval: self.failUnlessIn(int(doc), xrange(4)) return self._create_test_docs(4).addCallback(lambda _: self.db.iter()).addCallback(cb) def test_db_query(self): query = """function(doc) { if (doc.a==4) map(null, doc.b); }""" def cb(retval): result = list(retval) self.failUnlessEqual(1, len(result)) self.failUnlessEqual('3', result[0].id) self.failUnlessEqual(16, result[0].value) return self._create_test_docs(4).addCallback(lambda _: self.db.query(query)).addCallback(cb) def test_db_query_and_redo(self): query = """function(doc) { if (doc.a==4) map(null, doc.b); }""" def update(doc): doc['a'] = 4 return self.db.set('0', doc) def cb(retval): result = list(retval) self.failUnlessEqual(2, len(result)) return self._create_test_docs(4).addCallback( lambda _: self.db.get('0') ).addCallback( update ).addCallback( lambda _: self.db.query(query) ).addCallback(cb) def test_db_query_add_and_redo(self): query = """function(doc) { if (doc.a==4) map(null, doc.b); }""" def update(doc): doc['a'] = 4 return defer.DeferredList([self.db.set('0', doc), self.db.create({'a': 3, 'b': 9}), self.db.create({'a': 4, 'b': 16})]) def cb(retval): result = list(retval) self.failUnlessEqual(3, len(result)) return self.db.len().addCallback(self.failUnlessEqual, 6) return self._create_test_docs(4).addCallback( lambda _: self.db.get('0') ).addCallback( update ).addCallback( lambda _: self.db.query(query) ).addCallback(cb) def test_db_query_delete_and_redo(self): query = """function(doc) { if (doc.a==4) map(null, doc.b); }""" def update(): return defer.DeferredList([self.db.create({'a': 3, 'b': 9}), self.db.create({'a': 4, 'b': 16})]) def cb(retval): result = list(retval) self.failUnlessEqual(2, len(result)) return self.db.len().addCallback(self.failUnlessEqual, 5) return self._create_test_docs(4).addCallback( lambda _: self.db.delete('0') ).addCallback( lambda _: update() ).addCallback( lambda _: self.db.query(query) ).addCallback(cb) def test_db_conflict_detection(self): doc1 = {'a': 1, 'b': 1} data = {'_id': 'bar', 'a': 3, 'b': 1} def eb(failure): failure.trap(defer.FirstError, couchdb.ServerError) failure.value.subFailure.trap(couchdb.ServerError) return failure.value.subFailure def update(doc): self.failUnlessEqual(doc1['_id'], doc.id) self.failUnlessEqual(doc1['_rev'], doc.rev) # make conflicting modifications doc['a'] = 2 doc1['a'] = 3 dl = defer.DeferredList( [self.db.set('bar', doc), self.db.set('bar', doc1)], fireOnOneErrback=True, consumeErrors=True ) return self.failUnlessFailure(dl.addErrback(eb), couchdb.ServerError) def update_without_revision(): # try submitting without the revision info return self.failUnlessFailure(self.db.set('bar', data), couchdb.ServerError) return self.db.set('bar', doc1).addCallback( lambda _: self.db.get('bar') ).addCallback( update ).addCallback( lambda _: update_without_revision() ).addCallback( lambda _: self.db.delete('bar') ).addCallback( lambda _:self.db.set('bar', data) ) def test_db_update(self): num = os.environ.get('COUCHDB_DOC_X_TEST', 100) docs = [{'integer': i, 'string': str(i)} for i in xrange(num)] def cb(retval): size = len([d for d in retval]) return self.db.len().addCallback(self.failUnlessEqual, size) return self.db.update(docs).addCallback(cb) def test_db_lot_of_docs(self): num = os.environ.get('COUCHDB_DOC_X_TEST', 100) query = """function(doc) { map(doc.integer, null); }""" dl = defer.DeferredList([self.db.set(str(i), {'integer': i, 'string': str(i)}) for i in xrange(num)]) def check_ascending(retval): results = list(retval) self.failUnlessEqual(num, len(results)) for idx, row in enumerate(results): self.failUnlessEqual(idx, row.key) def check_descending(retval): results = list(retval) self.failUnlessEqual(num, len(results)) for idx, row in enumerate(results): self.failUnlessEqual(num - idx - 1, row.key) return dl.addCallbacks(lambda _: self.db.len(), pprint).addCallback( self.failUnlessEqual, num ).addCallback( lambda _: self.db.query(query) ).addCallback( check_ascending ).addCallback( lambda _: self.db.query(query, descending=True) ).addCallback( check_descending ) def test_db_multiple_rows(self): query = """function(doc){ for (var i = 0; i < doc.cities.length; i++) { map(doc.cities[i] + ", " + doc._id, null); } }""" dl = [ self.db.set('NC', {'cities': ["Charlotte", "Raleigh"]}), self.db.set('MA', {'cities': ["Boston", "Lowell", "Worcester", "Cambridge", "Springfield"]}), self.db.set('FL', {'cities': ["Miami", "Tampa", "Orlando", "Springfield"]}) ] def cb1(retval): results = list(retval) self.failUnlessEqual(11, len(results)) self.failUnlessEqual("Boston, MA", results[0].key); self.failUnlessEqual("Cambridge, MA", results[1].key); self.failUnlessEqual("Charlotte, NC", results[2].key); self.failUnlessEqual("Lowell, MA", results[3].key); self.failUnlessEqual("Miami, FL", results[4].key); self.failUnlessEqual("Orlando, FL", results[5].key); self.failUnlessEqual("Raleigh, NC", results[6].key); self.failUnlessEqual("Springfield, FL", results[7].key); self.failUnlessEqual("Springfield, MA", results[8].key); self.failUnlessEqual("Tampa, FL", results[9].key); self.failUnlessEqual("Worcester, MA", results[10].key); def cb2(doc): doc['cities'].append("Wilmington") return self.db.set('NC', doc) def cb3(retval): results = list(retval) self.failUnlessEqual(12, len(results)) self.failUnlessEqual("Wilmington, NC", results[10].key) def cb4(retval): results = list(retval) self.failUnlessEqual(7, len(results)) self.failUnlessEqual("Charlotte, NC", results[0].key); self.failUnlessEqual("Miami, FL", results[1].key); self.failUnlessEqual("Orlando, FL", results[2].key); self.failUnlessEqual("Raleigh, NC", results[3].key); self.failUnlessEqual("Springfield, FL", results[4].key); self.failUnlessEqual("Tampa, FL", results[5].key); self.failUnlessEqual("Wilmington, NC", results[6].key) return defer.DeferredList(dl).addCallback( lambda _: self.db.query(query) ).addCallback( cb1 ).addCallback( lambda _: self.db.get('NC') ).addCallback( cb2 ).addCallback( lambda _: self.db.query(query) ).addCallback( cb3 ).addCallback( lambda _: self.db.delete('MA') ).addCallback( lambda _: self.db.query(query) ).addCallback( cb4 ) def test_large_docs(self): num = 10 longtext = "0123456789\n" * 10 #XXX* (2 ** 10) # 10 * 100Kb docs = [{'longtext': longtext} for i in xrange(num)] query = """function(doc) { map(null, doc.longtext); }""" return self.db.update(docs).addCallback( lambda _:self.db.query(query) ).addCallback( lambda r: self.failUnlessEqual(len(list(r)), num) ) def test_utf8_encoding(self): texts = [ u"1. Ascii: hello", u"2. Russian: На берегу пустынных волн", u"3. Math: ∮ E⋅da = Q, n → ∞, ∑ f(i) = ∏ g(i),", u"4. Geek: STARGΛ̊TE SG-1", u"5. Braille: ⡌⠁⠧⠑ ⠼⠁⠒ ⡍⠜⠇⠑⠹⠰⠎ ⡣⠕⠌" ] query = """function(doc) { map(doc.text, null); }""" def cb(retval): for idx, row in enumerate(retval): self.failUnlessEqual(texts[idx], row.key) dl = [self.db.set(str(idx), {'text': text}) for idx, text in enumerate(texts)] return defer.DeferredList(dl).addCallback( lambda _: self.db.query(query) ).addCallback(cb) def test_db_design_doc(self): v = { 'views': { 'all_docs': 'function(doc) { map(doc.integer, null) }', 'no_docs': 'function(doc) {}', 'single_doc': 'function(doc) { if (doc._id == "1") map(null, 1) }' } } dl = [self.db.set(str(i), {'integer': i, 'string': str(i)}) for i in xrange(50)] dl.append(self.db.set('_design/test', v)) def cb1(retval): for idx, row in enumerate(retval): self.failUnlessEqual(idx, row.key) return defer.DeferredList(dl).addCallback( lambda _: self.db.view('_view/test/all_docs') ).addCallback( cb1 ).addCallback( lambda _: self.db.view('_view/test/no_docs') ).addCallback( lambda r: self.failUnlessEqual(0, len(list(r))) ).addCallback( lambda _: self.db.view('_view/test/single_doc') ).addCallback( lambda r: self.failUnlessEqual(1, len(list(r))) ) def test_db_collation(self): values = [ None, False, True, 1, 2, 3.0, 4, 'a', 'A', 'aa', 'b', 'B', 'ba', 'bb', ['a'], ['b'], ['b', 'c'], ['b', 'c', 'a'], ['b', 'd'], ['b', 'd', 'e'], {'a': 1}, {'a': 2}, {'b': 1}, {'b': 2}, {'b': 2, 'c': 2}, ] dl = [self.db.set(str(idx + 1), {'foo': value}) for idx, value in enumerate(values)] dl.insert(0, self.db.set('0', {'bar': 0})) query = """function(doc) { map(doc.foo, null); }""" def cb1(retval): rows = iter(retval) self.failUnlessEqual(None, rows.next().value) for idx, row in enumerate(rows): self.failUnlessEqual(values[idx + 1], row.key) def cb2(retval): for idx, row in enumerate(retval): if idx < len(values): self.failUnlessEqual(values[len(values) - 1 - idx], row.key) else: self.failUnlessEqual(None, row.value) def cb3(): dl = [self.db.query(query, key=value).addCallback(cb4, value) for value in values] return defer.DeferredList(dl) def cb4(retval, value): rows = list(retval) self.failUnlessEqual(1, len(rows)) self.failUnlessEqual(value, rows[0].key) return defer.DeferredList(dl).addCallback( lambda _: self.db.query(query) ).addCallback( cb1 ).addCallback( lambda _: self.db.query(query, descending=True) ).addCallback( cb2 ).addCallback( lambda _: cb3() )